+
−# copyright 2003-2012 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+
−# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
+
−#
+
−# This file is part of CubicWeb.
+
−#
+
−# CubicWeb is free software: you can redistribute it and/or modify it under the
+
−# terms of the GNU Lesser General Public License as published by the Free
+
−# Software Foundation, either version 2.1 of the License, or (at your option)
+
−# any later version.
+
−#
+
−# CubicWeb is distributed in the hope that it will be useful, but WITHOUT
+
−# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+
−# FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+
−# details.
+
−#
+
−# You should have received a copy of the GNU Lesser General Public License along
+
−# with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
+
−
+
−from datetime import datetime, timedelta
+
−from itertools import repeat
+
−
+
−from cubicweb.devtools import TestServerConfiguration, init_test_database
+
−from cubicweb.devtools.testlib import CubicWebTC, Tags
+
−from cubicweb.devtools.repotest import do_monkey_patch, undo_monkey_patch
+
−from cubicweb.devtools import get_test_db_handler
+
−
+
−class ExternalSource1Configuration(TestServerConfiguration):
+
−    sourcefile = 'sources_extern'
+
−
+
−class ExternalSource2Configuration(TestServerConfiguration):
+
−    sourcefile = 'sources_multi'
+
−
+
−MTIME = datetime.utcnow() - timedelta(0, 10)
+
−
+
−EXTERN_SOURCE_CFG = u'''
+
−cubicweb-user = admin
+
−cubicweb-password = gingkow
+
−base-url=http://extern.org/
+
−'''
+
−
+
−# hi-jacking
+
−from cubicweb.server.sources.pyrorql import PyroRQLSource
+
−from cubicweb.dbapi import Connection
+
−
+
−PyroRQLSource_get_connection = PyroRQLSource.get_connection
+
−Connection_close = Connection.close
+
−
+
−def add_extern_mapping(source):
+
−    source.init_mapping(zip(('Card', 'Affaire', 'State',
+
−                             'in_state', 'documented_by', 'multisource_inlined_rel'),
+
−                            repeat(u'write')))
+
−
+
−
+
−def pre_setup_database_extern(session, config):
+
−    session.execute('INSERT Card X: X title "C3: An external card", X wikiid "aaa"')
+
−    session.execute('INSERT Card X: X title "C4: Ze external card", X wikiid "zzz"')
+
−    session.execute('INSERT Affaire X: X ref "AFFREF"')
+
−    session.commit()
+
−
+
−def pre_setup_database_multi(session, config):
+
−    session.create_entity('CWSource', name=u'extern', type=u'pyrorql',
+
−                          url=u'pyro:///extern', config=EXTERN_SOURCE_CFG)
+
−    session.commit()
+
−
+
−
+
−class TwoSourcesTC(CubicWebTC):
+
−    """Main repo -> extern-multi -> extern
+
−                  \-------------/
+
−    """
+
−    test_db_id= 'cw-server-multisources'
+
−    tags = CubicWebTC.tags | Tags(('multisources'))
+
−
+
−
+
−    def _init_repo(self):
+
−        repo2_handler = get_test_db_handler(self._cfg2)
+
−        repo2_handler.build_db_cache('4cards-1affaire',pre_setup_func=pre_setup_database_extern)
+
−        self.repo2, self.cnx2 = repo2_handler.get_repo_and_cnx('4cards-1affaire')
+
−
+
−        repo3_handler = get_test_db_handler(self._cfg3)
+
−        repo3_handler.build_db_cache('multisource',pre_setup_func=pre_setup_database_multi)
+
−        self.repo3, self.cnx3 = repo3_handler.get_repo_and_cnx('multisource')
+
−
+
−
+
−        super(TwoSourcesTC, self)._init_repo()
+
−
+
−    def setUp(self):
+
−        self._cfg2 = ExternalSource1Configuration('data', apphome=TwoSourcesTC.datadir)
+
−        self._cfg3 = ExternalSource2Configuration('data', apphome=TwoSourcesTC.datadir)
+
−        TestServerConfiguration.no_sqlite_wrap = True
+
−        # hi-jack PyroRQLSource.get_connection to access existing connection (no
+
−        # pyro connection)
+
−        PyroRQLSource.get_connection = lambda x: x.uri == 'extern-multi' and self.cnx3 or self.cnx2
+
−        # also necessary since the repository is closing its initial connections
+
−        # pool though we want to keep cnx2 valid
+
−        Connection.close = lambda x: None
+
−        CubicWebTC.setUp(self)
+
−        self.addCleanup(self.cnx2.close)
+
−        self.addCleanup(self.cnx3.close)
+
−        do_monkey_patch()
+
−
+
−    def tearDown(self):
+
−        for source in self.repo.sources[1:]:
+
−            self.repo.remove_source(source.uri)
+
−        super(TwoSourcesTC, self).tearDown()
+
−        PyroRQLSource.get_connection = PyroRQLSource_get_connection
+
−        Connection.close = Connection_close
+
−        TestServerConfiguration.no_sqlite_wrap = False
+
−        undo_monkey_patch()
+
−
+
−    @staticmethod
+
−    def pre_setup_database(session, config):
+
−        for uri, src_config in [('extern', EXTERN_SOURCE_CFG),
+
−                            ('extern-multi', '''
+
−cubicweb-user = admin
+
−cubicweb-password = gingkow
+
−''')]:
+
−            source = session.create_entity('CWSource', name=unicode(uri),
+
−                                           type=u'pyrorql', url=u'pyro:///extern-multi',
+
−                                           config=unicode(src_config))
+
−            session.commit()
+
−            add_extern_mapping(source)
+
−
+
−        session.commit()
+
−        # trigger discovery
+
−        session.execute('Card X')
+
−        session.execute('Affaire X')
+
−        session.execute('State X')
+
−
+
−    def setup_database(self):
+
−        cu2 = self.cnx2.cursor()
+
−        self.ec1 = cu2.execute('Any X WHERE X is Card, X title "C3: An external card", X wikiid "aaa"')[0][0]
+
−        self.aff1 = cu2.execute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0]
+
−        cu2.close()
+
−        # 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.sexecute('Card X WHERE X eid > 1')
+
−        self.assertEqual(len(rset), 4)
+
−        rset = self.sexecute('Any X,T WHERE X title T, X eid > 1')
+
−        self.assertEqual(len(rset), 4)
+
−
+
−    def test_metainformation(self):
+
−        rset = self.sexecute('Card X ORDERBY T WHERE X title T')
+
−        # 2 added to the system source, 2 added to the external source
+
−        self.assertEqual(len(rset), 4)
+
−        # since they are orderd by eid, we know the 3 first one is coming from the system source
+
−        # and the others from external source
+
−        self.assertEqual(rset.get_entity(0, 0).cw_metainformation(),
+
−                          {'source': {'type': 'native', 'uri': 'system', 'use-cwuri-as-url': False},
+
−                           'type': u'Card', 'extid': None})
+
−        externent = rset.get_entity(3, 0)
+
−        metainf = externent.cw_metainformation()
+
−        self.assertEqual(metainf['source'], {'type': 'pyrorql', 'base-url': 'http://extern.org/', 'uri': 'extern', 'use-cwuri-as-url': False})
+
−        self.assertEqual(metainf['type'], 'Card')
+
−        self.assert_(metainf['extid'])
+
−        etype = self.sexecute('Any ETN WHERE X is ET, ET name ETN, X eid %(x)s',
+
−                             {'x': externent.eid})[0][0]
+
−        self.assertEqual(etype, 'Card')
+
−
+
−    def test_order_limit_offset(self):
+
−        rsetbase = self.sexecute('Any W,X ORDERBY W,X WHERE X wikiid W')
+
−        self.assertEqual(len(rsetbase), 4)
+
−        self.assertEqual(sorted(rsetbase.rows), rsetbase.rows)
+
−        rset = self.sexecute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W')
+
−        self.assertEqual(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.assertTrue(self.sexecute('Any X WHERE X has_text "affref"'))
+
−        self.assertTrue(self.sexecute('Affaire X WHERE X has_text "affref"'))
+
−        self.assertTrue(self.sexecute('Any X ORDERBY FTIRANK(X) WHERE X has_text "affref"'))
+
−        self.assertTrue(self.sexecute('Affaire X ORDERBY FTIRANK(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.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.sexecute('SET X owned_by U WHERE X eid %(x)s, U login "anon"', {'x': aff1})
+
−        self.commit()
+
−        cnx = self.login('anon')
+
−        cu = cnx.cursor()
+
−        rset = cu.execute('Any X WHERE X has_text "card"')
+
−        # 5: 4 card + 1 readable affaire
+
−        self.assertEqual(len(rset), 5, zip(rset.rows, rset.description))
+
−        rset = cu.execute('Any X ORDERBY FTIRANK(X) WHERE X has_text "card"')
+
−        self.assertEqual(len(rset), 5, zip(rset.rows, rset.description))
+
−        Connection_close(cnx.cnx) # cnx is a TestCaseConnectionProxy
+
−
+
−    def test_synchronization(self):
+
−        cu = self.cnx2.cursor()
+
−        assert cu.execute('Any X WHERE X eid %(x)s', {'x': self.aff1})
+
−        cu.execute('SET X ref "BLAH" WHERE X eid %(x)s', {'x': self.aff1})
+
−        aff2 = cu.execute('INSERT Affaire X: X ref "AFFREUX"')[0][0]
+
−        self.cnx2.commit()
+
−        try:
+
−            # force sync
+
−            self.repo.sources_by_uri['extern'].synchronize(MTIME)
+
−            self.assertTrue(self.sexecute('Any X WHERE X has_text "blah"'))
+
−            self.assertTrue(self.sexecute('Any X WHERE X has_text "affreux"'))
+
−            cu.execute('DELETE Affaire X WHERE X eid %(x)s', {'x': aff2})
+
−            self.cnx2.commit()
+
−            self.repo.sources_by_uri['extern'].synchronize(MTIME)
+
−            rset = self.sexecute('Any X WHERE X has_text "affreux"')
+
−            self.assertFalse(rset)
+
−        finally:
+
−            # restore state
+
−            cu.execute('SET X ref "AFFREF" WHERE X eid %(x)s', {'x': self.aff1})
+
−            self.cnx2.commit()
+
−
+
−    def test_simplifiable_var(self):
+
−        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})
+
−        self.assertEqual(len(rset), 1)
+
−        self.assertEqual(rset[0][1], "pitetre")
+
−
+
−    def test_simplifiable_var_2(self):
+
−        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})
+
−        self.assertEqual(len(rset), 1)
+
−
+
−    def test_sort_func(self):
+
−        self.sexecute('Affaire X ORDERBY DUMB_SORT(RF) WHERE X ref RF')
+
−
+
−    def test_sort_func_ambigous(self):
+
−        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(self.ec1),
+
−                                   'Card', self.session)
+
−        rset = self.sexecute('Any X WHERE X eid IN (%s, %s)' % (iec1, self.ic1))
+
−        self.assertEqual(sorted(r[0] for r in rset.rows), sorted([iec1, self.ic1]))
+
−
+
−    def test_greater_eid(self):
+
−        rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1))
+
−        self.assertEqual(len(rset.rows), 2) # self.ic1 and self.ic2
+
−        cu = self.cnx2.cursor()
+
−        ec2 = cu.execute('INSERT Card X: X title "glup"')[0][0]
+
−        self.cnx2.commit()
+
−        # 'X eid > something' should not trigger discovery
+
−        rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1))
+
−        self.assertEqual(len(rset.rows), 2)
+
−        # trigger discovery using another query
+
−        crset = self.sexecute('Card X WHERE X title "glup"')
+
−        self.assertEqual(len(crset.rows), 1)
+
−        rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1))
+
−        self.assertEqual(len(rset.rows), 3)
+
−        rset = self.sexecute('Any MAX(X)')
+
−        self.assertEqual(len(rset.rows), 1)
+
−        self.assertEqual(rset.rows[0][0], crset[0][0])
+
−
+
−    def test_attr_unification_1(self):
+
−        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.assertEqual(len(rset), 1, rset.rows)
+
−
+
−    def test_attr_unification_2(self):
+
−        cu = self.cnx2.cursor()
+
−        ec2 = cu.execute('INSERT Card X: X title "AFFREF"')[0][0]
+
−        self.cnx2.commit()
+
−        try:
+
−            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.assertEqual(len(rset), 2, rset.rows)
+
−        finally:
+
−            cu.execute('DELETE Card X WHERE X eid %(x)s', {'x': ec2})
+
−            self.cnx2.commit()
+
−
+
−    def test_attr_unification_neq_1(self):
+
−        # XXX complete
+
−        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.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.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.assertEqual(sorted(r[0] for r in rset.rows),
+
−                          sorted(r[0] for r in afeids + ueids))
+
−
+
−    def test_subquery1(self):
+
−        rsetbase = self.sexecute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)')
+
−        self.assertEqual(len(rsetbase), 4)
+
−        self.assertEqual(sorted(rsetbase.rows), rsetbase.rows)
+
−        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.assertEqual(rset.rows, rsetbase.rows[2:4])
+
−        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.assertEqual(rset.rows, rsetbase.rows[2:4])
+
−        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.assertEqual(rset.rows, rsetbase.rows[2:4])
+
−
+
−    def test_subquery2(self):
+
−        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.assertEqual(len(rset), 1)
+
−        self.assertEqual(rset[0][1], "pitetre")
+
−
+
−    def test_not_relation(self):
+
−        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.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}))
+
−        self.assertSetEqual(notstates, states)
+
−        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})[0]
+
−        self.assertEqual(aff1statename, 'pitetre')
+
−        states.add((userstate.eid, userstate.name))
+
−        states.remove((aff1stateeid, aff1statename))
+
−        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}))
+
−        self.assertSetEqual(notstates, states)
+
−
+
−    def test_absolute_url_base_url(self):
+
−        cu = self.cnx2.cursor()
+
−        ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0]
+
−        self.cnx2.commit()
+
−        lc = self.sexecute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0)
+
−        self.assertEqual(lc.absolute_url(), 'http://extern.org/%s' % ceid)
+
−        cu.execute('DELETE Card X WHERE X eid %(x)s', {'x':ceid})
+
−        self.cnx2.commit()
+
−
+
−    def test_absolute_url_no_base_url(self):
+
−        cu = self.cnx3.cursor()
+
−        ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0]
+
−        self.cnx3.commit()
+
−        lc = self.sexecute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0)
+
−        self.assertEqual(lc.absolute_url(), 'http://testing.fr/cubicweb/%s' % lc.eid)
+
−        cu.execute('DELETE Card X WHERE X eid %(x)s', {'x':ceid})
+
−        self.cnx3.commit()
+
−
+
−    def test_crossed_relation_noeid_needattr(self):
+
−        """http://www.cubicweb.org/ticket/1382452"""
+
−        aff1 = self.sexecute('INSERT Affaire X: X ref "AFFREF"')[0][0]
+
−        # link within extern source
+
−        ec1 = self.sexecute('Card X WHERE X wikiid "zzz"')[0][0]
+
−        self.sexecute('SET A documented_by C WHERE E eid %(a)s, C eid %(c)s',
+
−                      {'a': aff1, 'c': ec1})
+
−        # link from system to extern source
+
−        self.sexecute('SET A documented_by C WHERE E eid %(a)s, C eid %(c)s',
+
−                      {'a': aff1, 'c': self.ic2})
+
−        rset = self.sexecute('DISTINCT Any DEP WHERE P ref "AFFREF", P documented_by DEP, DEP wikiid LIKE "z%"')
+
−        self.assertEqual(sorted(rset.rows), [[ec1], [self.ic2]])
+
−
+
−    def test_nonregr1(self):
+
−        ueid = self.session.user.eid
+
−        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):
+
−        iworkflowable = self.session.user.cw_adapt_to('IWorkflowable')
+
−        iworkflowable.fire_transition('deactivate')
+
−        treid = iworkflowable.latest_trinfo().eid
+
−        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.assertEqual(len(rset), 1)
+
−        self.assertEqual(rset.rows[0], [self.session.user.eid])
+
−
+
−    def test_nonregr3(self):
+
−        self.sexecute('DELETE Card X WHERE X eid %(x)s, NOT X multisource_inlined_rel Y', {'x': self.ic1})
+
−
+
−    def test_nonregr4(self):
+
−        self.sexecute('Any X,S,U WHERE X in_state S, X todo_by U')
+
−
+
−    def test_delete_source(self):
+
−        req = self.request()
+
−        req.execute('DELETE CWSource S WHERE S name "extern"')
+
−        self.commit()
+
−        cu = self.session.system_sql("SELECT * FROM entities WHERE source='extern'")
+
−        self.assertFalse(cu.fetchall())
+
−
+
−if __name__ == '__main__':
+
−    from logilab.common.testlib import unittest_main
+
−    unittest_main()