24 sourcefile = 'sources_extern' |
24 sourcefile = 'sources_extern' |
25 |
25 |
26 class ExternalSource2Configuration(TestServerConfiguration): |
26 class ExternalSource2Configuration(TestServerConfiguration): |
27 sourcefile = 'sources_multi2' |
27 sourcefile = 'sources_multi2' |
28 |
28 |
29 repo2, cnx2 = init_test_database('sqlite', config=ExternalSource1Configuration('data')) |
|
30 cu = cnx2.cursor() |
|
31 ec1 = cu.execute('INSERT Card X: X title "C3: An external card", X wikiid "aaa"')[0][0] |
|
32 cu.execute('INSERT Card X: X title "C4: Ze external card", X wikiid "zzz"') |
|
33 aff1 = cu.execute('INSERT Affaire X: X ref "AFFREF"')[0][0] |
|
34 cnx2.commit() |
|
35 |
|
36 MTIME = datetime.now() - timedelta(0, 10) |
29 MTIME = datetime.now() - timedelta(0, 10) |
37 |
30 repo2, cnx2 = init_test_database(config=ExternalSource1Configuration('data')) |
38 repo3, cnx3 = init_test_database('sqlite', config=ExternalSource2Configuration('data')) |
31 repo3, cnx3 = init_test_database(config=ExternalSource2Configuration('data')) |
39 |
32 |
40 # XXX, access existing connection, no pyro connection |
33 # hi-jacking |
41 from cubicweb.server.sources.pyrorql import PyroRQLSource |
34 from cubicweb.server.sources.pyrorql import PyroRQLSource |
42 PyroRQLSource.get_connection = lambda x: x.uri == 'extern-multi' and cnx3 or cnx2 |
|
43 # necessary since the repository is closing its initial connections pool though |
|
44 # we want to keep cnx2 valid |
|
45 from cubicweb.dbapi import Connection |
35 from cubicweb.dbapi import Connection |
46 Connection.close = lambda x: None |
36 |
47 |
37 PyroRQLSource_get_connection = PyroRQLSource.get_connection |
48 class TwoSourcesTC(RepositoryBasedTC): |
38 Connection_close = Connection.close |
49 repo_config = TwoSourcesConfiguration('data') |
39 |
|
40 def setup_module(*args): |
|
41 # hi-jack PyroRQLSource.get_connection to access existing connection (no |
|
42 # pyro connection) |
|
43 PyroRQLSource.get_connection = lambda x: x.uri == 'extern-multi' and cnx3 or cnx2 |
|
44 # also necessary since the repository is closing its initial connections |
|
45 # pool though we want to keep cnx2 valid |
|
46 Connection.close = lambda x: None |
|
47 |
|
48 def teardown_module(*args): |
|
49 PyroRQLSource.get_connection = PyroRQLSource_get_connection |
|
50 Connection.close = Connection_close |
|
51 |
|
52 |
|
53 class TwoSourcesTC(CubicWebTC): |
|
54 config = TwoSourcesConfiguration('data') |
|
55 |
|
56 @classmethod |
|
57 def _refresh_repo(cls): |
|
58 super(TwoSourcesTC, cls)._refresh_repo() |
|
59 cnx2.rollback() |
|
60 refresh_repo(repo2) |
|
61 cnx3.rollback() |
|
62 refresh_repo(repo3) |
50 |
63 |
51 def setUp(self): |
64 def setUp(self): |
52 RepositoryBasedTC.setUp(self) |
65 CubicWebTC.setUp(self) |
53 self.repo.sources[-1]._query_cache.clear() |
66 do_monkey_patch() |
54 self.repo.sources[-2]._query_cache.clear() |
67 |
|
68 def tearDown(self): |
|
69 CubicWebTC.tearDown(self) |
|
70 undo_monkey_patch() |
|
71 |
|
72 def setup_database(self): |
|
73 cu = cnx2.cursor() |
|
74 self.ec1 = cu.execute('INSERT Card X: X title "C3: An external card", X wikiid "aaa"')[0][0] |
|
75 cu.execute('INSERT Card X: X title "C4: Ze external card", X wikiid "zzz"') |
|
76 self.aff1 = cu.execute('INSERT Affaire X: X ref "AFFREF"')[0][0] |
|
77 cnx2.commit() |
55 # trigger discovery |
78 # trigger discovery |
56 self.execute('Card X') |
79 self.sexecute('Card X') |
57 self.execute('Affaire X') |
80 self.sexecute('Affaire X') |
58 self.execute('State X') |
81 self.sexecute('State X') |
59 self.commit() |
|
60 # don't delete external entities! |
|
61 self.maxeid = self.session.system_sql('SELECT MAX(eid) FROM entities').fetchone()[0] |
|
62 # add some entities |
82 # add some entities |
63 self.ic1 = self.execute('INSERT Card X: X title "C1: An internal card", X wikiid "aaai"')[0][0] |
83 self.ic1 = self.sexecute('INSERT Card X: X title "C1: An internal card", X wikiid "aaai"')[0][0] |
64 self.ic2 = self.execute('INSERT Card X: X title "C2: Ze internal card", X wikiid "zzzi"')[0][0] |
84 self.ic2 = self.sexecute('INSERT Card X: X title "C2: Ze internal card", X wikiid "zzzi"')[0][0] |
65 self.commit() |
|
66 do_monkey_patch() |
|
67 |
|
68 def tearDown(self): |
|
69 RepositoryBasedTC.tearDown(self) |
|
70 undo_monkey_patch() |
|
71 |
85 |
72 def test_eid_comp(self): |
86 def test_eid_comp(self): |
73 rset = self.execute('Card X WHERE X eid > 1') |
87 rset = self.sexecute('Card X WHERE X eid > 1') |
74 self.assertEquals(len(rset), 4) |
88 self.assertEquals(len(rset), 4) |
75 rset = self.execute('Any X,T WHERE X title T, X eid > 1') |
89 rset = self.sexecute('Any X,T WHERE X title T, X eid > 1') |
76 self.assertEquals(len(rset), 4) |
90 self.assertEquals(len(rset), 4) |
77 |
91 |
78 def test_metainformation(self): |
92 def test_metainformation(self): |
79 rset = self.execute('Card X ORDERBY T WHERE X title T') |
93 rset = self.sexecute('Card X ORDERBY T WHERE X title T') |
80 # 2 added to the system source, 2 added to the external source |
94 # 2 added to the system source, 2 added to the external source |
81 self.assertEquals(len(rset), 4) |
95 self.assertEquals(len(rset), 4) |
82 # since they are orderd by eid, we know the 3 first one is coming from the system source |
96 # since they are orderd by eid, we know the 3 first one is coming from the system source |
83 # and the others from external source |
97 # and the others from external source |
84 self.assertEquals(rset.get_entity(0, 0).metainformation(), |
98 self.assertEquals(rset.get_entity(0, 0).metainformation(), |
87 externent = rset.get_entity(3, 0) |
101 externent = rset.get_entity(3, 0) |
88 metainf = externent.metainformation() |
102 metainf = externent.metainformation() |
89 self.assertEquals(metainf['source'], {'adapter': 'pyrorql', 'base-url': 'http://extern.org/', 'uri': 'extern'}) |
103 self.assertEquals(metainf['source'], {'adapter': 'pyrorql', 'base-url': 'http://extern.org/', 'uri': 'extern'}) |
90 self.assertEquals(metainf['type'], 'Card') |
104 self.assertEquals(metainf['type'], 'Card') |
91 self.assert_(metainf['extid']) |
105 self.assert_(metainf['extid']) |
92 etype = self.execute('Any ETN WHERE X is ET, ET name ETN, X eid %(x)s', |
106 etype = self.sexecute('Any ETN WHERE X is ET, ET name ETN, X eid %(x)s', |
93 {'x': externent.eid}, 'x')[0][0] |
107 {'x': externent.eid}, 'x')[0][0] |
94 self.assertEquals(etype, 'Card') |
108 self.assertEquals(etype, 'Card') |
95 |
109 |
96 def test_order_limit_offset(self): |
110 def test_order_limit_offset(self): |
97 rsetbase = self.execute('Any W,X ORDERBY W,X WHERE X wikiid W') |
111 rsetbase = self.sexecute('Any W,X ORDERBY W,X WHERE X wikiid W') |
98 self.assertEquals(len(rsetbase), 4) |
112 self.assertEquals(len(rsetbase), 4) |
99 self.assertEquals(sorted(rsetbase.rows), rsetbase.rows) |
113 self.assertEquals(sorted(rsetbase.rows), rsetbase.rows) |
100 rset = self.execute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W') |
114 rset = self.sexecute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W') |
101 self.assertEquals(rset.rows, rsetbase.rows[2:4]) |
115 self.assertEquals(rset.rows, rsetbase.rows[2:4]) |
102 |
116 |
103 def test_has_text(self): |
117 def test_has_text(self): |
104 self.repo.sources_by_uri['extern'].synchronize(MTIME) # in case fti_update has been run before |
118 self.repo.sources_by_uri['extern'].synchronize(MTIME) # in case fti_update has been run before |
105 self.failUnless(self.execute('Any X WHERE X has_text "affref"')) |
119 self.failUnless(self.sexecute('Any X WHERE X has_text "affref"')) |
106 self.failUnless(self.execute('Affaire X WHERE X has_text "affref"')) |
120 self.failUnless(self.sexecute('Affaire X WHERE X has_text "affref"')) |
107 |
121 |
108 def test_anon_has_text(self): |
122 def test_anon_has_text(self): |
109 self.repo.sources_by_uri['extern'].synchronize(MTIME) # in case fti_update has been run before |
123 self.repo.sources_by_uri['extern'].synchronize(MTIME) # in case fti_update has been run before |
110 self.execute('INSERT Affaire X: X ref "no readable card"')[0][0] |
124 self.sexecute('INSERT Affaire X: X ref "no readable card"')[0][0] |
111 aff1 = self.execute('INSERT Affaire X: X ref "card"')[0][0] |
125 aff1 = self.sexecute('INSERT Affaire X: X ref "card"')[0][0] |
112 # grant read access |
126 # grant read access |
113 self.execute('SET X owned_by U WHERE X eid %(x)s, U login "anon"', {'x': aff1}, 'x') |
127 self.sexecute('SET X owned_by U WHERE X eid %(x)s, U login "anon"', {'x': aff1}, 'x') |
114 self.commit() |
128 self.commit() |
115 cnx = self.login('anon') |
129 cnx = self.login('anon') |
116 cu = cnx.cursor() |
130 cu = cnx.cursor() |
117 rset = cu.execute('Any X WHERE X has_text "card"') |
131 rset = cu.execute('Any X WHERE X has_text "card"') |
118 self.assertEquals(len(rset), 5, zip(rset.rows, rset.description)) |
132 self.assertEquals(len(rset), 5, zip(rset.rows, rset.description)) |
119 cnx.close() |
133 cnx.close() |
120 |
134 |
121 def test_synchronization(self): |
135 def test_synchronization(self): |
122 cu = cnx2.cursor() |
136 cu = cnx2.cursor() |
123 assert cu.execute('Any X WHERE X eid %(x)s', {'x': aff1}, 'x') |
137 assert cu.execute('Any X WHERE X eid %(x)s', {'x': self.aff1}, 'x') |
124 cu.execute('SET X ref "BLAH" WHERE X eid %(x)s', {'x': aff1}, 'x') |
138 cu.execute('SET X ref "BLAH" WHERE X eid %(x)s', {'x': self.aff1}, 'x') |
125 aff2 = cu.execute('INSERT Affaire X: X ref "AFFREUX"')[0][0] |
139 aff2 = cu.execute('INSERT Affaire X: X ref "AFFREUX"')[0][0] |
126 cnx2.commit() |
140 cnx2.commit() |
127 try: |
141 try: |
128 # force sync |
142 # force sync |
129 self.repo.sources_by_uri['extern'].synchronize(MTIME) |
143 self.repo.sources_by_uri['extern'].synchronize(MTIME) |
130 self.failUnless(self.execute('Any X WHERE X has_text "blah"')) |
144 self.failUnless(self.sexecute('Any X WHERE X has_text "blah"')) |
131 self.failUnless(self.execute('Any X WHERE X has_text "affreux"')) |
145 self.failUnless(self.sexecute('Any X WHERE X has_text "affreux"')) |
132 cu.execute('DELETE Affaire X WHERE X eid %(x)s', {'x': aff2}) |
146 cu.execute('DELETE Affaire X WHERE X eid %(x)s', {'x': aff2}) |
133 cnx2.commit() |
147 cnx2.commit() |
134 self.repo.sources_by_uri['extern'].synchronize(MTIME) |
148 self.repo.sources_by_uri['extern'].synchronize(MTIME) |
135 rset = self.execute('Any X WHERE X has_text "affreux"') |
149 rset = self.sexecute('Any X WHERE X has_text "affreux"') |
136 self.failIf(rset) |
150 self.failIf(rset) |
137 finally: |
151 finally: |
138 # restore state |
152 # restore state |
139 cu.execute('SET X ref "AFFREF" WHERE X eid %(x)s', {'x': aff1}, 'x') |
153 cu.execute('SET X ref "AFFREF" WHERE X eid %(x)s', {'x': self.aff1}, 'x') |
140 cnx2.commit() |
154 cnx2.commit() |
141 |
155 |
142 def test_simplifiable_var(self): |
156 def test_simplifiable_var(self): |
143 affeid = self.execute('Affaire X WHERE X ref "AFFREF"')[0][0] |
157 affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0] |
144 rset = self.execute('Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB', |
158 rset = self.sexecute('Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB', |
145 {'x': affeid}, 'x') |
159 {'x': affeid}, 'x') |
146 self.assertEquals(len(rset), 1) |
160 self.assertEquals(len(rset), 1) |
147 self.assertEquals(rset[0][1], "pitetre") |
161 self.assertEquals(rset[0][1], "pitetre") |
148 |
162 |
149 def test_simplifiable_var_2(self): |
163 def test_simplifiable_var_2(self): |
150 affeid = self.execute('Affaire X WHERE X ref "AFFREF"')[0][0] |
164 affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0] |
151 rset = self.execute('Any E WHERE E eid %(x)s, E in_state S, NOT S name "moved"', |
165 rset = self.sexecute('Any E WHERE E eid %(x)s, E in_state S, NOT S name "moved"', |
152 {'x': affeid, 'u': self.session.user.eid}, 'x') |
166 {'x': affeid, 'u': self.session.user.eid}, 'x') |
153 self.assertEquals(len(rset), 1) |
167 self.assertEquals(len(rset), 1) |
154 |
168 |
155 def test_sort_func(self): |
169 def test_sort_func(self): |
156 self.execute('Affaire X ORDERBY DUMB_SORT(RF) WHERE X ref RF') |
170 self.sexecute('Affaire X ORDERBY DUMB_SORT(RF) WHERE X ref RF') |
157 |
171 |
158 def test_sort_func_ambigous(self): |
172 def test_sort_func_ambigous(self): |
159 self.execute('Any X ORDERBY DUMB_SORT(RF) WHERE X title RF') |
173 self.sexecute('Any X ORDERBY DUMB_SORT(RF) WHERE X title RF') |
160 |
174 |
161 def test_in_eid(self): |
175 def test_in_eid(self): |
162 iec1 = self.repo.extid2eid(self.repo.sources_by_uri['extern'], str(ec1), |
176 iec1 = self.repo.extid2eid(self.repo.sources_by_uri['extern'], str(self.ec1), |
163 'Card', self.session) |
177 'Card', self.session) |
164 rset = self.execute('Any X WHERE X eid IN (%s, %s)' % (iec1, self.ic1)) |
178 rset = self.sexecute('Any X WHERE X eid IN (%s, %s)' % (iec1, self.ic1)) |
165 self.assertEquals(sorted(r[0] for r in rset.rows), sorted([iec1, self.ic1])) |
179 self.assertEquals(sorted(r[0] for r in rset.rows), sorted([iec1, self.ic1])) |
166 |
180 |
167 def test_greater_eid(self): |
181 def test_greater_eid(self): |
168 rset = self.execute('Any X WHERE X eid > %s' % self.maxeid) |
182 rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1)) |
169 self.assertEquals(len(rset.rows), 2) # self.ic1 and self.ic2 |
183 self.assertEquals(len(rset.rows), 2) # self.ic1 and self.ic2 |
|
184 cu = cnx2.cursor() |
170 ec2 = cu.execute('INSERT Card X: X title "glup"')[0][0] |
185 ec2 = cu.execute('INSERT Card X: X title "glup"')[0][0] |
171 cnx2.commit() |
186 cnx2.commit() |
172 # 'X eid > something' should not trigger discovery |
187 # 'X eid > something' should not trigger discovery |
173 rset = self.execute('Any X WHERE X eid > %s' % self.maxeid) |
188 rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1)) |
174 self.assertEquals(len(rset.rows), 2) |
189 self.assertEquals(len(rset.rows), 2) |
175 # trigger discovery using another query |
190 # trigger discovery using another query |
176 crset = self.execute('Card X WHERE X title "glup"') |
191 crset = self.sexecute('Card X WHERE X title "glup"') |
177 self.assertEquals(len(crset.rows), 1) |
192 self.assertEquals(len(crset.rows), 1) |
178 rset = self.execute('Any X WHERE X eid > %s' % self.maxeid) |
193 rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1)) |
179 self.assertEquals(len(rset.rows), 3) |
194 self.assertEquals(len(rset.rows), 3) |
180 rset = self.execute('Any MAX(X)') |
195 rset = self.sexecute('Any MAX(X)') |
181 self.assertEquals(len(rset.rows), 1) |
196 self.assertEquals(len(rset.rows), 1) |
182 self.assertEquals(rset.rows[0][0], crset[0][0]) |
197 self.assertEquals(rset.rows[0][0], crset[0][0]) |
183 |
198 |
184 def test_attr_unification_1(self): |
199 def test_attr_unification_1(self): |
185 n1 = self.execute('INSERT Note X: X type "AFFREF"')[0][0] |
200 n1 = self.sexecute('INSERT Note X: X type "AFFREF"')[0][0] |
186 n2 = self.execute('INSERT Note X: X type "AFFREU"')[0][0] |
201 n2 = self.sexecute('INSERT Note X: X type "AFFREU"')[0][0] |
187 rset = self.execute('Any X,Y WHERE X is Note, Y is Affaire, X type T, Y ref T') |
202 rset = self.sexecute('Any X,Y WHERE X is Note, Y is Affaire, X type T, Y ref T') |
188 self.assertEquals(len(rset), 1, rset.rows) |
203 self.assertEquals(len(rset), 1, rset.rows) |
189 |
204 |
190 def test_attr_unification_2(self): |
205 def test_attr_unification_2(self): |
|
206 cu = cnx2.cursor() |
191 ec2 = cu.execute('INSERT Card X: X title "AFFREF"')[0][0] |
207 ec2 = cu.execute('INSERT Card X: X title "AFFREF"')[0][0] |
192 cnx2.commit() |
208 cnx2.commit() |
193 try: |
209 try: |
194 c1 = self.execute('INSERT Card C: C title "AFFREF"')[0][0] |
210 c1 = self.sexecute('INSERT Card C: C title "AFFREF"')[0][0] |
195 rset = self.execute('Any X,Y WHERE X is Card, Y is Affaire, X title T, Y ref T') |
211 rset = self.sexecute('Any X,Y WHERE X is Card, Y is Affaire, X title T, Y ref T') |
196 self.assertEquals(len(rset), 2, rset.rows) |
212 self.assertEquals(len(rset), 2, rset.rows) |
197 finally: |
213 finally: |
198 cu.execute('DELETE Card X WHERE X eid %(x)s', {'x': ec2}, 'x') |
214 cu.execute('DELETE Card X WHERE X eid %(x)s', {'x': ec2}, 'x') |
199 cnx2.commit() |
215 cnx2.commit() |
200 |
216 |
201 def test_attr_unification_neq_1(self): |
217 def test_attr_unification_neq_1(self): |
202 # XXX complete |
218 # XXX complete |
203 self.execute('Any X,Y WHERE X is Note, Y is Affaire, X creation_date D, Y creation_date > D') |
219 self.sexecute('Any X,Y WHERE X is Note, Y is Affaire, X creation_date D, Y creation_date > D') |
204 |
220 |
205 def test_attr_unification_neq_2(self): |
221 def test_attr_unification_neq_2(self): |
206 # XXX complete |
222 # XXX complete |
207 self.execute('Any X,Y WHERE X is Card, Y is Affaire, X creation_date D, Y creation_date > D') |
223 self.sexecute('Any X,Y WHERE X is Card, Y is Affaire, X creation_date D, Y creation_date > D') |
208 |
224 |
209 def test_union(self): |
225 def test_union(self): |
210 afeids = self.execute('Affaire X') |
226 afeids = self.sexecute('Affaire X') |
211 ueids = self.execute('CWUser X') |
227 ueids = self.sexecute('CWUser X') |
212 rset = self.execute('(Any X WHERE X is Affaire) UNION (Any X WHERE X is CWUser)') |
228 rset = self.sexecute('(Any X WHERE X is Affaire) UNION (Any X WHERE X is CWUser)') |
213 self.assertEquals(sorted(r[0] for r in rset.rows), |
229 self.assertEquals(sorted(r[0] for r in rset.rows), |
214 sorted(r[0] for r in afeids + ueids)) |
230 sorted(r[0] for r in afeids + ueids)) |
215 |
231 |
216 def test_subquery1(self): |
232 def test_subquery1(self): |
217 rsetbase = self.execute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)') |
233 rsetbase = self.sexecute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)') |
218 self.assertEquals(len(rsetbase), 4) |
234 self.assertEquals(len(rsetbase), 4) |
219 self.assertEquals(sorted(rsetbase.rows), rsetbase.rows) |
235 self.assertEquals(sorted(rsetbase.rows), rsetbase.rows) |
220 rset = self.execute('Any W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)') |
236 rset = self.sexecute('Any W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)') |
221 self.assertEquals(rset.rows, rsetbase.rows[2:4]) |
237 self.assertEquals(rset.rows, rsetbase.rows[2:4]) |
222 rset = self.execute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X WHERE X wikiid W)') |
238 rset = self.sexecute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X WHERE X wikiid W)') |
223 self.assertEquals(rset.rows, rsetbase.rows[2:4]) |
239 self.assertEquals(rset.rows, rsetbase.rows[2:4]) |
224 rset = self.execute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W)') |
240 rset = self.sexecute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W)') |
225 self.assertEquals(rset.rows, rsetbase.rows[2:4]) |
241 self.assertEquals(rset.rows, rsetbase.rows[2:4]) |
226 |
242 |
227 def test_subquery2(self): |
243 def test_subquery2(self): |
228 affeid = self.execute('Affaire X WHERE X ref "AFFREF"')[0][0] |
244 affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0] |
229 rset = self.execute('Any X,AA,AB WITH X,AA,AB BEING (Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB)', |
245 rset = self.sexecute('Any X,AA,AB WITH X,AA,AB BEING (Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB)', |
230 {'x': affeid}) |
246 {'x': affeid}) |
231 self.assertEquals(len(rset), 1) |
247 self.assertEquals(len(rset), 1) |
232 self.assertEquals(rset[0][1], "pitetre") |
248 self.assertEquals(rset[0][1], "pitetre") |
233 |
249 |
234 def test_not_relation(self): |
250 def test_not_relation(self): |
235 states = set(tuple(x) for x in self.execute('Any S,SN WHERE S is State, S name SN')) |
251 states = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN')) |
236 self.session.user.clear_all_caches() |
252 self.session.user.clear_all_caches() |
237 userstate = self.session.user.in_state[0] |
253 userstate = self.session.user.in_state[0] |
238 states.remove((userstate.eid, userstate.name)) |
254 states.remove((userstate.eid, userstate.name)) |
239 notstates = set(tuple(x) for x in self.execute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s', |
255 notstates = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s', |
240 {'x': self.session.user.eid}, 'x')) |
256 {'x': self.session.user.eid}, 'x')) |
241 self.assertSetEquals(notstates, states) |
257 self.assertSetEquals(notstates, states) |
242 aff1 = self.execute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0] |
258 aff1 = self.sexecute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0] |
243 aff1stateeid, aff1statename = self.execute('Any S,SN WHERE X eid %(x)s, X in_state S, S name SN', {'x': aff1}, 'x')[0] |
259 aff1stateeid, aff1statename = self.sexecute('Any S,SN WHERE X eid %(x)s, X in_state S, S name SN', {'x': aff1}, 'x')[0] |
244 self.assertEquals(aff1statename, 'pitetre') |
260 self.assertEquals(aff1statename, 'pitetre') |
245 states.add((userstate.eid, userstate.name)) |
261 states.add((userstate.eid, userstate.name)) |
246 states.remove((aff1stateeid, aff1statename)) |
262 states.remove((aff1stateeid, aff1statename)) |
247 notstates = set(tuple(x) for x in self.execute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s', |
263 notstates = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s', |
248 {'x': aff1}, 'x')) |
264 {'x': aff1}, 'x')) |
249 self.assertSetEquals(notstates, states) |
265 self.assertSetEquals(notstates, states) |
250 |
266 |
251 def test_absolute_url_base_url(self): |
267 def test_absolute_url_base_url(self): |
|
268 cu = cnx2.cursor() |
252 ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0] |
269 ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0] |
253 cnx2.commit() |
270 cnx2.commit() |
254 lc = self.execute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0) |
271 lc = self.sexecute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0) |
255 self.assertEquals(lc.absolute_url(), 'http://extern.org/card/eid/%s' % ceid) |
272 self.assertEquals(lc.absolute_url(), 'http://extern.org/card/eid/%s' % ceid) |
|
273 cu.execute('DELETE Card X WHERE X eid %(x)s', {'x':ceid}) |
|
274 cnx2.commit() |
256 |
275 |
257 def test_absolute_url_no_base_url(self): |
276 def test_absolute_url_no_base_url(self): |
258 cu = cnx3.cursor() |
277 cu = cnx3.cursor() |
259 ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0] |
278 ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0] |
260 cnx3.commit() |
279 cnx3.commit() |
261 lc = self.execute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0) |
280 lc = self.sexecute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0) |
262 self.assertEquals(lc.absolute_url(), 'http://testing.fr/cubicweb/card/eid/%s' % lc.eid) |
281 self.assertEquals(lc.absolute_url(), 'http://testing.fr/cubicweb/card/eid/%s' % lc.eid) |
|
282 cu.execute('DELETE Card X WHERE X eid %(x)s', {'x':ceid}) |
|
283 cnx3.commit() |
263 |
284 |
264 def test_nonregr1(self): |
285 def test_nonregr1(self): |
265 ueid = self.session.user.eid |
286 ueid = self.session.user.eid |
266 affaire = self.execute('Affaire X WHERE X ref "AFFREF"').get_entity(0, 0) |
287 affaire = self.sexecute('Affaire X WHERE X ref "AFFREF"').get_entity(0, 0) |
267 self.execute('Any U WHERE U in_group G, (G name IN ("managers", "logilab") OR (X require_permission P?, P name "bla", P require_group G)), X eid %(x)s, U eid %(u)s', |
288 self.sexecute('Any U WHERE U in_group G, (G name IN ("managers", "logilab") OR (X require_permission P?, P name "bla", P require_group G)), X eid %(x)s, U eid %(u)s', |
268 {'x': affaire.eid, 'u': ueid}) |
289 {'x': affaire.eid, 'u': ueid}) |
269 |
290 |
270 def test_nonregr2(self): |
291 def test_nonregr2(self): |
271 self.session.user.fire_transition('deactivate') |
292 self.session.user.fire_transition('deactivate') |
272 treid = self.session.user.latest_trinfo().eid |
293 treid = self.session.user.latest_trinfo().eid |
273 rset = self.execute('Any X ORDERBY D DESC WHERE E eid %(x)s, E wf_info_for X, X modification_date D', |
294 rset = self.sexecute('Any X ORDERBY D DESC WHERE E eid %(x)s, E wf_info_for X, X modification_date D', |
274 {'x': treid}) |
295 {'x': treid}) |
275 self.assertEquals(len(rset), 1) |
296 self.assertEquals(len(rset), 1) |
276 self.assertEquals(rset.rows[0], [self.session.user.eid]) |
297 self.assertEquals(rset.rows[0], [self.session.user.eid]) |
277 |
298 |
278 def test_nonregr3(self): |
299 def test_nonregr3(self): |
279 self.execute('DELETE Card X WHERE X eid %(x)s, NOT X multisource_inlined_rel Y', {'x': self.ic1}) |
300 self.sexecute('DELETE Card X WHERE X eid %(x)s, NOT X multisource_inlined_rel Y', {'x': self.ic1}) |
280 |
301 |
281 if __name__ == '__main__': |
302 if __name__ == '__main__': |
282 from logilab.common.testlib import unittest_main |
303 from logilab.common.testlib import unittest_main |
283 unittest_main() |
304 unittest_main() |