37 from logilab.common.adbh import _GenericAdvFuncHelper |
37 from logilab.common.adbh import _GenericAdvFuncHelper |
38 TYPEMAP = _GenericAdvFuncHelper.TYPE_MAPPING |
38 TYPEMAP = _GenericAdvFuncHelper.TYPE_MAPPING |
39 |
39 |
40 class MakeSchemaTC(TestCase): |
40 class MakeSchemaTC(TestCase): |
41 def test_known_values(self): |
41 def test_known_values(self): |
42 solution = {'A': 'String', 'B': 'EUser'} |
42 solution = {'A': 'String', 'B': 'CWUser'} |
43 self.assertEquals(make_schema((Variable('A'), Variable('B')), solution, |
43 self.assertEquals(make_schema((Variable('A'), Variable('B')), solution, |
44 'table0', TYPEMAP), |
44 'table0', TYPEMAP), |
45 ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'})) |
45 ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'})) |
46 |
46 |
47 |
47 |
48 repo, cnx = init_test_database('sqlite') |
48 repo, cnx = init_test_database('sqlite') |
49 |
49 |
50 |
50 |
51 |
51 |
52 class UtilsTC(BaseQuerierTC): |
52 class UtilsTC(BaseQuerierTC): |
53 repo = repo |
53 repo = repo |
54 |
54 |
55 def get_max_eid(self): |
55 def get_max_eid(self): |
56 # no need for cleanup here |
56 # no need for cleanup here |
57 return None |
57 return None |
58 def cleanup(self): |
58 def cleanup(self): |
59 # no need for cleanup here |
59 # no need for cleanup here |
60 pass |
60 pass |
61 |
61 |
62 def test_preprocess_1(self): |
62 def test_preprocess_1(self): |
63 reid = self.execute('Any X WHERE X is ERType, X name "owned_by"')[0][0] |
63 reid = self.execute('Any X WHERE X is CWRType, X name "owned_by"')[0][0] |
64 rqlst = self._prepare('Any COUNT(RDEF) WHERE RDEF relation_type X, X eid %(x)s', {'x': reid}) |
64 rqlst = self._prepare('Any COUNT(RDEF) WHERE RDEF relation_type X, X eid %(x)s', {'x': reid}) |
65 self.assertEquals(rqlst.solutions, [{'RDEF': 'EFRDef'}, {'RDEF': 'ENFRDef'}]) |
65 self.assertEquals(rqlst.solutions, [{'RDEF': 'CWAttribute'}, {'RDEF': 'CWRelation'}]) |
66 |
66 |
67 def test_preprocess_2(self): |
67 def test_preprocess_2(self): |
68 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
68 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
69 #geid = self.execute("EGroup G WHERE G name 'users'")[0][0] |
69 #geid = self.execute("CWGroup G WHERE G name 'users'")[0][0] |
70 #self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s", |
70 #self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s", |
71 # {'g': geid, 't': teid}, 'g') |
71 # {'g': geid, 't': teid}, 'g') |
72 rqlst = self._prepare('Any X WHERE E eid %(x)s, E tags X', {'x': teid}) |
72 rqlst = self._prepare('Any X WHERE E eid %(x)s, E tags X', {'x': teid}) |
73 # the query may be optimized, should keep only one solution |
73 # the query may be optimized, should keep only one solution |
74 # (any one, etype will be discarded) |
74 # (any one, etype will be discarded) |
75 self.assertEquals(len(rqlst.solutions), 1) |
75 self.assertEquals(len(rqlst.solutions), 1) |
76 |
76 |
77 def test_preprocess_security(self): |
77 def test_preprocess_security(self): |
78 plan = self._prepare_plan('Any ETN,COUNT(X) GROUPBY ETN ' |
78 plan = self._prepare_plan('Any ETN,COUNT(X) GROUPBY ETN ' |
79 'WHERE X is ET, ET name ETN') |
79 'WHERE X is ET, ET name ETN') |
80 plan.session = self._user_session(('users',))[1] |
80 plan.session = self._user_session(('users',))[1] |
81 union = plan.rqlst |
81 union = plan.rqlst |
94 'Any ETN,X WHERE X is ET, ET name ETN, (EXISTS(X owned_by %(B)s))' |
94 'Any ETN,X WHERE X is ET, ET name ETN, (EXISTS(X owned_by %(B)s))' |
95 ' OR ((((EXISTS(D concerne C?, C owned_by %(B)s, X identity D, C is Division, D is Affaire))' |
95 ' OR ((((EXISTS(D concerne C?, C owned_by %(B)s, X identity D, C is Division, D is Affaire))' |
96 ' OR (EXISTS(H concerne G?, G owned_by %(B)s, G is SubDivision, X identity H, H is Affaire)))' |
96 ' OR (EXISTS(H concerne G?, G owned_by %(B)s, G is SubDivision, X identity H, H is Affaire)))' |
97 ' OR (EXISTS(I concerne F?, F owned_by %(B)s, F is Societe, X identity I, I is Affaire)))' |
97 ' OR (EXISTS(I concerne F?, F owned_by %(B)s, F is Societe, X identity I, I is Affaire)))' |
98 ' OR (EXISTS(J concerne E?, E owned_by %(B)s, E is Note, X identity J, J is Affaire)))' |
98 ' OR (EXISTS(J concerne E?, E owned_by %(B)s, E is Note, X identity J, J is Affaire)))' |
99 ', ET is EEType, X is Affaire') |
99 ', ET is CWEType, X is Affaire') |
100 self.assertEquals(solutions, [{'C': 'Division', |
100 self.assertEquals(solutions, [{'C': 'Division', |
101 'D': 'Affaire', |
101 'D': 'Affaire', |
102 'E': 'Note', |
102 'E': 'Note', |
103 'F': 'Societe', |
103 'F': 'Societe', |
104 'G': 'SubDivision', |
104 'G': 'SubDivision', |
105 'H': 'Affaire', |
105 'H': 'Affaire', |
106 'I': 'Affaire', |
106 'I': 'Affaire', |
107 'J': 'Affaire', |
107 'J': 'Affaire', |
108 'X': 'Affaire', |
108 'X': 'Affaire', |
109 'ET': 'EEType', 'ETN': 'String'}]) |
109 'ET': 'CWEType', 'ETN': 'String'}]) |
110 rql, solutions = partrqls[1] |
110 rql, solutions = partrqls[1] |
111 self.assertEquals(rql, 'Any ETN,X WHERE X is ET, ET name ETN, ET is EEType, ' |
111 self.assertEquals(rql, 'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, ' |
112 'X is IN(Bookmark, Card, Comment, Division, ECache, EConstraint, EConstraintType, EEType, EFRDef, EGroup, ENFRDef, EPermission, EProperty, ERType, EUser, Email, EmailAddress, EmailPart, EmailThread, File, Folder, Image, Note, Personne, RQLExpression, Societe, State, SubDivision, Tag, TrInfo, Transition)') |
112 'X is IN(Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWUser, Card, Comment, Division, Email, EmailAddress, EmailPart, EmailThread, File, Folder, Image, Note, Personne, RQLExpression, Societe, State, SubDivision, Tag, TrInfo, Transition)') |
113 self.assertListEquals(sorted(solutions), |
113 self.assertListEquals(sorted(solutions), |
114 sorted([{'X': 'Bookmark', 'ETN': 'String', 'ET': 'EEType'}, |
114 sorted([{'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'}, |
115 {'X': 'Card', 'ETN': 'String', 'ET': 'EEType'}, |
115 {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'}, |
116 {'X': 'Comment', 'ETN': 'String', 'ET': 'EEType'}, |
116 {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'}, |
117 {'X': 'Division', 'ETN': 'String', 'ET': 'EEType'}, |
117 {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'}, |
118 {'X': 'ECache', 'ETN': 'String', 'ET': 'EEType'}, |
118 {'X': 'CWCache', 'ETN': 'String', 'ET': 'CWEType'}, |
119 {'X': 'EConstraint', 'ETN': 'String', 'ET': 'EEType'}, |
119 {'X': 'CWConstraint', 'ETN': 'String', 'ET': 'CWEType'}, |
120 {'X': 'EConstraintType', 'ETN': 'String', 'ET': 'EEType'}, |
120 {'X': 'CWConstraintType', 'ETN': 'String', 'ET': 'CWEType'}, |
121 {'X': 'EEType', 'ETN': 'String', 'ET': 'EEType'}, |
121 {'X': 'CWEType', 'ETN': 'String', 'ET': 'CWEType'}, |
122 {'X': 'EFRDef', 'ETN': 'String', 'ET': 'EEType'}, |
122 {'X': 'CWAttribute', 'ETN': 'String', 'ET': 'CWEType'}, |
123 {'X': 'EGroup', 'ETN': 'String', 'ET': 'EEType'}, |
123 {'X': 'CWGroup', 'ETN': 'String', 'ET': 'CWEType'}, |
124 {'X': 'Email', 'ETN': 'String', 'ET': 'EEType'}, |
124 {'X': 'Email', 'ETN': 'String', 'ET': 'CWEType'}, |
125 {'X': 'EmailAddress', 'ETN': 'String', 'ET': 'EEType'}, |
125 {'X': 'EmailAddress', 'ETN': 'String', 'ET': 'CWEType'}, |
126 {'X': 'EmailPart', 'ETN': 'String', 'ET': 'EEType'}, |
126 {'X': 'EmailPart', 'ETN': 'String', 'ET': 'CWEType'}, |
127 {'X': 'EmailThread', 'ETN': 'String', 'ET': 'EEType'}, |
127 {'X': 'EmailThread', 'ETN': 'String', 'ET': 'CWEType'}, |
128 {'X': 'ENFRDef', 'ETN': 'String', 'ET': 'EEType'}, |
128 {'X': 'CWRelation', 'ETN': 'String', 'ET': 'CWEType'}, |
129 {'X': 'EPermission', 'ETN': 'String', 'ET': 'EEType'}, |
129 {'X': 'CWPermission', 'ETN': 'String', 'ET': 'CWEType'}, |
130 {'X': 'EProperty', 'ETN': 'String', 'ET': 'EEType'}, |
130 {'X': 'CWProperty', 'ETN': 'String', 'ET': 'CWEType'}, |
131 {'X': 'ERType', 'ETN': 'String', 'ET': 'EEType'}, |
131 {'X': 'CWRType', 'ETN': 'String', 'ET': 'CWEType'}, |
132 {'X': 'EUser', 'ETN': 'String', 'ET': 'EEType'}, |
132 {'X': 'CWUser', 'ETN': 'String', 'ET': 'CWEType'}, |
133 {'X': 'File', 'ETN': 'String', 'ET': 'EEType'}, |
133 {'X': 'File', 'ETN': 'String', 'ET': 'CWEType'}, |
134 {'X': 'Folder', 'ETN': 'String', 'ET': 'EEType'}, |
134 {'X': 'Folder', 'ETN': 'String', 'ET': 'CWEType'}, |
135 {'X': 'Image', 'ETN': 'String', 'ET': 'EEType'}, |
135 {'X': 'Image', 'ETN': 'String', 'ET': 'CWEType'}, |
136 {'X': 'Note', 'ETN': 'String', 'ET': 'EEType'}, |
136 {'X': 'Note', 'ETN': 'String', 'ET': 'CWEType'}, |
137 {'X': 'Personne', 'ETN': 'String', 'ET': 'EEType'}, |
137 {'X': 'Personne', 'ETN': 'String', 'ET': 'CWEType'}, |
138 {'X': 'RQLExpression', 'ETN': 'String', 'ET': 'EEType'}, |
138 {'X': 'RQLExpression', 'ETN': 'String', 'ET': 'CWEType'}, |
139 {'X': 'Societe', 'ETN': 'String', 'ET': 'EEType'}, |
139 {'X': 'Societe', 'ETN': 'String', 'ET': 'CWEType'}, |
140 {'X': 'State', 'ETN': 'String', 'ET': 'EEType'}, |
140 {'X': 'State', 'ETN': 'String', 'ET': 'CWEType'}, |
141 {'X': 'SubDivision', 'ETN': 'String', 'ET': 'EEType'}, |
141 {'X': 'SubDivision', 'ETN': 'String', 'ET': 'CWEType'}, |
142 {'X': 'Tag', 'ETN': 'String', 'ET': 'EEType'}, |
142 {'X': 'Tag', 'ETN': 'String', 'ET': 'CWEType'}, |
143 {'X': 'Transition', 'ETN': 'String', 'ET': 'EEType'}, |
143 {'X': 'Transition', 'ETN': 'String', 'ET': 'CWEType'}, |
144 {'X': 'TrInfo', 'ETN': 'String', 'ET': 'EEType'}])) |
144 {'X': 'TrInfo', 'ETN': 'String', 'ET': 'CWEType'}])) |
145 rql, solutions = partrqls[2] |
145 rql, solutions = partrqls[2] |
146 self.assertEquals(rql, |
146 self.assertEquals(rql, |
147 'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), ' |
147 'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), ' |
148 'ET is EEType, X is Basket') |
148 'ET is CWEType, X is Basket') |
149 self.assertEquals(solutions, [{'ET': 'EEType', |
149 self.assertEquals(solutions, [{'ET': 'CWEType', |
150 'X': 'Basket', |
150 'X': 'Basket', |
151 'ETN': 'String', |
151 'ETN': 'String', |
152 }]) |
152 }]) |
153 |
153 |
154 def test_preprocess_security_aggregat(self): |
154 def test_preprocess_security_aggregat(self): |
200 class QuerierTC(BaseQuerierTC): |
200 class QuerierTC(BaseQuerierTC): |
201 repo = repo |
201 repo = repo |
202 |
202 |
203 def test_encoding_pb(self): |
203 def test_encoding_pb(self): |
204 self.assertRaises(RQLSyntaxError, self.execute, |
204 self.assertRaises(RQLSyntaxError, self.execute, |
205 'Any X WHERE X is ERType, X name "öwned_by"') |
205 'Any X WHERE X is CWRType, X name "öwned_by"') |
206 |
206 |
207 def test_unknown_eid(self): |
207 def test_unknown_eid(self): |
208 # should return an empty result set |
208 # should return an empty result set |
209 self.failIf(self.execute('Any X WHERE X eid 99999999')) |
209 self.failIf(self.execute('Any X WHERE X eid 99999999')) |
210 |
210 |
211 # selection queries tests ################################################# |
211 # selection queries tests ################################################# |
212 |
212 |
213 def test_select_1(self): |
213 def test_select_1(self): |
214 rset = self.execute('Any X ORDERBY X WHERE X is EGroup') |
214 rset = self.execute('Any X ORDERBY X WHERE X is CWGroup') |
215 result, descr = rset.rows, rset.description |
215 result, descr = rset.rows, rset.description |
216 self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,)]) |
216 self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,)]) |
217 self.assertEquals(descr, [('EGroup',), ('EGroup',), ('EGroup',), ('EGroup',)]) |
217 self.assertEquals(descr, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)]) |
218 |
218 |
219 def test_select_2(self): |
219 def test_select_2(self): |
220 rset = self.execute('Any X ORDERBY N WHERE X is EGroup, X name N') |
220 rset = self.execute('Any X ORDERBY N WHERE X is CWGroup, X name N') |
221 self.assertEquals(tuplify(rset.rows), [(3,), (1,), (4,), (2,)]) |
221 self.assertEquals(tuplify(rset.rows), [(3,), (1,), (4,), (2,)]) |
222 self.assertEquals(rset.description, [('EGroup',), ('EGroup',), ('EGroup',), ('EGroup',)]) |
222 self.assertEquals(rset.description, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)]) |
223 rset = self.execute('Any X ORDERBY N DESC WHERE X is EGroup, X name N') |
223 rset = self.execute('Any X ORDERBY N DESC WHERE X is CWGroup, X name N') |
224 self.assertEquals(tuplify(rset.rows), [(2,), (4,), (1,), (3,)]) |
224 self.assertEquals(tuplify(rset.rows), [(2,), (4,), (1,), (3,)]) |
225 |
225 |
226 def test_select_3(self): |
226 def test_select_3(self): |
227 rset = self.execute('Any N GROUPBY N WHERE X is EGroup, X name N') |
227 rset = self.execute('Any N GROUPBY N WHERE X is CWGroup, X name N') |
228 result, descr = rset.rows, rset.description |
228 result, descr = rset.rows, rset.description |
229 result.sort() |
229 result.sort() |
230 self.assertEquals(tuplify(result), [('guests',), ('managers',), ('owners',), ('users',)]) |
230 self.assertEquals(tuplify(result), [('guests',), ('managers',), ('owners',), ('users',)]) |
231 self.assertEquals(descr, [('String',), ('String',), ('String',), ('String',)]) |
231 self.assertEquals(descr, [('String',), ('String',), ('String',), ('String',)]) |
232 |
232 |
233 def test_select_is(self): |
233 def test_select_is(self): |
234 rset = self.execute('Any X, TN ORDERBY TN LIMIT 10 WHERE X is T, T name TN') |
234 rset = self.execute('Any X, TN ORDERBY TN LIMIT 10 WHERE X is T, T name TN') |
235 result, descr = rset.rows, rset.description |
235 result, descr = rset.rows, rset.description |
236 self.assertEquals(result[0][1], descr[0][0]) |
236 self.assertEquals(result[0][1], descr[0][0]) |
237 |
237 |
238 def test_select_is_aggr(self): |
238 def test_select_is_aggr(self): |
239 rset = self.execute('Any TN, COUNT(X) GROUPBY TN ORDERBY 2 DESC WHERE X is T, T name TN') |
239 rset = self.execute('Any TN, COUNT(X) GROUPBY TN ORDERBY 2 DESC WHERE X is T, T name TN') |
240 result, descr = rset.rows, rset.description |
240 result, descr = rset.rows, rset.description |
241 self.assertEquals(descr[0][0], 'String') |
241 self.assertEquals(descr[0][0], 'String') |
242 self.assertEquals(descr[0][1], 'Int') |
242 self.assertEquals(descr[0][1], 'Int') |
243 self.assertEquals(result[0][0], 'ENFRDef') |
243 self.assertEquals(result[0][0], 'CWRelation') |
244 |
244 |
245 def test_select_groupby_orderby(self): |
245 def test_select_groupby_orderby(self): |
246 rset = self.execute('Any N GROUPBY N ORDERBY N WHERE X is EGroup, X name N') |
246 rset = self.execute('Any N GROUPBY N ORDERBY N WHERE X is CWGroup, X name N') |
247 self.assertEquals(tuplify(rset.rows), [('guests',), ('managers',), ('owners',), ('users',)]) |
247 self.assertEquals(tuplify(rset.rows), [('guests',), ('managers',), ('owners',), ('users',)]) |
248 self.assertEquals(rset.description, [('String',), ('String',), ('String',), ('String',)]) |
248 self.assertEquals(rset.description, [('String',), ('String',), ('String',), ('String',)]) |
249 |
249 |
250 def test_select_complex_groupby(self): |
250 def test_select_complex_groupby(self): |
251 rset = self.execute('Any N GROUPBY N WHERE X name N') |
251 rset = self.execute('Any N GROUPBY N WHERE X name N') |
252 rset = self.execute('Any N,MAX(D) GROUPBY N LIMIT 5 WHERE X name N, X creation_date D') |
252 rset = self.execute('Any N,MAX(D) GROUPBY N LIMIT 5 WHERE X name N, X creation_date D') |
253 |
253 |
254 def test_select_inlined_groupby(self): |
254 def test_select_inlined_groupby(self): |
255 seid = self.execute('State X WHERE X name "deactivated"')[0][0] |
255 seid = self.execute('State X WHERE X name "deactivated"')[0][0] |
256 rset = self.execute('Any U,L,S GROUPBY U,L,S WHERE X in_state S, U login L, S eid %s' % seid) |
256 rset = self.execute('Any U,L,S GROUPBY U,L,S WHERE X in_state S, U login L, S eid %s' % seid) |
257 |
257 |
258 def test_select_complex_orderby(self): |
258 def test_select_complex_orderby(self): |
259 rset1 = self.execute('Any N ORDERBY N WHERE X name N') |
259 rset1 = self.execute('Any N ORDERBY N WHERE X name N') |
260 self.assertEquals(sorted(rset1.rows), rset1.rows) |
260 self.assertEquals(sorted(rset1.rows), rset1.rows) |
261 rset = self.execute('Any N ORDERBY N LIMIT 5 OFFSET 1 WHERE X name N') |
261 rset = self.execute('Any N ORDERBY N LIMIT 5 OFFSET 1 WHERE X name N') |
262 self.assertEquals(rset.rows[0][0], rset1.rows[1][0]) |
262 self.assertEquals(rset.rows[0][0], rset1.rows[1][0]) |
263 self.assertEquals(len(rset), 5) |
263 self.assertEquals(len(rset), 5) |
264 |
264 |
265 def test_select_5(self): |
265 def test_select_5(self): |
266 rset = self.execute('Any X, TMP ORDERBY TMP WHERE X name TMP, X is EGroup') |
266 rset = self.execute('Any X, TMP ORDERBY TMP WHERE X name TMP, X is CWGroup') |
267 self.assertEquals(tuplify(rset.rows), [(3, 'guests',), (1, 'managers',), (4, 'owners',), (2, 'users',)]) |
267 self.assertEquals(tuplify(rset.rows), [(3, 'guests',), (1, 'managers',), (4, 'owners',), (2, 'users',)]) |
268 self.assertEquals(rset.description, [('EGroup', 'String',), ('EGroup', 'String',), ('EGroup', 'String',), ('EGroup', 'String',)]) |
268 self.assertEquals(rset.description, [('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',)]) |
269 |
269 |
270 def test_select_6(self): |
270 def test_select_6(self): |
271 self.execute("INSERT Personne X: X nom 'bidule'")[0] |
271 self.execute("INSERT Personne X: X nom 'bidule'")[0] |
272 rset = self.execute('Any Y where X name TMP, Y nom in (TMP, "bidule")') |
272 rset = self.execute('Any Y where X name TMP, Y nom in (TMP, "bidule")') |
273 #self.assertEquals(rset.description, [('Personne',), ('Personne',)]) |
273 #self.assertEquals(rset.description, [('Personne',), ('Personne',)]) |
274 self.assert_(('Personne',) in rset.description) |
274 self.assert_(('Personne',) in rset.description) |
275 rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")') |
275 rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")') |
276 self.assert_(('Personne',) in rset.description) |
276 self.assert_(('Personne',) in rset.description) |
277 |
277 |
278 def test_select_not_attr(self): |
278 def test_select_not_attr(self): |
279 self.execute("INSERT Personne X: X nom 'bidule'") |
279 self.execute("INSERT Personne X: X nom 'bidule'") |
280 self.execute("INSERT Societe X: X nom 'chouette'") |
280 self.execute("INSERT Societe X: X nom 'chouette'") |
281 rset = self.execute('Personne X WHERE NOT X nom "bidule"') |
281 rset = self.execute('Personne X WHERE NOT X nom "bidule"') |
282 self.assertEquals(len(rset.rows), 0, rset.rows) |
282 self.assertEquals(len(rset.rows), 0, rset.rows) |
283 rset = self.execute('Personne X WHERE NOT X nom "bid"') |
283 rset = self.execute('Personne X WHERE NOT X nom "bid"') |
284 self.assertEquals(len(rset.rows), 1, rset.rows) |
284 self.assertEquals(len(rset.rows), 1, rset.rows) |
285 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
285 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
286 rset = self.execute('Personne X WHERE NOT X travaille S') |
286 rset = self.execute('Personne X WHERE NOT X travaille S') |
287 self.assertEquals(len(rset.rows), 0, rset.rows) |
287 self.assertEquals(len(rset.rows), 0, rset.rows) |
288 |
288 |
289 def test_select_is_in(self): |
289 def test_select_is_in(self): |
290 self.execute("INSERT Personne X: X nom 'bidule'") |
290 self.execute("INSERT Personne X: X nom 'bidule'") |
291 self.execute("INSERT Societe X: X nom 'chouette'") |
291 self.execute("INSERT Societe X: X nom 'chouette'") |
292 self.assertEquals(len(self.execute("Any X WHERE X is IN (Personne, Societe)")), |
292 self.assertEquals(len(self.execute("Any X WHERE X is IN (Personne, Societe)")), |
293 2) |
293 2) |
294 |
294 |
295 def test_select_not_rel(self): |
295 def test_select_not_rel(self): |
296 self.execute("INSERT Personne X: X nom 'bidule'") |
296 self.execute("INSERT Personne X: X nom 'bidule'") |
297 self.execute("INSERT Societe X: X nom 'chouette'") |
297 self.execute("INSERT Societe X: X nom 'chouette'") |
298 self.execute("INSERT Personne X: X nom 'autre'") |
298 self.execute("INSERT Personne X: X nom 'autre'") |
299 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
299 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
300 rset = self.execute('Personne X WHERE NOT X travaille S') |
300 rset = self.execute('Personne X WHERE NOT X travaille S') |
301 self.assertEquals(len(rset.rows), 1, rset.rows) |
301 self.assertEquals(len(rset.rows), 1, rset.rows) |
302 rset = self.execute('Personne X WHERE NOT X travaille S, S nom "chouette"') |
302 rset = self.execute('Personne X WHERE NOT X travaille S, S nom "chouette"') |
303 self.assertEquals(len(rset.rows), 1, rset.rows) |
303 self.assertEquals(len(rset.rows), 1, rset.rows) |
304 |
304 |
305 def test_select_nonregr_inlined(self): |
305 def test_select_nonregr_inlined(self): |
306 self.execute("INSERT Note X: X para 'bidule'") |
306 self.execute("INSERT Note X: X para 'bidule'") |
307 self.execute("INSERT Personne X: X nom 'chouette'") |
307 self.execute("INSERT Personne X: X nom 'chouette'") |
308 self.execute("INSERT Personne X: X nom 'autre'") |
308 self.execute("INSERT Personne X: X nom 'autre'") |
309 self.execute("SET X ecrit_par P WHERE X para 'bidule', P nom 'chouette'") |
309 self.execute("SET X ecrit_par P WHERE X para 'bidule', P nom 'chouette'") |
310 rset = self.execute('Any U,T ORDERBY T DESC WHERE U is EUser, ' |
310 rset = self.execute('Any U,T ORDERBY T DESC WHERE U is CWUser, ' |
311 'N ecrit_par U, N type T')#, {'x': self.ueid}) |
311 'N ecrit_par U, N type T')#, {'x': self.ueid}) |
312 self.assertEquals(len(rset.rows), 0) |
312 self.assertEquals(len(rset.rows), 0) |
313 |
313 |
314 def test_select_nonregr_edition_not(self): |
314 def test_select_nonregr_edition_not(self): |
315 groupeids = set((1, 2, 3)) |
315 groupeids = set((1, 2, 3)) |
316 groupreadperms = set(r[0] for r in self.execute('Any Y WHERE X name "EGroup", Y eid IN(1, 2, 3), X read_permission Y')) |
316 groupreadperms = set(r[0] for r in self.execute('Any Y WHERE X name "CWGroup", Y eid IN(1, 2, 3), X read_permission Y')) |
317 rset = self.execute('DISTINCT Any Y WHERE X is EEType, X name "EGroup", Y eid IN(1, 2, 3), NOT X read_permission Y') |
317 rset = self.execute('DISTINCT Any Y WHERE X is CWEType, X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y') |
318 self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms)) |
318 self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms)) |
319 rset = self.execute('DISTINCT Any Y WHERE X name "EGroup", Y eid IN(1, 2, 3), NOT X read_permission Y') |
319 rset = self.execute('DISTINCT Any Y WHERE X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y') |
320 self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms)) |
320 self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms)) |
321 |
321 |
322 def test_select_outer_join(self): |
322 def test_select_outer_join(self): |
323 peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
323 peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
324 peid2 = self.execute("INSERT Personne X: X nom 'autre'")[0][0] |
324 peid2 = self.execute("INSERT Personne X: X nom 'autre'")[0][0] |
325 seid1 = self.execute("INSERT Societe X: X nom 'chouette'")[0][0] |
325 seid1 = self.execute("INSERT Societe X: X nom 'chouette'")[0][0] |
326 seid2 = self.execute("INSERT Societe X: X nom 'chouetos'")[0][0] |
326 seid2 = self.execute("INSERT Societe X: X nom 'chouetos'")[0][0] |
357 'WF from_state FS?, WF to_state TS, WF comment C,' |
357 'WF from_state FS?, WF to_state TS, WF comment C,' |
358 'WF creation_date D, WF owned_by U, X eid %(x)s', |
358 'WF creation_date D, WF owned_by U, X eid %(x)s', |
359 {'x': ueid}, 'x') |
359 {'x': ueid}, 'x') |
360 self.assertEquals(len(rset), 2) |
360 self.assertEquals(len(rset), 2) |
361 finally: |
361 finally: |
362 self.execute('DELETE EUser X WHERE X eid %s' % ueid) |
362 self.execute('DELETE CWUser X WHERE X eid %s' % ueid) |
363 self.commit() |
363 self.commit() |
364 |
364 |
365 def test_select_ambigous_outer_join(self): |
365 def test_select_ambigous_outer_join(self): |
366 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
366 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
367 self.execute("INSERT Tag X: X name 'tagbis'")[0][0] |
367 self.execute("INSERT Tag X: X name 'tagbis'")[0][0] |
368 geid = self.execute("EGroup G WHERE G name 'users'")[0][0] |
368 geid = self.execute("CWGroup G WHERE G name 'users'")[0][0] |
369 self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s", |
369 self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s", |
370 {'g': geid, 't': teid}, 'g') |
370 {'g': geid, 't': teid}, 'g') |
371 rset = self.execute("Any GN,TN ORDERBY GN WHERE T? tags G, T name TN, G name GN") |
371 rset = self.execute("Any GN,TN ORDERBY GN WHERE T? tags G, T name TN, G name GN") |
372 self.failUnless(['users', 'tag'] in rset.rows) |
372 self.failUnless(['users', 'tag'] in rset.rows) |
373 self.failUnless(['activated', None] in rset.rows) |
373 self.failUnless(['activated', None] in rset.rows) |
374 rset = self.execute("Any GN,TN ORDERBY GN WHERE T tags G?, T name TN, G name GN") |
374 rset = self.execute("Any GN,TN ORDERBY GN WHERE T tags G?, T name TN, G name GN") |
375 self.assertEquals(rset.rows, [[None, 'tagbis'], ['users', 'tag']]) |
375 self.assertEquals(rset.rows, [[None, 'tagbis'], ['users', 'tag']]) |
376 |
376 |
377 def test_select_not_inline_rel(self): |
377 def test_select_not_inline_rel(self): |
378 self.execute("INSERT Personne X: X nom 'bidule'") |
378 self.execute("INSERT Personne X: X nom 'bidule'") |
379 self.execute("INSERT Note X: X type 'a'") |
379 self.execute("INSERT Note X: X type 'a'") |
380 self.execute("INSERT Note X: X type 'b'") |
380 self.execute("INSERT Note X: X type 'b'") |
381 self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'") |
381 self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'") |
382 rset = self.execute('Note X WHERE NOT X ecrit_par P') |
382 rset = self.execute('Note X WHERE NOT X ecrit_par P') |
383 self.assertEquals(len(rset.rows), 1, rset.rows) |
383 self.assertEquals(len(rset.rows), 1, rset.rows) |
384 |
384 |
385 def test_select_not_unlinked_multiple_solutions(self): |
385 def test_select_not_unlinked_multiple_solutions(self): |
386 self.execute("INSERT Personne X: X nom 'bidule'") |
386 self.execute("INSERT Personne X: X nom 'bidule'") |
387 self.execute("INSERT Note X: X type 'a'") |
387 self.execute("INSERT Note X: X type 'a'") |
388 self.execute("INSERT Note X: X type 'b'") |
388 self.execute("INSERT Note X: X type 'b'") |
389 self.execute("SET Y evaluee X WHERE X type 'a', Y nom 'bidule'") |
389 self.execute("SET Y evaluee X WHERE X type 'a', Y nom 'bidule'") |
421 self.assertEquals(len(rset.rows), 1) |
421 self.assertEquals(len(rset.rows), 1) |
422 self.assertEquals(len(rset.rows[0]), 1) |
422 self.assertEquals(len(rset.rows[0]), 1) |
423 self.assertEquals(rset.description, [('Int',)]) |
423 self.assertEquals(rset.description, [('Int',)]) |
424 |
424 |
425 def test_select_custom_aggregat_concat_string(self): |
425 def test_select_custom_aggregat_concat_string(self): |
426 rset = self.execute('Any CONCAT_STRINGS(N) WHERE X is EGroup, X name N') |
426 rset = self.execute('Any CONCAT_STRINGS(N) WHERE X is CWGroup, X name N') |
427 self.failUnless(rset) |
427 self.failUnless(rset) |
428 self.failUnlessEqual(sorted(rset[0][0].split(', ')), ['guests', 'managers', |
428 self.failUnlessEqual(sorted(rset[0][0].split(', ')), ['guests', 'managers', |
429 'owners', 'users']) |
429 'owners', 'users']) |
430 |
430 |
431 def test_select_custom_regproc_limit_size(self): |
431 def test_select_custom_regproc_limit_size(self): |
432 rset = self.execute('Any TEXT_LIMIT_SIZE(N, 3) WHERE X is EGroup, X name N, X name "managers"') |
432 rset = self.execute('Any TEXT_LIMIT_SIZE(N, 3) WHERE X is CWGroup, X name N, X name "managers"') |
433 self.failUnless(rset) |
433 self.failUnless(rset) |
434 self.failUnlessEqual(rset[0][0], 'man...') |
434 self.failUnlessEqual(rset[0][0], 'man...') |
435 self.execute("INSERT Basket X: X name 'bidule', X description '<b>hop hop</b>', X description_format 'text/html'") |
435 self.execute("INSERT Basket X: X name 'bidule', X description '<b>hop hop</b>', X description_format 'text/html'") |
436 rset = self.execute('Any LIMIT_SIZE(D, DF, 3) WHERE X is Basket, X description D, X description_format DF') |
436 rset = self.execute('Any LIMIT_SIZE(D, DF, 3) WHERE X is Basket, X description D, X description_format DF') |
437 self.failUnless(rset) |
437 self.failUnless(rset) |
438 self.failUnlessEqual(rset[0][0], 'hop...') |
438 self.failUnlessEqual(rset[0][0], 'hop...') |
439 |
439 |
440 def test_select_regproc_orderby(self): |
440 def test_select_regproc_orderby(self): |
441 rset = self.execute('DISTINCT Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is EGroup, X name N, X name "managers"') |
441 rset = self.execute('DISTINCT Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N, X name "managers"') |
442 self.failUnlessEqual(len(rset), 1) |
442 self.failUnlessEqual(len(rset), 1) |
443 self.failUnlessEqual(rset[0][1], 'managers') |
443 self.failUnlessEqual(rset[0][1], 'managers') |
444 rset = self.execute('Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is EGroup, X name N, NOT U in_group X, U login "admin"') |
444 rset = self.execute('Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N, NOT U in_group X, U login "admin"') |
445 self.failUnlessEqual(len(rset), 3) |
445 self.failUnlessEqual(len(rset), 3) |
446 self.failUnlessEqual(rset[0][1], 'owners') |
446 self.failUnlessEqual(rset[0][1], 'owners') |
447 |
447 |
448 def test_select_aggregat_sort(self): |
448 def test_select_aggregat_sort(self): |
449 rset = self.execute('Any G, COUNT(U) GROUPBY G ORDERBY 2 WHERE U in_group G') |
449 rset = self.execute('Any G, COUNT(U) GROUPBY G ORDERBY 2 WHERE U in_group G') |
450 self.assertEquals(len(rset.rows), 2) |
450 self.assertEquals(len(rset.rows), 2) |
451 self.assertEquals(len(rset.rows[0]), 2) |
451 self.assertEquals(len(rset.rows[0]), 2) |
452 self.assertEquals(rset.description[0], ('EGroup', 'Int',)) |
452 self.assertEquals(rset.description[0], ('CWGroup', 'Int',)) |
453 |
453 |
454 def test_select_aggregat_having(self): |
454 def test_select_aggregat_having(self): |
455 rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N ' |
455 rset = self.execute('Any N,COUNT(RDEF) GROUPBY N ORDERBY 2,N ' |
456 'WHERE RT name N, RDEF relation_type RT ' |
456 'WHERE RT name N, RDEF relation_type RT ' |
457 'HAVING COUNT(RDEF) > 10') |
457 'HAVING COUNT(RDEF) > 10') |
507 biduleeids = [r[0] for r in rset.rows] |
507 biduleeids = [r[0] for r in rset.rows] |
508 rset = self.execute(u'Any N where NOT N has_text "bidüle"') |
508 rset = self.execute(u'Any N where NOT N has_text "bidüle"') |
509 self.failIf([r[0] for r in rset.rows if r[0] in biduleeids]) |
509 self.failIf([r[0] for r in rset.rows if r[0] in biduleeids]) |
510 # duh? |
510 # duh? |
511 rset = self.execute('Any X WHERE X has_text %(text)s', {'text': u'ça'}) |
511 rset = self.execute('Any X WHERE X has_text %(text)s', {'text': u'ça'}) |
512 |
512 |
513 def test_select_searchable_text_2(self): |
513 def test_select_searchable_text_2(self): |
514 rset = self.execute("INSERT Personne X: X nom 'bidule'") |
514 rset = self.execute("INSERT Personne X: X nom 'bidule'") |
515 rset = self.execute("INSERT Personne X: X nom 'chouette'") |
515 rset = self.execute("INSERT Personne X: X nom 'chouette'") |
516 rset = self.execute("INSERT Societe X: X nom 'bidule'") |
516 rset = self.execute("INSERT Societe X: X nom 'bidule'") |
517 self.commit() |
517 self.commit() |
518 rset = self.execute('Personne N where N has_text "bidule"') |
518 rset = self.execute('Personne N where N has_text "bidule"') |
519 self.assertEquals(len(rset.rows), 1, rset.rows) |
519 self.assertEquals(len(rset.rows), 1, rset.rows) |
520 |
520 |
521 def test_select_searchable_text_3(self): |
521 def test_select_searchable_text_3(self): |
522 rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'M'") |
522 rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'M'") |
523 rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'F'") |
523 rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'F'") |
524 rset = self.execute("INSERT Societe X: X nom 'bidule'") |
524 rset = self.execute("INSERT Societe X: X nom 'bidule'") |
525 self.commit() |
525 self.commit() |
526 rset = self.execute('Any X where X has_text "bidule" and X sexe "M"') |
526 rset = self.execute('Any X where X has_text "bidule" and X sexe "M"') |
527 self.assertEquals(len(rset.rows), 1, rset.rows) |
527 self.assertEquals(len(rset.rows), 1, rset.rows) |
528 |
528 |
529 def test_select_multiple_searchable_text(self): |
529 def test_select_multiple_searchable_text(self): |
530 self.execute(u"INSERT Personne X: X nom 'bidüle'") |
530 self.execute(u"INSERT Personne X: X nom 'bidüle'") |
531 self.execute("INSERT Societe X: X nom 'chouette', S travaille X") |
531 self.execute("INSERT Societe X: X nom 'chouette', S travaille X") |
532 self.execute(u"INSERT Personne X: X nom 'bidüle'") |
532 self.execute(u"INSERT Personne X: X nom 'bidüle'") |
533 self.commit() |
533 self.commit() |
534 rset = self.execute('Personne X WHERE X has_text %(text)s, X travaille S, S has_text %(text2)s', |
534 rset = self.execute('Personne X WHERE X has_text %(text)s, X travaille S, S has_text %(text2)s', |
535 {'text': u'bidüle', |
535 {'text': u'bidüle', |
536 'text2': u'chouette',} |
536 'text2': u'chouette',} |
537 ) |
537 ) |
538 self.assertEquals(len(rset.rows), 1, rset.rows) |
538 self.assertEquals(len(rset.rows), 1, rset.rows) |
539 |
539 |
540 def test_select_no_descr(self): |
540 def test_select_no_descr(self): |
541 rset = self.execute('Any X WHERE X is EGroup', build_descr=0) |
541 rset = self.execute('Any X WHERE X is CWGroup', build_descr=0) |
542 rset.rows.sort() |
542 rset.rows.sort() |
543 self.assertEquals(tuplify(rset.rows), [(1,), (2,), (3,), (4,)]) |
543 self.assertEquals(tuplify(rset.rows), [(1,), (2,), (3,), (4,)]) |
544 self.assertEquals(rset.description, ()) |
544 self.assertEquals(rset.description, ()) |
545 |
545 |
546 def test_select_limit_offset(self): |
546 def test_select_limit_offset(self): |
547 rset = self.execute('EGroup X ORDERBY N LIMIT 2 WHERE X name N') |
547 rset = self.execute('CWGroup X ORDERBY N LIMIT 2 WHERE X name N') |
548 self.assertEquals(tuplify(rset.rows), [(3,), (1,)]) |
548 self.assertEquals(tuplify(rset.rows), [(3,), (1,)]) |
549 self.assertEquals(rset.description, [('EGroup',), ('EGroup',)]) |
549 self.assertEquals(rset.description, [('CWGroup',), ('CWGroup',)]) |
550 rset = self.execute('EGroup X ORDERBY N LIMIT 2 OFFSET 2 WHERE X name N') |
550 rset = self.execute('CWGroup X ORDERBY N LIMIT 2 OFFSET 2 WHERE X name N') |
551 self.assertEquals(tuplify(rset.rows), [(4,), (2,)]) |
551 self.assertEquals(tuplify(rset.rows), [(4,), (2,)]) |
552 |
552 |
553 def test_select_symetric(self): |
553 def test_select_symetric(self): |
554 self.execute("INSERT Personne X: X nom 'machin'") |
554 self.execute("INSERT Personne X: X nom 'machin'") |
555 self.execute("INSERT Personne X: X nom 'bidule'") |
555 self.execute("INSERT Personne X: X nom 'bidule'") |
556 self.execute("INSERT Personne X: X nom 'chouette'") |
556 self.execute("INSERT Personne X: X nom 'chouette'") |
557 self.execute("INSERT Personne X: X nom 'trucmuche'") |
557 self.execute("INSERT Personne X: X nom 'trucmuche'") |
648 |
648 |
649 def test_select_explicit_eid(self): |
649 def test_select_explicit_eid(self): |
650 rset = self.execute('Any X,E WHERE X owned_by U, X eid E, U eid %(u)s', {'u': self.session.user.eid}) |
650 rset = self.execute('Any X,E WHERE X owned_by U, X eid E, U eid %(u)s', {'u': self.session.user.eid}) |
651 self.failUnless(rset) |
651 self.failUnless(rset) |
652 self.assertEquals(rset.description[0][1], 'Int') |
652 self.assertEquals(rset.description[0][1], 'Int') |
653 |
653 |
654 # def test_select_rewritten_optional(self): |
654 # def test_select_rewritten_optional(self): |
655 # eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0] |
655 # eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0] |
656 # rset = self.execute('Any X WHERE X eid %(x)s, EXISTS(X owned_by U) OR EXISTS(X concerne S?, S owned_by U)', |
656 # rset = self.execute('Any X WHERE X eid %(x)s, EXISTS(X owned_by U) OR EXISTS(X concerne S?, S owned_by U)', |
657 # {'x': eid}, 'x') |
657 # {'x': eid}, 'x') |
658 # self.assertEquals(rset.rows, [[eid]]) |
658 # self.assertEquals(rset.rows, [[eid]]) |
659 |
659 |
660 def test_today_bug(self): |
660 def test_today_bug(self): |
661 self.execute("INSERT Tag X: X name 'bidule', X creation_date TODAY") |
661 self.execute("INSERT Tag X: X name 'bidule', X creation_date NOW") |
662 self.execute("INSERT Tag Y: Y name 'toto'") |
662 self.execute("INSERT Tag Y: Y name 'toto'") |
663 rset = self.execute("Any D WHERE X name in ('bidule', 'toto') , X creation_date D") |
663 rset = self.execute("Any D WHERE X name in ('bidule', 'toto') , X creation_date D") |
664 self.assert_(isinstance(rset.rows[0][0], DateTimeType), rset.rows) |
664 self.assert_(isinstance(rset.rows[0][0], datetime), rset.rows) |
665 rset = self.execute('Tag X WHERE X creation_date TODAY') |
665 rset = self.execute('Tag X WHERE X creation_date TODAY') |
666 self.assertEqual(len(rset.rows), 2) |
666 self.assertEqual(len(rset.rows), 2) |
667 rset = self.execute('Any MAX(D) WHERE X is Tag, X creation_date D') |
667 rset = self.execute('Any MAX(D) WHERE X is Tag, X creation_date D') |
668 self.failUnless(isinstance(rset[0][0], DateTimeType), type(rset[0][0])) |
668 self.failUnless(isinstance(rset[0][0], datetime), type(rset[0][0])) |
669 |
669 |
670 def test_today(self): |
670 def test_today(self): |
671 self.execute("INSERT Tag X: X name 'bidule', X creation_date TODAY") |
671 self.execute("INSERT Tag X: X name 'bidule', X creation_date TODAY") |
672 self.execute("INSERT Tag Y: Y name 'toto'") |
672 self.execute("INSERT Tag Y: Y name 'toto'") |
673 rset = self.execute('Tag X WHERE X creation_date TODAY') |
673 rset = self.execute('Tag X WHERE X creation_date TODAY') |
674 self.assertEqual(len(rset.rows), 2) |
674 self.assertEqual(len(rset.rows), 2) |
675 |
675 |
676 def test_select_boolean(self): |
676 def test_select_boolean(self): |
677 rset = self.execute('Any N WHERE X is EEType, X name N, X final %(val)s', |
677 rset = self.execute('Any N WHERE X is CWEType, X name N, X final %(val)s', |
678 {'val': True}) |
678 {'val': True}) |
679 self.assertEquals(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes', |
679 self.assertEquals(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes', |
680 'Date', 'Datetime', |
680 'Date', 'Datetime', |
681 'Decimal', 'Float', |
681 'Decimal', 'Float', |
682 'Int', 'Interval', |
682 'Int', 'Interval', |
683 'Password', 'String', |
683 'Password', 'String', |
684 'Time']) |
684 'Time']) |
685 rset = self.execute('Any N WHERE X is EEType, X name N, X final TRUE') |
685 rset = self.execute('Any N WHERE X is CWEType, X name N, X final TRUE') |
686 self.assertEquals(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes', |
686 self.assertEquals(sorted(r[0] for r in rset.rows), ['Boolean', 'Bytes', |
687 'Date', 'Datetime', |
687 'Date', 'Datetime', |
688 'Decimal', 'Float', |
688 'Decimal', 'Float', |
689 'Int', 'Interval', |
689 'Int', 'Interval', |
690 'Password', 'String', |
690 'Password', 'String', |
691 'Time']) |
691 'Time']) |
692 |
692 |
693 def test_select_constant(self): |
693 def test_select_constant(self): |
694 rset = self.execute('Any X, "toto" ORDERBY X WHERE X is EGroup') |
694 rset = self.execute('Any X, "toto" ORDERBY X WHERE X is CWGroup') |
695 self.assertEquals(rset.rows, |
695 self.assertEquals(rset.rows, |
696 map(list, zip((1,2,3,4), ('toto','toto','toto','toto',)))) |
696 map(list, zip((1,2,3,4), ('toto','toto','toto','toto',)))) |
697 self.assertIsInstance(rset[0][1], unicode) |
697 self.assertIsInstance(rset[0][1], unicode) |
698 self.assertEquals(rset.description, |
698 self.assertEquals(rset.description, |
699 zip(('EGroup', 'EGroup', 'EGroup', 'EGroup'), |
699 zip(('CWGroup', 'CWGroup', 'CWGroup', 'CWGroup'), |
700 ('String', 'String', 'String', 'String',))) |
700 ('String', 'String', 'String', 'String',))) |
701 rset = self.execute('Any X, %(value)s ORDERBY X WHERE X is EGroup', {'value': 'toto'}) |
701 rset = self.execute('Any X, %(value)s ORDERBY X WHERE X is CWGroup', {'value': 'toto'}) |
702 self.assertEquals(rset.rows, |
702 self.assertEquals(rset.rows, |
703 map(list, zip((1,2,3,4), ('toto','toto','toto','toto',)))) |
703 map(list, zip((1,2,3,4), ('toto','toto','toto','toto',)))) |
704 self.assertIsInstance(rset[0][1], unicode) |
704 self.assertIsInstance(rset[0][1], unicode) |
705 self.assertEquals(rset.description, |
705 self.assertEquals(rset.description, |
706 zip(('EGroup', 'EGroup', 'EGroup', 'EGroup'), |
706 zip(('CWGroup', 'CWGroup', 'CWGroup', 'CWGroup'), |
707 ('String', 'String', 'String', 'String',))) |
707 ('String', 'String', 'String', 'String',))) |
708 rset = self.execute('Any X,GN WHERE X is EUser, G is EGroup, X login "syt", X in_group G, G name GN') |
708 rset = self.execute('Any X,GN WHERE X is CWUser, G is CWGroup, X login "syt", X in_group G, G name GN') |
709 |
709 |
710 def test_select_union(self): |
710 def test_select_union(self): |
711 rset = self.execute('Any X,N ORDERBY N WITH X,N BEING ' |
711 rset = self.execute('Any X,N ORDERBY N WITH X,N BEING ' |
712 '((Any X,N WHERE X name N, X transition_of E, E name %(name)s)' |
712 '((Any X,N WHERE X name N, X transition_of E, E name %(name)s)' |
713 ' UNION ' |
713 ' UNION ' |
714 '(Any X,N WHERE X name N, X state_of E, E name %(name)s))', |
714 '(Any X,N WHERE X name N, X state_of E, E name %(name)s))', |
715 {'name': 'EUser'}) |
715 {'name': 'CWUser'}) |
716 self.assertEquals([x[1] for x in rset.rows], |
716 self.assertEquals([x[1] for x in rset.rows], |
717 ['activate', 'activated', 'deactivate', 'deactivated']) |
717 ['activate', 'activated', 'deactivate', 'deactivated']) |
718 self.assertEquals(rset.description, |
718 self.assertEquals(rset.description, |
719 [('Transition', 'String'), ('State', 'String'), |
719 [('Transition', 'String'), ('State', 'String'), |
720 ('Transition', 'String'), ('State', 'String')]) |
720 ('Transition', 'String'), ('State', 'String')]) |
721 |
721 |
722 def test_select_union_aggregat(self): |
722 def test_select_union_aggregat(self): |
723 # meaningless, the goal in to have group by done on different attribute |
723 # meaningless, the goal in to have group by done on different attribute |
724 # for each sub-query |
724 # for each sub-query |
725 self.execute('(Any N,COUNT(X) GROUPBY N WHERE X name N, X is State)' |
725 self.execute('(Any N,COUNT(X) GROUPBY N WHERE X name N, X is State)' |
726 ' UNION ' |
726 ' UNION ' |
727 '(Any N,COUNT(X) GROUPBY N ORDERBY 2 WHERE X login N)') |
727 '(Any N,COUNT(X) GROUPBY N ORDERBY 2 WHERE X login N)') |
728 |
728 |
729 def test_select_union_aggregat_independant_group(self): |
729 def test_select_union_aggregat_independant_group(self): |
730 self.execute('INSERT State X: X name "hop"') |
730 self.execute('INSERT State X: X name "hop"') |
731 self.execute('INSERT State X: X name "hop"') |
731 self.execute('INSERT State X: X name "hop"') |
732 self.execute('INSERT Transition X: X name "hop"') |
732 self.execute('INSERT Transition X: X name "hop"') |
733 self.execute('INSERT Transition X: X name "hop"') |
733 self.execute('INSERT Transition X: X name "hop"') |
734 rset = self.execute('Any N,NX ORDERBY 2 WITH N,NX BEING ' |
734 rset = self.execute('Any N,NX ORDERBY 2 WITH N,NX BEING ' |
735 '((Any N,COUNT(X) GROUPBY N WHERE X name N, X is State HAVING COUNT(X)>1)' |
735 '((Any N,COUNT(X) GROUPBY N WHERE X name N, X is State HAVING COUNT(X)>1)' |
736 ' UNION ' |
736 ' UNION ' |
737 '(Any N,COUNT(X) GROUPBY N WHERE X name N, X is Transition HAVING COUNT(X)>1))') |
737 '(Any N,COUNT(X) GROUPBY N WHERE X name N, X is Transition HAVING COUNT(X)>1))') |
738 self.assertEquals(rset.rows, [[u'hop', 2], [u'hop', 2]]) |
738 self.assertEquals(rset.rows, [[u'hop', 2], [u'hop', 2]]) |
739 |
739 |
740 def test_select_union_selection_with_diff_variables(self): |
740 def test_select_union_selection_with_diff_variables(self): |
741 rset = self.execute('(Any N WHERE X name N, X is State)' |
741 rset = self.execute('(Any N WHERE X name N, X is State)' |
742 ' UNION ' |
742 ' UNION ' |
743 '(Any NN WHERE XX name NN, XX is Transition)') |
743 '(Any NN WHERE XX name NN, XX is Transition)') |
744 self.assertEquals(sorted(r[0] for r in rset.rows), |
744 self.assertEquals(sorted(r[0] for r in rset.rows), |
745 ['abort', 'activate', 'activated', 'ben non', |
745 ['abort', 'activate', 'activated', 'ben non', |
746 'deactivate', 'deactivated', 'done', 'en cours', |
746 'deactivate', 'deactivated', 'done', 'en cours', |
747 'end', 'finie', 'markasdone', 'pitetre', 'redoit', |
747 'end', 'finie', 'markasdone', 'pitetre', 'redoit', |
748 'start', 'todo']) |
748 'start', 'todo']) |
749 |
749 |
750 def test_exists(self): |
750 def test_exists(self): |
751 geid = self.execute("INSERT EGroup X: X name 'lulufanclub'")[0][0] |
751 geid = self.execute("INSERT CWGroup X: X name 'lulufanclub'")[0][0] |
752 self.execute("SET U in_group G WHERE G name 'lulufanclub'") |
752 self.execute("SET U in_group G WHERE G name 'lulufanclub'") |
753 peid = self.execute("INSERT Personne X: X prenom 'lulu', X nom 'petit'")[0][0] |
753 peid = self.execute("INSERT Personne X: X prenom 'lulu', X nom 'petit'")[0][0] |
754 rset = self.execute("Any X WHERE X prenom 'lulu'," |
754 rset = self.execute("Any X WHERE X prenom 'lulu'," |
755 "EXISTS (U in_group G, G name 'lulufanclub' OR G name 'managers');") |
755 "EXISTS (U in_group G, G name 'lulufanclub' OR G name 'managers');") |
756 self.assertEquals(rset.rows, [[peid]]) |
756 self.assertEquals(rset.rows, [[peid]]) |
762 self.assertEquals(eid, 1) |
762 self.assertEquals(eid, 1) |
763 login = self.execute('Any L WHERE X login "admin", X identity Y, Y login L')[0][0] |
763 login = self.execute('Any L WHERE X login "admin", X identity Y, Y login L')[0][0] |
764 self.assertEquals(login, 'admin') |
764 self.assertEquals(login, 'admin') |
765 |
765 |
766 def test_select_date_mathexp(self): |
766 def test_select_date_mathexp(self): |
767 rset = self.execute('Any X, TODAY - CD WHERE X is EUser, X creation_date CD') |
767 rset = self.execute('Any X, TODAY - CD WHERE X is CWUser, X creation_date CD') |
768 self.failUnless(rset) |
768 self.failUnless(rset) |
769 self.failUnlessEqual(rset.description[0][1], 'Interval') |
769 self.failUnlessEqual(rset.description[0][1], 'Interval') |
770 eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0] |
770 eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0] |
771 rset = self.execute('Any X, NOW - CD WHERE X is Personne, X creation_date CD') |
771 rset = self.execute('Any X, NOW - CD WHERE X is Personne, X creation_date CD') |
772 self.failUnlessEqual(rset.description[0][1], 'Interval') |
772 self.failUnlessEqual(rset.description[0][1], 'Interval') |
773 # sqlite bug |
|
774 #from mx.DateTime import DateTimeDeltaType |
|
775 #self.assertIsInstance(rset[0][1], DateTimeDeltaType) |
|
776 #self.failUnless(rset[0][1].seconds > 0) |
|
777 |
773 |
778 def test_select_subquery_aggregat(self): |
774 def test_select_subquery_aggregat(self): |
779 # percent users by groups |
775 # percent users by groups |
780 self.execute('SET X in_group G WHERE G name "users"') |
776 self.execute('SET X in_group G WHERE G name "users"') |
781 rset = self.execute('Any GN, COUNT(X)*100/T GROUPBY GN ORDERBY 2,1' |
777 rset = self.execute('Any GN, COUNT(X)*100/T GROUPBY GN ORDERBY 2,1' |
782 ' WHERE G name GN, X in_group G' |
778 ' WHERE G name GN, X in_group G' |
783 ' WITH T BEING (Any COUNT(U) WHERE U is EUser)') |
779 ' WITH T BEING (Any COUNT(U) WHERE U is CWUser)') |
784 self.assertEquals(rset.rows, [[u'guests', 50], [u'managers', 50], [u'users', 100]]) |
780 self.assertEquals(rset.rows, [[u'guests', 50], [u'managers', 50], [u'users', 100]]) |
785 self.assertEquals(rset.description, [('String', 'Int'), ('String', 'Int'), ('String', 'Int')]) |
781 self.assertEquals(rset.description, [('String', 'Int'), ('String', 'Int'), ('String', 'Int')]) |
786 |
782 |
787 def test_select_subquery_const(self): |
783 def test_select_subquery_const(self): |
788 rset = self.execute('Any X WITH X BEING ((Any NULL) UNION (Any "toto"))') |
784 rset = self.execute('Any X WITH X BEING ((Any NULL) UNION (Any "toto"))') |
789 self.assertEquals(rset.rows, [[None], ['toto']]) |
785 self.assertEquals(rset.rows, [[None], ['toto']]) |
790 self.assertEquals(rset.description, [(None,), ('String',)]) |
786 self.assertEquals(rset.description, [(None,), ('String',)]) |
791 |
787 |
792 # insertion queries tests ################################################# |
788 # insertion queries tests ################################################# |
793 |
789 |
794 def test_insert_is(self): |
790 def test_insert_is(self): |
795 eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0] |
791 eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0] |
796 etype, = self.execute("Any TN WHERE X is T, X eid %s, T name TN" % eid)[0] |
792 etype, = self.execute("Any TN WHERE X is T, X eid %s, T name TN" % eid)[0] |
797 self.assertEquals(etype, 'Personne') |
793 self.assertEquals(etype, 'Personne') |
798 self.execute("INSERT Personne X: X nom 'managers'") |
794 self.execute("INSERT Personne X: X nom 'managers'") |
799 |
795 |
800 def test_insert_1(self): |
796 def test_insert_1(self): |
801 rset = self.execute("INSERT Personne X: X nom 'bidule'") |
797 rset = self.execute("INSERT Personne X: X nom 'bidule'") |
802 self.assertEquals(len(rset.rows), 1) |
798 self.assertEquals(len(rset.rows), 1) |
803 self.assertEquals(rset.description, [('Personne',)]) |
799 self.assertEquals(rset.description, [('Personne',)]) |
804 rset = self.execute('Personne X WHERE X nom "bidule"') |
800 rset = self.execute('Personne X WHERE X nom "bidule"') |
981 self.assertEquals(rset.rows, []) |
977 self.assertEquals(rset.rows, []) |
982 rset = self.execute("Folder X WHERE X eid %(x)s", {'x': eid}, 'x') |
978 rset = self.execute("Folder X WHERE X eid %(x)s", {'x': eid}, 'x') |
983 self.assertEquals(rset.rows, []) |
979 self.assertEquals(rset.rows, []) |
984 rset = self.execute("Folder X WHERE X eid %s" %eid) |
980 rset = self.execute("Folder X WHERE X eid %s" %eid) |
985 self.assertEquals(rset.rows, []) |
981 self.assertEquals(rset.rows, []) |
986 |
982 |
987 # update queries tests #################################################### |
983 # update queries tests #################################################### |
988 |
984 |
989 def test_update_1(self): |
985 def test_update_1(self): |
990 self.execute("INSERT Personne Y: Y nom 'toto'") |
986 self.execute("INSERT Personne Y: Y nom 'toto'") |
991 rset = self.execute('Personne X WHERE X nom "toto"') |
987 rset = self.execute('Personne X WHERE X nom "toto"') |
992 self.assertEqual(len(rset.rows), 1) |
988 self.assertEqual(len(rset.rows), 1) |
993 self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'") |
989 self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'") |
994 rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z') |
990 rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z') |
995 self.assertEqual(tuplify(rset.rows), [('tutu', 'original')]) |
991 self.assertEqual(tuplify(rset.rows), [('tutu', 'original')]) |
996 |
992 |
997 def test_update_2(self): |
993 def test_update_2(self): |
998 self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
994 self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
999 #rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto"') |
995 #rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto"') |
1000 #self.assertEqual(len(rset.rows), 1) |
996 #self.assertEqual(len(rset.rows), 1) |
1001 #rset = self.execute('Any X, Y WHERE X travaille Y') |
997 #rset = self.execute('Any X, Y WHERE X travaille Y') |
1002 #self.assertEqual(len(rset.rows), 0) |
998 #self.assertEqual(len(rset.rows), 0) |
1003 self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'") |
999 self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'") |
1004 rset = self.execute('Any X, Y WHERE X travaille Y') |
1000 rset = self.execute('Any X, Y WHERE X travaille Y') |
1005 self.assertEqual(len(rset.rows), 1) |
1001 self.assertEqual(len(rset.rows), 1) |
1006 |
1002 |
1007 def test_update_2bis(self): |
1003 def test_update_2bis(self): |
1008 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1004 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1009 eid1, eid2 = rset[0][0], rset[0][1] |
1005 eid1, eid2 = rset[0][0], rset[0][1] |
1010 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1006 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1011 {'x': str(eid1), 'y': str(eid2)}) |
1007 {'x': str(eid1), 'y': str(eid2)}) |
1012 rset = self.execute('Any X, Y WHERE X travaille Y') |
1008 rset = self.execute('Any X, Y WHERE X travaille Y') |
1013 self.assertEqual(len(rset.rows), 1) |
1009 self.assertEqual(len(rset.rows), 1) |
1014 |
1010 |
1015 def test_update_2ter(self): |
1011 def test_update_2ter(self): |
1016 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1012 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1017 eid1, eid2 = rset[0][0], rset[0][1] |
1013 eid1, eid2 = rset[0][0], rset[0][1] |
1018 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1014 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1019 {'x': unicode(eid1), 'y': unicode(eid2)}) |
1015 {'x': unicode(eid1), 'y': unicode(eid2)}) |
1020 rset = self.execute('Any X, Y WHERE X travaille Y') |
1016 rset = self.execute('Any X, Y WHERE X travaille Y') |
1021 self.assertEqual(len(rset.rows), 1) |
1017 self.assertEqual(len(rset.rows), 1) |
1022 |
1018 |
1023 ## def test_update_4(self): |
1019 ## def test_update_4(self): |
1024 ## self.execute("SET X know Y WHERE X ami Y") |
1020 ## self.execute("SET X know Y WHERE X ami Y") |
1025 |
1021 |
1026 def test_update_multiple1(self): |
1022 def test_update_multiple1(self): |
1027 peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0] |
1023 peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0] |
1028 peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0] |
1024 peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0] |
1029 self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'") |
1025 self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'") |
1030 self.assertEquals(self.execute('Any X WHERE X nom "toto"').rows, [[peid1]]) |
1026 self.assertEquals(self.execute('Any X WHERE X nom "toto"').rows, [[peid1]]) |
1031 self.assertEquals(self.execute('Any X WHERE X nom "tutu"').rows, [[peid2]]) |
1027 self.assertEquals(self.execute('Any X WHERE X nom "tutu"').rows, [[peid2]]) |
1032 |
1028 |
1033 def test_update_multiple2(self): |
1029 def test_update_multiple2(self): |
1034 ueid = self.execute("INSERT EUser X: X login 'bob', X upassword 'toto'")[0][0] |
1030 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0] |
1035 peid1 = self.execute("INSERT Personne Y: Y nom 'turlu'")[0][0] |
1031 peid1 = self.execute("INSERT Personne Y: Y nom 'turlu'")[0][0] |
1036 peid2 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0] |
1032 peid2 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0] |
1037 self.execute('SET P1 owned_by U, P2 owned_by U ' |
1033 self.execute('SET P1 owned_by U, P2 owned_by U ' |
1038 'WHERE P1 eid %s, P2 eid %s, U eid %s' % (peid1, peid2, ueid)) |
1034 'WHERE P1 eid %s, P2 eid %s, U eid %s' % (peid1, peid2, ueid)) |
1039 self.failUnless(self.execute('Any X WHERE X eid %s, X owned_by U, U eid %s' |
1035 self.failUnless(self.execute('Any X WHERE X eid %s, X owned_by U, U eid %s' |
1056 def test_update_string_concat(self): |
1052 def test_update_string_concat(self): |
1057 beid = self.execute("INSERT Bookmark Y: Y title 'toto', Y path '/view'")[0][0] |
1053 beid = self.execute("INSERT Bookmark Y: Y title 'toto', Y path '/view'")[0][0] |
1058 self.execute('SET X title XN + %(suffix)s WHERE X is Bookmark, X title XN', {'suffix': u'-moved'}) |
1054 self.execute('SET X title XN + %(suffix)s WHERE X is Bookmark, X title XN', {'suffix': u'-moved'}) |
1059 newname = self.execute('Any XN WHERE X eid %(x)s, X title XN', {'x': beid}, 'x')[0][0] |
1055 newname = self.execute('Any XN WHERE X eid %(x)s, X title XN', {'x': beid}, 'x')[0][0] |
1060 self.assertEquals(newname, 'toto-moved') |
1056 self.assertEquals(newname, 'toto-moved') |
1061 |
1057 |
1062 def test_update_query_error(self): |
1058 def test_update_query_error(self): |
1063 self.execute("INSERT Personne Y: Y nom 'toto'") |
1059 self.execute("INSERT Personne Y: Y nom 'toto'") |
1064 self.assertRaises(Exception, self.execute, "SET X nom 'toto', X is Personne") |
1060 self.assertRaises(Exception, self.execute, "SET X nom 'toto', X is Personne") |
1065 self.assertRaises(QueryError, self.execute, "SET X nom 'toto', X has_text 'tutu' WHERE X is Personne") |
1061 self.assertRaises(QueryError, self.execute, "SET X nom 'toto', X has_text 'tutu' WHERE X is Personne") |
1066 self.assertRaises(QueryError, self.execute, "SET X login 'tutu', X eid %s" % cnx.user(self.session).eid) |
1062 self.assertRaises(QueryError, self.execute, "SET X login 'tutu', X eid %s" % cnx.user(self.session).eid) |
1067 |
1063 |
1068 |
1064 |
1069 # upassword encryption tests ################################################# |
1065 # upassword encryption tests ################################################# |
1070 |
1066 |
1071 def test_insert_upassword(self): |
1067 def test_insert_upassword(self): |
1072 rset = self.execute("INSERT EUser X: X login 'bob', X upassword 'toto'") |
1068 rset = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'") |
1073 self.assertEquals(len(rset.rows), 1) |
1069 self.assertEquals(len(rset.rows), 1) |
1074 self.assertEquals(rset.description, [('EUser',)]) |
1070 self.assertEquals(rset.description, [('CWUser',)]) |
1075 self.assertRaises(Unauthorized, |
1071 self.assertRaises(Unauthorized, |
1076 self.execute, "Any P WHERE X is EUser, X login 'bob', X upassword P") |
1072 self.execute, "Any P WHERE X is CWUser, X login 'bob', X upassword P") |
1077 cursor = self.pool['system'] |
1073 cursor = self.pool['system'] |
1078 cursor.execute("SELECT %supassword from %sEUser WHERE %slogin='bob'" |
1074 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1079 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1075 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1080 passwd = cursor.fetchone()[0].getvalue() |
1076 passwd = cursor.fetchone()[0].getvalue() |
1081 self.assertEquals(passwd, crypt_password('toto', passwd[:2])) |
1077 self.assertEquals(passwd, crypt_password('toto', passwd[:2])) |
1082 rset = self.execute("Any X WHERE X is EUser, X login 'bob', X upassword '%s'" % passwd) |
1078 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword '%s'" % passwd) |
1083 self.assertEquals(len(rset.rows), 1) |
1079 self.assertEquals(len(rset.rows), 1) |
1084 self.assertEquals(rset.description, [('EUser',)]) |
1080 self.assertEquals(rset.description, [('CWUser',)]) |
1085 |
1081 |
1086 def test_update_upassword(self): |
1082 def test_update_upassword(self): |
1087 cursor = self.pool['system'] |
1083 cursor = self.pool['system'] |
1088 rset = self.execute("INSERT EUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'}) |
1084 rset = self.execute("INSERT CWUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'}) |
1089 self.assertEquals(rset.description[0][0], 'EUser') |
1085 self.assertEquals(rset.description[0][0], 'CWUser') |
1090 rset = self.execute("SET X upassword %(pwd)s WHERE X is EUser, X login 'bob'", |
1086 rset = self.execute("SET X upassword %(pwd)s WHERE X is CWUser, X login 'bob'", |
1091 {'pwd': 'tutu'}) |
1087 {'pwd': 'tutu'}) |
1092 cursor.execute("SELECT %supassword from %sEUser WHERE %slogin='bob'" |
1088 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1093 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1089 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1094 passwd = cursor.fetchone()[0].getvalue() |
1090 passwd = cursor.fetchone()[0].getvalue() |
1095 self.assertEquals(passwd, crypt_password('tutu', passwd[:2])) |
1091 self.assertEquals(passwd, crypt_password('tutu', passwd[:2])) |
1096 rset = self.execute("Any X WHERE X is EUser, X login 'bob', X upassword '%s'" % passwd) |
1092 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword '%s'" % passwd) |
1097 self.assertEquals(len(rset.rows), 1) |
1093 self.assertEquals(len(rset.rows), 1) |
1098 self.assertEquals(rset.description, [('EUser',)]) |
1094 self.assertEquals(rset.description, [('CWUser',)]) |
1099 |
1095 |
1100 # non regression tests #################################################### |
1096 # non regression tests #################################################### |
1101 |
1097 |
1102 def test_nonregr_1(self): |
1098 def test_nonregr_1(self): |
1103 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
1099 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
1104 self.execute("SET X tags Y WHERE X name 'tag', Y is State, Y name 'activated'") |
1100 self.execute("SET X tags Y WHERE X name 'tag', Y is State, Y name 'activated'") |
1105 rset = self.execute('Any X WHERE T tags X') |
1101 rset = self.execute('Any X WHERE T tags X') |
1106 self.assertEquals(len(rset.rows), 1, rset.rows) |
1102 self.assertEquals(len(rset.rows), 1, rset.rows) |
1245 self.assertEquals(len(rset), 1) |
1241 self.assertEquals(len(rset), 1) |
1246 |
1242 |
1247 |
1243 |
1248 def test_nonregr_set_datetime(self): |
1244 def test_nonregr_set_datetime(self): |
1249 # huum, psycopg specific |
1245 # huum, psycopg specific |
1250 self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': today()}) |
1246 self.execute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()}) |
1251 |
1247 |
1252 def test_nonregr_set_query(self): |
1248 def test_nonregr_set_query(self): |
1253 ueid = self.execute("INSERT EUser X: X login 'bob', X upassword 'toto'")[0][0] |
1249 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0] |
1254 self.execute("SET E in_group G, E in_state S, " |
1250 self.execute("SET E in_group G, E in_state S, " |
1255 "E firstname %(firstname)s, E surname %(surname)s " |
1251 "E firstname %(firstname)s, E surname %(surname)s " |
1256 "WHERE E eid %(x)s, G name 'users', S name 'activated'", |
1252 "WHERE E eid %(x)s, G name 'users', S name 'activated'", |
1257 {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x') |
1253 {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x') |
1258 |
1254 |
1259 def test_nonregr_u_owned_by_u(self): |
1255 def test_nonregr_u_owned_by_u(self): |
1260 ueid = self.execute("INSERT EUser X: X login 'bob', X upassword 'toto', X in_group G " |
1256 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G " |
1261 "WHERE G name 'users'")[0][0] |
1257 "WHERE G name 'users'")[0][0] |
1262 rset = self.execute("EUser U") |
1258 rset = self.execute("CWUser U") |
1263 self.assertEquals(len(rset), 3) # bob + admin + anon |
1259 self.assertEquals(len(rset), 3) # bob + admin + anon |
1264 rset = self.execute("Any U WHERE NOT U owned_by U") |
1260 rset = self.execute("Any U WHERE NOT U owned_by U") |
1265 self.assertEquals(len(rset), 0) # even admin created at repo initialization time should belong to itself |
1261 self.assertEquals(len(rset), 0) # even admin created at repo initialization time should belong to itself |
1266 |
1262 |
1267 def test_nonreg_update_index(self): |
1263 def test_nonreg_update_index(self): |
1268 # this is the kind of queries generated by "cubicweb-ctl db-check -ry" |
1264 # this is the kind of queries generated by "cubicweb-ctl db-check -ry" |
1269 self.execute("SET X description D WHERE X is State, X description D") |
1265 self.execute("SET X description D WHERE X is State, X description D") |
1270 |
1266 |
1271 def test_nonregr_is(self): |
1267 def test_nonregr_is(self): |
1272 uteid = self.execute('Any ET WHERE ET name "EUser"')[0][0] |
1268 uteid = self.execute('Any ET WHERE ET name "CWUser"')[0][0] |
1273 self.execute('Any X, ET WHERE X is ET, ET eid %s' % uteid) |
1269 self.execute('Any X, ET WHERE X is ET, ET eid %s' % uteid) |
1274 |
1270 |
1275 def test_nonregr_orderby(self): |
1271 def test_nonregr_orderby(self): |
1276 seid = self.execute('Any X WHERE X name "activated"')[0][0] |
1272 seid = self.execute('Any X WHERE X name "activated"')[0][0] |
1277 self.execute('Any X,S, MAX(T) GROUPBY X,S ORDERBY S WHERE X is EUser, T tags X, S eid IN(%s), X in_state S' % seid) |
1273 self.execute('Any X,S, MAX(T) GROUPBY X,S ORDERBY S WHERE X is CWUser, T tags X, S eid IN(%s), X in_state S' % seid) |
1278 |
1274 |
1279 def test_nonregr_solution_cache(self): |
1275 def test_nonregr_solution_cache(self): |
1280 self.skip('XXX should be fixed or documented') # (doesn't occur if cache key is provided.) |
1276 self.skip('XXX should be fixed or documented') # (doesn't occur if cache key is provided.) |
1281 rset = self.execute('Any X WHERE X is EUser, X eid %(x)s', {'x':self.ueid}) |
1277 rset = self.execute('Any X WHERE X is CWUser, X eid %(x)s', {'x':self.ueid}) |
1282 self.assertEquals(len(rset), 1) |
1278 self.assertEquals(len(rset), 1) |
1283 rset = self.execute('Any X WHERE X is EUser, X eid %(x)s', {'x':12345}) |
1279 rset = self.execute('Any X WHERE X is CWUser, X eid %(x)s', {'x':12345}) |
1284 self.assertEquals(len(rset), 0) |
1280 self.assertEquals(len(rset), 0) |
1285 |
1281 |
1286 def test_nonregr_final_norestr(self): |
1282 def test_nonregr_final_norestr(self): |
1287 self.assertRaises(BadRQLQuery, self.execute, 'Date X') |
1283 self.assertRaises(BadRQLQuery, self.execute, 'Date X') |
1288 |
1284 |