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