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', |
288 self.assert_(('Personne',) in rset.description) |
291 self.assert_(('Personne',) in rset.description) |
289 rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")') |
292 rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")') |
290 self.assert_(('Personne',) in rset.description) |
293 self.assert_(('Personne',) in rset.description) |
291 |
294 |
292 def test_select_not_attr(self): |
295 def test_select_not_attr(self): |
293 self.execute("INSERT Personne X: X nom 'bidule'") |
296 peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
294 self.execute("INSERT Societe X: X nom 'chouette'") |
297 seid = self.execute("INSERT Societe X: X nom 'chouette'")[0][0] |
295 rset = self.execute('Personne X WHERE NOT X nom "bidule"') |
298 rset = self.execute('Personne X WHERE NOT X nom "bidule"') |
296 self.assertEquals(len(rset.rows), 0, rset.rows) |
299 self.assertEquals(len(rset.rows), 0, rset.rows) |
297 rset = self.execute('Personne X WHERE NOT X nom "bid"') |
300 rset = self.execute('Personne X WHERE NOT X nom "bid"') |
298 self.assertEquals(len(rset.rows), 1, rset.rows) |
301 self.assertEquals(len(rset.rows), 1, rset.rows) |
299 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
302 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
353 rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?', |
356 rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?', |
354 {'x':peid1}, 'x') |
357 {'x':peid1}, 'x') |
355 self.assertEquals(rset.rows, [[peid1]]) |
358 self.assertEquals(rset.rows, [[peid1]]) |
356 |
359 |
357 def test_select_left_outer_join(self): |
360 def test_select_left_outer_join(self): |
358 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G " |
361 rset = self.execute('DISTINCT Any G WHERE U? in_group G') |
359 "WHERE G name 'users'")[0][0] |
362 self.assertEquals(len(rset), 4) |
360 self.commit() |
363 rset = self.execute('DISTINCT Any G WHERE U? in_group G, U eid %(x)s', |
361 try: |
364 {'x': self.session.user.eid}, 'x') |
362 rset = self.execute('Any FS,TS,C,D,U ORDERBY D DESC ' |
365 self.assertEquals(len(rset), 4) |
363 'WHERE WF wf_info_for X,' |
|
364 'WF from_state FS?, WF to_state TS, WF comment C,' |
|
365 'WF creation_date D, WF owned_by U, X eid %(x)s', |
|
366 {'x': ueid}, 'x') |
|
367 self.assertEquals(len(rset), 1) |
|
368 self.execute('SET X in_state S WHERE X eid %(x)s, S name "deactivated"', |
|
369 {'x': ueid}, 'x') |
|
370 rset = self.execute('Any FS,TS,C,D,U ORDERBY D DESC ' |
|
371 'WHERE WF wf_info_for X,' |
|
372 'WF from_state FS?, WF to_state TS, WF comment C,' |
|
373 'WF creation_date D, WF owned_by U, X eid %(x)s', |
|
374 {'x': ueid}, 'x') |
|
375 self.assertEquals(len(rset), 2) |
|
376 finally: |
|
377 self.execute('DELETE CWUser X WHERE X eid %s' % ueid) |
|
378 self.commit() |
|
379 |
366 |
380 def test_select_ambigous_outer_join(self): |
367 def test_select_ambigous_outer_join(self): |
381 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
368 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
382 self.execute("INSERT Tag X: X name 'tagbis'")[0][0] |
369 self.execute("INSERT Tag X: X name 'tagbis'")[0][0] |
383 geid = self.execute("CWGroup G WHERE G name 'users'")[0][0] |
370 geid = self.execute("CWGroup G WHERE G name 'users'")[0][0] |
469 def test_select_aggregat_having(self): |
456 def test_select_aggregat_having(self): |
470 rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N ' |
457 rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N ' |
471 'WHERE RT name N, RDEF relation_type RT ' |
458 'WHERE RT name N, RDEF relation_type RT ' |
472 'HAVING COUNT(RDEF) > 10') |
459 'HAVING COUNT(RDEF) > 10') |
473 self.assertListEquals(rset.rows, |
460 self.assertListEquals(rset.rows, |
474 [[u'description', 11], |
461 [[u'description_format', 13], |
475 [u'name', 13], [u'created_by', 34], |
462 [u'description', 14], |
476 [u'creation_date', 34], [u'cwuri', 34], |
463 [u'name', 16], |
477 ['in_basket', 34], |
464 [u'created_by', 38], |
478 [u'is', 34], [u'is_instance_of', 34], |
465 [u'creation_date', 38], |
479 [u'modification_date', 34], [u'owned_by', 34]]) |
466 [u'cwuri', 38], |
|
467 [u'in_basket', 38], |
|
468 [u'is', 38], |
|
469 [u'is_instance_of', 38], |
|
470 [u'modification_date', 38], |
|
471 [u'owned_by', 38]]) |
480 |
472 |
481 def test_select_aggregat_having_dumb(self): |
473 def test_select_aggregat_having_dumb(self): |
482 # dumb but should not raise an error |
474 # dumb but should not raise an error |
483 rset = self.execute('Any U,COUNT(X) GROUPBY U ' |
475 rset = self.execute('Any U,COUNT(X) GROUPBY U ' |
484 'WHERE U eid %(x)s, X owned_by U ' |
476 'WHERE U eid %(x)s, X owned_by U ' |
724 ('String', 'String', 'String', 'String',))) |
716 ('String', 'String', 'String', 'String',))) |
725 rset = self.execute('Any X,GN WHERE X is CWUser, G is CWGroup, X login "syt", X in_group G, G name GN') |
717 rset = self.execute('Any X,GN WHERE X is CWUser, G is CWGroup, X login "syt", X in_group G, G name GN') |
726 |
718 |
727 def test_select_union(self): |
719 def test_select_union(self): |
728 rset = self.execute('Any X,N ORDERBY N WITH X,N BEING ' |
720 rset = self.execute('Any X,N ORDERBY N WITH X,N BEING ' |
729 '((Any X,N WHERE X name N, X transition_of E, E name %(name)s)' |
721 '((Any X,N WHERE X name N, X transition_of WF, WF workflow_of E, E name %(name)s)' |
730 ' UNION ' |
722 ' UNION ' |
731 '(Any X,N WHERE X name N, X state_of E, E name %(name)s))', |
723 '(Any X,N WHERE X name N, X state_of WF, WF workflow_of E, E name %(name)s))', |
732 {'name': 'CWUser'}) |
724 {'name': 'CWUser'}) |
733 self.assertEquals([x[1] for x in rset.rows], |
725 self.assertEquals([x[1] for x in rset.rows], |
734 ['activate', 'activated', 'deactivate', 'deactivated']) |
726 ['activate', 'activated', 'deactivate', 'deactivated']) |
735 self.assertEquals(rset.description, |
727 self.assertEquals(rset.description, |
736 [('Transition', 'String'), ('State', 'String'), |
728 [('Transition', 'String'), ('State', 'String'), |
998 self.assertEquals(rset.rows, []) |
990 self.assertEquals(rset.rows, []) |
999 |
991 |
1000 # update queries tests #################################################### |
992 # update queries tests #################################################### |
1001 |
993 |
1002 def test_update_1(self): |
994 def test_update_1(self): |
1003 self.execute("INSERT Personne Y: Y nom 'toto'") |
995 peid = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0] |
1004 rset = self.execute('Personne X WHERE X nom "toto"') |
996 rset = self.execute('Personne X WHERE X nom "toto"') |
1005 self.assertEqual(len(rset.rows), 1) |
997 self.assertEqual(len(rset.rows), 1) |
1006 self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'") |
998 rset = self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'") |
|
999 self.assertEqual(tuplify(rset.rows), [(peid, 'tutu', 'original')]) |
1007 rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z') |
1000 rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z') |
1008 self.assertEqual(tuplify(rset.rows), [('tutu', 'original')]) |
1001 self.assertEqual(tuplify(rset.rows), [('tutu', 'original')]) |
1009 |
1002 |
1010 def test_update_2(self): |
1003 def test_update_2(self): |
1011 self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1004 peid, seid = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'")[0] |
1012 #rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto"') |
1005 rset = self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'") |
1013 #self.assertEqual(len(rset.rows), 1) |
1006 self.assertEquals(tuplify(rset.rows), [(peid, seid)]) |
1014 #rset = self.execute('Any X, Y WHERE X travaille Y') |
|
1015 #self.assertEqual(len(rset.rows), 0) |
|
1016 self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'") |
|
1017 rset = self.execute('Any X, Y WHERE X travaille Y') |
1007 rset = self.execute('Any X, Y WHERE X travaille Y') |
1018 self.assertEqual(len(rset.rows), 1) |
1008 self.assertEqual(len(rset.rows), 1) |
1019 |
1009 |
1020 def test_update_2bis(self): |
1010 def test_update_2bis(self): |
1021 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1011 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1030 eid1, eid2 = rset[0][0], rset[0][1] |
1020 eid1, eid2 = rset[0][0], rset[0][1] |
1031 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1021 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1032 {'x': unicode(eid1), 'y': unicode(eid2)}) |
1022 {'x': unicode(eid1), 'y': unicode(eid2)}) |
1033 rset = self.execute('Any X, Y WHERE X travaille Y') |
1023 rset = self.execute('Any X, Y WHERE X travaille Y') |
1034 self.assertEqual(len(rset.rows), 1) |
1024 self.assertEqual(len(rset.rows), 1) |
1035 |
|
1036 ## def test_update_4(self): |
|
1037 ## self.execute("SET X know Y WHERE X ami Y") |
|
1038 |
1025 |
1039 def test_update_multiple1(self): |
1026 def test_update_multiple1(self): |
1040 peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0] |
1027 peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0] |
1041 peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0] |
1028 peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0] |
1042 self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'") |
1029 self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'") |
1133 |
1120 |
1134 def test_nonregr_3(self): |
1121 def test_nonregr_3(self): |
1135 """bad sql generated on the second query (destination_state is not |
1122 """bad sql generated on the second query (destination_state is not |
1136 detected as an inlined relation) |
1123 detected as an inlined relation) |
1137 """ |
1124 """ |
1138 rset = self.execute('Any S,ES,T WHERE S state_of ET, ET name "CWUser",' |
1125 rset = self.execute('Any S,ES,T WHERE S state_of WF, WF workflow_of ET, ET name "CWUser",' |
1139 'ES allowed_transition T, T destination_state S') |
1126 'ES allowed_transition T, T destination_state S') |
1140 self.assertEquals(len(rset.rows), 2) |
1127 self.assertEquals(len(rset.rows), 2) |
1141 |
1128 |
1142 def test_nonregr_4(self): |
1129 def test_nonregr_4(self): |
1143 # fix variables'type, else we get (nb of entity types with a 'name' attribute)**3 |
1130 # fix variables'type, else we get (nb of entity types with a 'name' attribute)**3 |
1263 # huum, psycopg specific |
1250 # huum, psycopg specific |
1264 self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()}) |
1251 self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()}) |
1265 |
1252 |
1266 def test_nonregr_set_query(self): |
1253 def test_nonregr_set_query(self): |
1267 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0] |
1254 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0] |
1268 self.execute("SET E in_group G, E in_state S, " |
1255 self.execute("SET E in_group G, E firstname %(firstname)s, E surname %(surname)s " |
1269 "E firstname %(firstname)s, E surname %(surname)s " |
1256 "WHERE E eid %(x)s, G name 'users'", |
1270 "WHERE E eid %(x)s, G name 'users', S name 'activated'", |
|
1271 {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x') |
1257 {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x') |
1272 |
1258 |
1273 def test_nonregr_u_owned_by_u(self): |
1259 def test_nonregr_u_owned_by_u(self): |
1274 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G " |
1260 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G " |
1275 "WHERE G name 'users'")[0][0] |
1261 "WHERE G name 'users'")[0][0] |