diff -r 58c519e5a31f -r b2530e3e0afb server/test/unittest_multisources.py --- a/server/test/unittest_multisources.py Tue Aug 11 17:04:59 2009 +0200 +++ b/server/test/unittest_multisources.py Tue Aug 11 17:13:32 2009 +0200 @@ -11,7 +11,7 @@ from logilab.common.decorators import cached from cubicweb.devtools import TestServerConfiguration, init_test_database -from cubicweb.devtools.apptest import RepositoryBasedTC +from cubicweb.devtools.testlib import CubicWebTC, refresh_repo from cubicweb.devtools.repotest import do_monkey_patch, undo_monkey_patch TestServerConfiguration.no_sqlite_wrap = True @@ -26,16 +26,9 @@ class ExternalSource2Configuration(TestServerConfiguration): sourcefile = 'sources_multi2' -repo2, cnx2 = init_test_database('sqlite', config=ExternalSource1Configuration('data')) -cu = cnx2.cursor() -ec1 = cu.execute('INSERT Card X: X title "C3: An external card", X wikiid "aaa"')[0][0] -cu.execute('INSERT Card X: X title "C4: Ze external card", X wikiid "zzz"') -aff1 = cu.execute('INSERT Affaire X: X ref "AFFREF", X in_state S WHERE S name "pitetre"')[0][0] -cnx2.commit() - MTIME = datetime.now() - timedelta(0, 10) - -repo3, cnx3 = init_test_database('sqlite', config=ExternalSource2Configuration('data')) +repo2, cnx2 = init_test_database(config=ExternalSource1Configuration('data')) +repo3, cnx3 = init_test_database(config=ExternalSource2Configuration('data')) # XXX, access existing connection, no pyro connection from cubicweb.server.sources.pyrorql import PyroRQLSource @@ -45,38 +38,47 @@ from cubicweb.dbapi import Connection Connection.close = lambda x: None -class TwoSourcesTC(RepositoryBasedTC): - repo_config = TwoSourcesConfiguration('data') +class TwoSourcesTC(CubicWebTC): + config = TwoSourcesConfiguration('data') + + @classmethod + def _refresh_repo(cls): + super(TwoSourcesTC, cls)._refresh_repo() + cnx2.rollback() + refresh_repo(repo2) + cnx3.rollback() + refresh_repo(repo3) def setUp(self): - RepositoryBasedTC.setUp(self) - self.repo.sources[-1]._query_cache.clear() - self.repo.sources[-2]._query_cache.clear() - # trigger discovery - self.execute('Card X') - self.execute('Affaire X') - self.execute('State X') - self.commit() - # don't delete external entities! - self.maxeid = self.session.system_sql('SELECT MAX(eid) FROM entities').fetchone()[0] - # add some entities - self.ic1 = self.execute('INSERT Card X: X title "C1: An internal card", X wikiid "aaai"')[0][0] - self.ic2 = self.execute('INSERT Card X: X title "C2: Ze internal card", X wikiid "zzzi"')[0][0] - self.commit() + CubicWebTC.setUp(self) do_monkey_patch() def tearDown(self): - RepositoryBasedTC.tearDown(self) + CubicWebTC.tearDown(self) undo_monkey_patch() + def setup_database(self): + cu = cnx2.cursor() + self.ec1 = cu.execute('INSERT Card X: X title "C3: An external card", X wikiid "aaa"')[0][0] + cu.execute('INSERT Card X: X title "C4: Ze external card", X wikiid "zzz"') + self.aff1 = cu.execute('INSERT Affaire X: X ref "AFFREF", X in_state S WHERE S name "pitetre"')[0][0] + cnx2.commit() + # trigger discovery + self.sexecute('Card X') + self.sexecute('Affaire X') + self.sexecute('State X') + # add some entities + self.ic1 = self.sexecute('INSERT Card X: X title "C1: An internal card", X wikiid "aaai"')[0][0] + self.ic2 = self.sexecute('INSERT Card X: X title "C2: Ze internal card", X wikiid "zzzi"')[0][0] + def test_eid_comp(self): - rset = self.execute('Card X WHERE X eid > 1') + rset = self.sexecute('Card X WHERE X eid > 1') self.assertEquals(len(rset), 4) - rset = self.execute('Any X,T WHERE X title T, X eid > 1') + rset = self.sexecute('Any X,T WHERE X title T, X eid > 1') self.assertEquals(len(rset), 4) def test_metainformation(self): - rset = self.execute('Card X ORDERBY T WHERE X title T') + rset = self.sexecute('Card X ORDERBY T WHERE X title T') # 2 added to the system source, 2 added to the external source self.assertEquals(len(rset), 4) # since they are orderd by eid, we know the 3 first one is coming from the system source @@ -89,28 +91,28 @@ self.assertEquals(metainf['source'], {'adapter': 'pyrorql', 'base-url': 'http://extern.org/', 'uri': 'extern'}) self.assertEquals(metainf['type'], 'Card') self.assert_(metainf['extid']) - etype = self.execute('Any ETN WHERE X is ET, ET name ETN, X eid %(x)s', + etype = self.sexecute('Any ETN WHERE X is ET, ET name ETN, X eid %(x)s', {'x': externent.eid}, 'x')[0][0] self.assertEquals(etype, 'Card') def test_order_limit_offset(self): - rsetbase = self.execute('Any W,X ORDERBY W,X WHERE X wikiid W') + rsetbase = self.sexecute('Any W,X ORDERBY W,X WHERE X wikiid W') self.assertEquals(len(rsetbase), 4) self.assertEquals(sorted(rsetbase.rows), rsetbase.rows) - rset = self.execute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W') + rset = self.sexecute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W') self.assertEquals(rset.rows, rsetbase.rows[2:4]) def test_has_text(self): self.repo.sources_by_uri['extern'].synchronize(MTIME) # in case fti_update has been run before - self.failUnless(self.execute('Any X WHERE X has_text "affref"')) - self.failUnless(self.execute('Affaire X WHERE X has_text "affref"')) + self.failUnless(self.sexecute('Any X WHERE X has_text "affref"')) + self.failUnless(self.sexecute('Affaire X WHERE X has_text "affref"')) def test_anon_has_text(self): self.repo.sources_by_uri['extern'].synchronize(MTIME) # in case fti_update has been run before - self.execute('INSERT Affaire X: X ref "no readable card"')[0][0] - aff1 = self.execute('INSERT Affaire X: X ref "card"')[0][0] + self.sexecute('INSERT Affaire X: X ref "no readable card"')[0][0] + aff1 = self.sexecute('INSERT Affaire X: X ref "card"')[0][0] # grant read access - self.execute('SET X owned_by U WHERE X eid %(x)s, U login "anon"', {'x': aff1}, 'x') + self.sexecute('SET X owned_by U WHERE X eid %(x)s, U login "anon"', {'x': aff1}, 'x') self.commit() cnx = self.login('anon') cu = cnx.cursor() @@ -120,79 +122,81 @@ def test_synchronization(self): cu = cnx2.cursor() - assert cu.execute('Any X WHERE X eid %(x)s', {'x': aff1}, 'x') - cu.execute('SET X ref "BLAH" WHERE X eid %(x)s', {'x': aff1}, 'x') + assert cu.execute('Any X WHERE X eid %(x)s', {'x': self.aff1}, 'x') + cu.execute('SET X ref "BLAH" WHERE X eid %(x)s', {'x': self.aff1}, 'x') aff2 = cu.execute('INSERT Affaire X: X ref "AFFREUX", X in_state S WHERE S name "pitetre"')[0][0] cnx2.commit() try: # force sync self.repo.sources_by_uri['extern'].synchronize(MTIME) - self.failUnless(self.execute('Any X WHERE X has_text "blah"')) - self.failUnless(self.execute('Any X WHERE X has_text "affreux"')) + self.failUnless(self.sexecute('Any X WHERE X has_text "blah"')) + self.failUnless(self.sexecute('Any X WHERE X has_text "affreux"')) cu.execute('DELETE Affaire X WHERE X eid %(x)s', {'x': aff2}) cnx2.commit() self.repo.sources_by_uri['extern'].synchronize(MTIME) - rset = self.execute('Any X WHERE X has_text "affreux"') + rset = self.sexecute('Any X WHERE X has_text "affreux"') self.failIf(rset) finally: # restore state - cu.execute('SET X ref "AFFREF" WHERE X eid %(x)s', {'x': aff1}, 'x') + cu.execute('SET X ref "AFFREF" WHERE X eid %(x)s', {'x': self.aff1}, 'x') cnx2.commit() def test_simplifiable_var(self): - affeid = self.execute('Affaire X WHERE X ref "AFFREF"')[0][0] - rset = self.execute('Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB', + affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0] + rset = self.sexecute('Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB', {'x': affeid}, 'x') self.assertEquals(len(rset), 1) self.assertEquals(rset[0][1], "pitetre") def test_simplifiable_var_2(self): - affeid = self.execute('Affaire X WHERE X ref "AFFREF"')[0][0] - rset = self.execute('Any E WHERE E eid %(x)s, E in_state S, NOT S name "moved"', + affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0] + rset = self.sexecute('Any E WHERE E eid %(x)s, E in_state S, NOT S name "moved"', {'x': affeid, 'u': self.session.user.eid}, 'x') self.assertEquals(len(rset), 1) def test_sort_func(self): - self.execute('Affaire X ORDERBY DUMB_SORT(RF) WHERE X ref RF') + self.sexecute('Affaire X ORDERBY DUMB_SORT(RF) WHERE X ref RF') def test_sort_func_ambigous(self): - self.execute('Any X ORDERBY DUMB_SORT(RF) WHERE X title RF') + self.sexecute('Any X ORDERBY DUMB_SORT(RF) WHERE X title RF') def test_in_eid(self): - iec1 = self.repo.extid2eid(self.repo.sources_by_uri['extern'], str(ec1), + iec1 = self.repo.extid2eid(self.repo.sources_by_uri['extern'], str(self.ec1), 'Card', self.session) - rset = self.execute('Any X WHERE X eid IN (%s, %s)' % (iec1, self.ic1)) + rset = self.sexecute('Any X WHERE X eid IN (%s, %s)' % (iec1, self.ic1)) self.assertEquals(sorted(r[0] for r in rset.rows), sorted([iec1, self.ic1])) def test_greater_eid(self): - rset = self.execute('Any X WHERE X eid > %s' % self.maxeid) + rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1)) self.assertEquals(len(rset.rows), 2) # self.ic1 and self.ic2 + cu = cnx2.cursor() ec2 = cu.execute('INSERT Card X: X title "glup"')[0][0] cnx2.commit() # 'X eid > something' should not trigger discovery - rset = self.execute('Any X WHERE X eid > %s' % self.maxeid) + rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1)) self.assertEquals(len(rset.rows), 2) # trigger discovery using another query - crset = self.execute('Card X WHERE X title "glup"') + crset = self.sexecute('Card X WHERE X title "glup"') self.assertEquals(len(crset.rows), 1) - rset = self.execute('Any X WHERE X eid > %s' % self.maxeid) + rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1)) self.assertEquals(len(rset.rows), 3) - rset = self.execute('Any MAX(X)') + rset = self.sexecute('Any MAX(X)') self.assertEquals(len(rset.rows), 1) self.assertEquals(rset.rows[0][0], crset[0][0]) def test_attr_unification_1(self): - n1 = self.execute('INSERT Note X: X type "AFFREF"')[0][0] - n2 = self.execute('INSERT Note X: X type "AFFREU"')[0][0] - rset = self.execute('Any X,Y WHERE X is Note, Y is Affaire, X type T, Y ref T') + n1 = self.sexecute('INSERT Note X: X type "AFFREF"')[0][0] + n2 = self.sexecute('INSERT Note X: X type "AFFREU"')[0][0] + rset = self.sexecute('Any X,Y WHERE X is Note, Y is Affaire, X type T, Y ref T') self.assertEquals(len(rset), 1, rset.rows) def test_attr_unification_2(self): + cu = cnx2.cursor() ec2 = cu.execute('INSERT Card X: X title "AFFREF"')[0][0] cnx2.commit() try: - c1 = self.execute('INSERT Card C: C title "AFFREF"')[0][0] - rset = self.execute('Any X,Y WHERE X is Card, Y is Affaire, X title T, Y ref T') + c1 = self.sexecute('INSERT Card C: C title "AFFREF"')[0][0] + rset = self.sexecute('Any X,Y WHERE X is Card, Y is Affaire, X title T, Y ref T') self.assertEquals(len(rset), 2, rset.rows) finally: cu.execute('DELETE Card X WHERE X eid %(x)s', {'x': ec2}, 'x') @@ -200,81 +204,86 @@ def test_attr_unification_neq_1(self): # XXX complete - self.execute('Any X,Y WHERE X is Note, Y is Affaire, X creation_date D, Y creation_date > D') + self.sexecute('Any X,Y WHERE X is Note, Y is Affaire, X creation_date D, Y creation_date > D') def test_attr_unification_neq_2(self): # XXX complete - self.execute('Any X,Y WHERE X is Card, Y is Affaire, X creation_date D, Y creation_date > D') + self.sexecute('Any X,Y WHERE X is Card, Y is Affaire, X creation_date D, Y creation_date > D') def test_union(self): - afeids = self.execute('Affaire X') - ueids = self.execute('CWUser X') - rset = self.execute('(Any X WHERE X is Affaire) UNION (Any X WHERE X is CWUser)') + afeids = self.sexecute('Affaire X') + ueids = self.sexecute('CWUser X') + rset = self.sexecute('(Any X WHERE X is Affaire) UNION (Any X WHERE X is CWUser)') self.assertEquals(sorted(r[0] for r in rset.rows), sorted(r[0] for r in afeids + ueids)) def test_subquery1(self): - rsetbase = self.execute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)') + rsetbase = self.sexecute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)') self.assertEquals(len(rsetbase), 4) self.assertEquals(sorted(rsetbase.rows), rsetbase.rows) - rset = self.execute('Any W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)') + rset = self.sexecute('Any W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)') self.assertEquals(rset.rows, rsetbase.rows[2:4]) - rset = self.execute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X WHERE X wikiid W)') + rset = self.sexecute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X WHERE X wikiid W)') self.assertEquals(rset.rows, rsetbase.rows[2:4]) - rset = self.execute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W)') + rset = self.sexecute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W)') self.assertEquals(rset.rows, rsetbase.rows[2:4]) def test_subquery2(self): - affeid = self.execute('Affaire X WHERE X ref "AFFREF"')[0][0] - 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)', + affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0] + 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)', {'x': affeid}) self.assertEquals(len(rset), 1) self.assertEquals(rset[0][1], "pitetre") def test_not_relation(self): - states = set(tuple(x) for x in self.execute('Any S,SN WHERE S is State, S name SN')) + states = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN')) userstate = self.session.user.in_state[0] states.remove((userstate.eid, userstate.name)) - 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', + 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', {'x': self.session.user.eid}, 'x')) self.assertEquals(notstates, states) - aff1 = self.execute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0] - aff1stateeid, aff1statename = self.execute('Any S,SN WHERE X eid %(x)s, X in_state S, S name SN', {'x': aff1}, 'x')[0] + aff1 = self.sexecute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0] + aff1stateeid, aff1statename = self.sexecute('Any S,SN WHERE X eid %(x)s, X in_state S, S name SN', {'x': aff1}, 'x')[0] self.assertEquals(aff1statename, 'pitetre') states.add((userstate.eid, userstate.name)) states.remove((aff1stateeid, aff1statename)) - 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', + 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', {'x': aff1}, 'x')) self.assertSetEquals(notstates, states) def test_absolute_url_base_url(self): + cu = cnx2.cursor() ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0] cnx2.commit() - lc = self.execute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0) + lc = self.sexecute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0) self.assertEquals(lc.absolute_url(), 'http://extern.org/card/eid/%s' % ceid) + cu.execute('DELETE Card X WHERE X eid %(x)s', {'x':ceid}) + cnx2.commit() def test_absolute_url_no_base_url(self): cu = cnx3.cursor() ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0] cnx3.commit() - lc = self.execute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0) + lc = self.sexecute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0) self.assertEquals(lc.absolute_url(), 'http://testing.fr/cubicweb/card/eid/%s' % lc.eid) + cu.execute('DELETE Card X WHERE X eid %(x)s', {'x':ceid}) + cnx3.commit() def test_nonregr1(self): ueid = self.session.user.eid - affaire = self.execute('Affaire X WHERE X ref "AFFREF"').get_entity(0, 0) - 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', + affaire = self.sexecute('Affaire X WHERE X ref "AFFREF"').get_entity(0, 0) + 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', {'x': affaire.eid, 'u': ueid}) def test_nonregr2(self): treid = self.session.user.latest_trinfo().eid - rset = self.execute('Any X ORDERBY D DESC WHERE E eid %(x)s, E wf_info_for X, X modification_date D', + rset = self.sexecute('Any X ORDERBY D DESC WHERE E eid %(x)s, E wf_info_for X, X modification_date D', {'x': treid}) self.assertEquals(len(rset), 1) self.assertEquals(rset.rows[0], [self.session.user.eid]) def test_nonregr3(self): - self.execute('DELETE Card X WHERE X eid %(x)s, NOT X multisource_inlined_rel Y', {'x': self.ic1}) + self.sexecute('DELETE Card X WHERE X eid %(x)s, NOT X multisource_inlined_rel Y', {'x': self.ic1}) if __name__ == '__main__': from logilab.common.testlib import unittest_main