107 'I': 'Affaire', |
107 'I': 'Affaire', |
108 'J': 'Affaire', |
108 'J': 'Affaire', |
109 'X': 'Affaire', |
109 'X': 'Affaire', |
110 'ET': 'CWEType', 'ETN': 'String'}]) |
110 'ET': 'CWEType', 'ETN': 'String'}]) |
111 rql, solutions = partrqls[1] |
111 rql, solutions = partrqls[1] |
112 self.assertEquals(rql, 'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, ' |
112 self.assertEquals(rql, 'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, X is IN(BaseTransition, Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWUser, Card, Comment, Division, Email, EmailAddress, EmailPart, EmailThread, ExternalUri, File, Folder, Image, Note, Personne, RQLExpression, Societe, State, SubDivision, SubWorkflowExitPoint, Tag, TrInfo, Transition, Workflow, WorkflowTransition)') |
113 'X is IN(Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWUser, Card, Comment, Division, Email, EmailAddress, EmailPart, EmailThread, ExternalUri, File, Folder, Image, Note, Personne, RQLExpression, Societe, State, SubDivision, Tag, TrInfo, Transition)') |
|
114 self.assertListEquals(sorted(solutions), |
113 self.assertListEquals(sorted(solutions), |
115 sorted([{'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'}, |
114 sorted([{'X': 'BaseTransition', 'ETN': 'String', 'ET': 'CWEType'}, |
|
115 {'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'}, |
116 {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'}, |
116 {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'}, |
117 {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'}, |
117 {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'}, |
118 {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'}, |
118 {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'}, |
119 {'X': 'CWCache', 'ETN': 'String', 'ET': 'CWEType'}, |
119 {'X': 'CWCache', 'ETN': 'String', 'ET': 'CWEType'}, |
120 {'X': 'CWConstraint', 'ETN': 'String', 'ET': 'CWEType'}, |
120 {'X': 'CWConstraint', 'ETN': 'String', 'ET': 'CWEType'}, |
139 {'X': 'Personne', 'ETN': 'String', 'ET': 'CWEType'}, |
139 {'X': 'Personne', 'ETN': 'String', 'ET': 'CWEType'}, |
140 {'X': 'RQLExpression', 'ETN': 'String', 'ET': 'CWEType'}, |
140 {'X': 'RQLExpression', 'ETN': 'String', 'ET': 'CWEType'}, |
141 {'X': 'Societe', 'ETN': 'String', 'ET': 'CWEType'}, |
141 {'X': 'Societe', 'ETN': 'String', 'ET': 'CWEType'}, |
142 {'X': 'State', 'ETN': 'String', 'ET': 'CWEType'}, |
142 {'X': 'State', 'ETN': 'String', 'ET': 'CWEType'}, |
143 {'X': 'SubDivision', 'ETN': 'String', 'ET': 'CWEType'}, |
143 {'X': 'SubDivision', 'ETN': 'String', 'ET': 'CWEType'}, |
|
144 {'X': 'SubWorkflowExitPoint', 'ETN': 'String', 'ET': 'CWEType'}, |
144 {'X': 'Tag', 'ETN': 'String', 'ET': 'CWEType'}, |
145 {'X': 'Tag', 'ETN': 'String', 'ET': 'CWEType'}, |
145 {'X': 'Transition', 'ETN': 'String', 'ET': 'CWEType'}, |
146 {'X': 'Transition', 'ETN': 'String', 'ET': 'CWEType'}, |
146 {'X': 'TrInfo', 'ETN': 'String', 'ET': 'CWEType'}])) |
147 {'X': 'TrInfo', 'ETN': 'String', 'ET': 'CWEType'}, |
|
148 {'X': 'Workflow', 'ETN': 'String', 'ET': 'CWEType'}, |
|
149 {'X': 'WorkflowTransition', 'ETN': 'String', 'ET': 'CWEType'}])) |
147 rql, solutions = partrqls[2] |
150 rql, solutions = partrqls[2] |
148 self.assertEquals(rql, |
151 self.assertEquals(rql, |
149 'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), ' |
152 'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), ' |
150 'ET is CWEType, X is Basket') |
153 'ET is CWEType, X is Basket') |
151 self.assertEquals(solutions, [{'ET': 'CWEType', |
154 self.assertEquals(solutions, [{'ET': 'CWEType', |
283 self.assert_(('Personne',) in rset.description) |
286 self.assert_(('Personne',) in rset.description) |
284 rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")') |
287 rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")') |
285 self.assert_(('Personne',) in rset.description) |
288 self.assert_(('Personne',) in rset.description) |
286 |
289 |
287 def test_select_not_attr(self): |
290 def test_select_not_attr(self): |
288 self.execute("INSERT Personne X: X nom 'bidule'") |
291 peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
289 self.execute("INSERT Societe X: X nom 'chouette'") |
292 seid = self.execute("INSERT Societe X: X nom 'chouette'")[0][0] |
290 rset = self.execute('Personne X WHERE NOT X nom "bidule"') |
293 rset = self.execute('Personne X WHERE NOT X nom "bidule"') |
291 self.assertEquals(len(rset.rows), 0, rset.rows) |
294 self.assertEquals(len(rset.rows), 0, rset.rows) |
292 rset = self.execute('Personne X WHERE NOT X nom "bid"') |
295 rset = self.execute('Personne X WHERE NOT X nom "bid"') |
293 self.assertEquals(len(rset.rows), 1, rset.rows) |
296 self.assertEquals(len(rset.rows), 1, rset.rows) |
294 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
297 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
348 rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?', |
351 rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?', |
349 {'x':peid1}, 'x') |
352 {'x':peid1}, 'x') |
350 self.assertEquals(rset.rows, [[peid1]]) |
353 self.assertEquals(rset.rows, [[peid1]]) |
351 |
354 |
352 def test_select_left_outer_join(self): |
355 def test_select_left_outer_join(self): |
353 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G " |
356 rset = self.execute('DISTINCT Any G WHERE U? in_group G') |
354 "WHERE G name 'users'")[0][0] |
357 self.assertEquals(len(rset), 4) |
355 self.commit() |
358 rset = self.execute('DISTINCT Any G WHERE U? in_group G, U eid %(x)s', |
356 try: |
359 {'x': self.session.user.eid}, 'x') |
357 rset = self.execute('Any FS,TS,C,D,U ORDERBY D DESC ' |
360 self.assertEquals(len(rset), 4) |
358 'WHERE WF wf_info_for X,' |
|
359 'WF from_state FS?, WF to_state TS, WF comment C,' |
|
360 'WF creation_date D, WF owned_by U, X eid %(x)s', |
|
361 {'x': ueid}, 'x') |
|
362 self.assertEquals(len(rset), 1) |
|
363 self.execute('SET X in_state S WHERE X eid %(x)s, S name "deactivated"', |
|
364 {'x': ueid}, 'x') |
|
365 rset = self.execute('Any FS,TS,C,D,U ORDERBY D DESC ' |
|
366 'WHERE WF wf_info_for X,' |
|
367 'WF from_state FS?, WF to_state TS, WF comment C,' |
|
368 'WF creation_date D, WF owned_by U, X eid %(x)s', |
|
369 {'x': ueid}, 'x') |
|
370 self.assertEquals(len(rset), 2) |
|
371 finally: |
|
372 self.execute('DELETE CWUser X WHERE X eid %s' % ueid) |
|
373 self.commit() |
|
374 |
361 |
375 def test_select_ambigous_outer_join(self): |
362 def test_select_ambigous_outer_join(self): |
376 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
363 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
377 self.execute("INSERT Tag X: X name 'tagbis'")[0][0] |
364 self.execute("INSERT Tag X: X name 'tagbis'")[0][0] |
378 geid = self.execute("CWGroup G WHERE G name 'users'")[0][0] |
365 geid = self.execute("CWGroup G WHERE G name 'users'")[0][0] |
464 def test_select_aggregat_having(self): |
451 def test_select_aggregat_having(self): |
465 rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N ' |
452 rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N ' |
466 'WHERE RT name N, RDEF relation_type RT ' |
453 'WHERE RT name N, RDEF relation_type RT ' |
467 'HAVING COUNT(RDEF) > 10') |
454 'HAVING COUNT(RDEF) > 10') |
468 self.assertListEquals(rset.rows, |
455 self.assertListEquals(rset.rows, |
469 [[u'description', 11], |
456 [[u'description_format', 13], |
470 [u'name', 13], [u'created_by', 34], |
457 [u'description', 14], |
471 [u'creation_date', 34], [u'cwuri', 34], |
458 [u'name', 16], |
472 ['in_basket', 34], |
459 [u'created_by', 38], |
473 [u'is', 34], [u'is_instance_of', 34], |
460 [u'creation_date', 38], |
474 [u'modification_date', 34], [u'owned_by', 34]]) |
461 [u'cwuri', 38], |
|
462 [u'in_basket', 38], |
|
463 [u'is', 38], |
|
464 [u'is_instance_of', 38], |
|
465 [u'modification_date', 38], |
|
466 [u'owned_by', 38]]) |
475 |
467 |
476 def test_select_aggregat_having_dumb(self): |
468 def test_select_aggregat_having_dumb(self): |
477 # dumb but should not raise an error |
469 # dumb but should not raise an error |
478 rset = self.execute('Any U,COUNT(X) GROUPBY U ' |
470 rset = self.execute('Any U,COUNT(X) GROUPBY U ' |
479 'WHERE U eid %(x)s, X owned_by U ' |
471 'WHERE U eid %(x)s, X owned_by U ' |
719 ('String', 'String', 'String', 'String',))) |
711 ('String', 'String', 'String', 'String',))) |
720 rset = self.execute('Any X,GN WHERE X is CWUser, G is CWGroup, X login "syt", X in_group G, G name GN') |
712 rset = self.execute('Any X,GN WHERE X is CWUser, G is CWGroup, X login "syt", X in_group G, G name GN') |
721 |
713 |
722 def test_select_union(self): |
714 def test_select_union(self): |
723 rset = self.execute('Any X,N ORDERBY N WITH X,N BEING ' |
715 rset = self.execute('Any X,N ORDERBY N WITH X,N BEING ' |
724 '((Any X,N WHERE X name N, X transition_of E, E name %(name)s)' |
716 '((Any X,N WHERE X name N, X transition_of WF, WF workflow_of E, E name %(name)s)' |
725 ' UNION ' |
717 ' UNION ' |
726 '(Any X,N WHERE X name N, X state_of E, E name %(name)s))', |
718 '(Any X,N WHERE X name N, X state_of WF, WF workflow_of E, E name %(name)s))', |
727 {'name': 'CWUser'}) |
719 {'name': 'CWUser'}) |
728 self.assertEquals([x[1] for x in rset.rows], |
720 self.assertEquals([x[1] for x in rset.rows], |
729 ['activate', 'activated', 'deactivate', 'deactivated']) |
721 ['activate', 'activated', 'deactivate', 'deactivated']) |
730 self.assertEquals(rset.description, |
722 self.assertEquals(rset.description, |
731 [('Transition', 'String'), ('State', 'String'), |
723 [('Transition', 'String'), ('State', 'String'), |
993 self.assertEquals(rset.rows, []) |
985 self.assertEquals(rset.rows, []) |
994 |
986 |
995 # update queries tests #################################################### |
987 # update queries tests #################################################### |
996 |
988 |
997 def test_update_1(self): |
989 def test_update_1(self): |
998 self.execute("INSERT Personne Y: Y nom 'toto'") |
990 peid = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0] |
999 rset = self.execute('Personne X WHERE X nom "toto"') |
991 rset = self.execute('Personne X WHERE X nom "toto"') |
1000 self.assertEqual(len(rset.rows), 1) |
992 self.assertEqual(len(rset.rows), 1) |
1001 self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'") |
993 rset = self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'") |
|
994 self.assertEqual(tuplify(rset.rows), [(peid, 'tutu', 'original')]) |
1002 rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z') |
995 rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z') |
1003 self.assertEqual(tuplify(rset.rows), [('tutu', 'original')]) |
996 self.assertEqual(tuplify(rset.rows), [('tutu', 'original')]) |
1004 |
997 |
1005 def test_update_2(self): |
998 def test_update_2(self): |
1006 self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
999 peid, seid = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")[0] |
1007 #rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto"') |
1000 rset = self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'") |
1008 #self.assertEqual(len(rset.rows), 1) |
1001 self.assertEquals(tuplify(rset.rows), [(peid, seid)]) |
1009 #rset = self.execute('Any X, Y WHERE X travaille Y') |
|
1010 #self.assertEqual(len(rset.rows), 0) |
|
1011 self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'") |
|
1012 rset = self.execute('Any X, Y WHERE X travaille Y') |
1002 rset = self.execute('Any X, Y WHERE X travaille Y') |
1013 self.assertEqual(len(rset.rows), 1) |
1003 self.assertEqual(len(rset.rows), 1) |
1014 |
1004 |
1015 def test_update_2bis(self): |
1005 def test_update_2bis(self): |
1016 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1006 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1025 eid1, eid2 = rset[0][0], rset[0][1] |
1015 eid1, eid2 = rset[0][0], rset[0][1] |
1026 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1016 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1027 {'x': unicode(eid1), 'y': unicode(eid2)}) |
1017 {'x': unicode(eid1), 'y': unicode(eid2)}) |
1028 rset = self.execute('Any X, Y WHERE X travaille Y') |
1018 rset = self.execute('Any X, Y WHERE X travaille Y') |
1029 self.assertEqual(len(rset.rows), 1) |
1019 self.assertEqual(len(rset.rows), 1) |
1030 |
|
1031 ## def test_update_4(self): |
|
1032 ## self.execute("SET X know Y WHERE X ami Y") |
|
1033 |
1020 |
1034 def test_update_multiple1(self): |
1021 def test_update_multiple1(self): |
1035 peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0] |
1022 peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0] |
1036 peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0] |
1023 peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0] |
1037 self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'") |
1024 self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'") |
1128 |
1115 |
1129 def test_nonregr_3(self): |
1116 def test_nonregr_3(self): |
1130 """bad sql generated on the second query (destination_state is not |
1117 """bad sql generated on the second query (destination_state is not |
1131 detected as an inlined relation) |
1118 detected as an inlined relation) |
1132 """ |
1119 """ |
1133 rset = self.execute('Any S,ES,T WHERE S state_of ET, ET name "CWUser",' |
1120 rset = self.execute('Any S,ES,T WHERE S state_of WF, WF workflow_of ET, ET name "CWUser",' |
1134 'ES allowed_transition T, T destination_state S') |
1121 'ES allowed_transition T, T destination_state S') |
1135 self.assertEquals(len(rset.rows), 2) |
1122 self.assertEquals(len(rset.rows), 2) |
1136 |
1123 |
1137 def test_nonregr_4(self): |
1124 def test_nonregr_4(self): |
1138 # fix variables'type, else we get (nb of entity types with a 'name' attribute)**3 |
1125 # fix variables'type, else we get (nb of entity types with a 'name' attribute)**3 |
1258 # huum, psycopg specific |
1245 # huum, psycopg specific |
1259 self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()}) |
1246 self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()}) |
1260 |
1247 |
1261 def test_nonregr_set_query(self): |
1248 def test_nonregr_set_query(self): |
1262 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0] |
1249 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0] |
1263 self.execute("SET E in_group G, E in_state S, " |
1250 self.execute("SET E in_group G, E firstname %(firstname)s, E surname %(surname)s " |
1264 "E firstname %(firstname)s, E surname %(surname)s " |
1251 "WHERE E eid %(x)s, G name 'users'", |
1265 "WHERE E eid %(x)s, G name 'users', S name 'activated'", |
|
1266 {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x') |
1252 {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x') |
1267 |
1253 |
1268 def test_nonregr_u_owned_by_u(self): |
1254 def test_nonregr_u_owned_by_u(self): |
1269 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G " |
1255 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G " |
1270 "WHERE G name 'users'")[0][0] |
1256 "WHERE G name 'users'")[0][0] |