server/test/unittest_multisources.py
author Pierre-Yves David <pierre-yves.david@logilab.fr>
Thu, 20 Jun 2013 16:00:07 +0200
changeset 9043 97c3bb9a7c99
parent 8675 b60329e40e26
permissions -rw-r--r--
[testlib] move repo and related attribute back on Instance instead of Class The repo and cnx was hold by the TestCase class to work as a cache and avoid recreation of the repo from scratch of each test. However since bad26a22fe29 the caching is done by the DatabaseHandler object and it is not necessary have a second layer of cache on the TestCase itself. We move the repo and cnx attribute back on the TestCase instance itself and make several class methods instance methods again. This will helps to change dbapi access in test from dbapi to repoapi. related to #2503918

# 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()