--- a/server/test/unittest_multisources.py Sat Oct 09 00:05:50 2010 +0200
+++ b/server/test/unittest_multisources.py Sat Oct 09 00:05:52 2010 +0200
@@ -1,4 +1,4 @@
-# copyright 2003-2010 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+ # copyright 2003-2010 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
#
# This file is part of CubicWeb.
@@ -15,29 +15,29 @@
#
# 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 os.path import dirname, join, abspath
+
from datetime import datetime, timedelta
-from logilab.common.decorators import cached
-
from cubicweb.devtools import TestServerConfiguration, init_test_database
from cubicweb.devtools.testlib import CubicWebTC, refresh_repo
from cubicweb.devtools.repotest import do_monkey_patch, undo_monkey_patch
-class TwoSourcesConfiguration(TestServerConfiguration):
- sourcefile = 'sources_multi'
-
-
class ExternalSource1Configuration(TestServerConfiguration):
sourcefile = 'sources_extern'
class ExternalSource2Configuration(TestServerConfiguration):
- sourcefile = 'sources_multi2'
+ sourcefile = 'sources_multi'
MTIME = datetime.now() - timedelta(0, 10)
-repo2, cnx2 = init_test_database(config=ExternalSource1Configuration('data'))
-repo3, cnx3 = init_test_database(config=ExternalSource2Configuration('data'))
+
+EXTERN_SOURCE_CFG = u'''
+pyro-ns-id = extern
+cubicweb-user = admin
+cubicweb-password = gingkow
+mapping-file = extern_mapping.py
+base-url=http://extern.org/
+'''
# hi-jacking
from cubicweb.server.sources.pyrorql import PyroRQLSource
@@ -47,6 +47,13 @@
Connection_close = Connection.close
def setup_module(*args):
+ global repo2, cnx2, repo3, cnx3
+ repo2, cnx2 = init_test_database(config=ExternalSource1Configuration('data'))
+ repo3, cnx3 = init_test_database(config=ExternalSource2Configuration('data'))
+ cnx3.request().create_entity('CWSource', name=u'extern', type=u'pyrorql',
+ config=EXTERN_SOURCE_CFG)
+ cnx3.commit()
+
TestServerConfiguration.no_sqlite_wrap = True
# hi-jack PyroRQLSource.get_connection to access existing connection (no
# pyro connection)
@@ -67,7 +74,6 @@
TestServerConfiguration.no_sqlite_wrap = False
class TwoSourcesTC(CubicWebTC):
- config = TwoSourcesConfiguration('data')
@classmethod
def _refresh_repo(cls):
@@ -82,6 +88,8 @@
do_monkey_patch()
def tearDown(self):
+ for source in self.repo.sources[1:]:
+ self.repo.remove_source(source.uri)
CubicWebTC.tearDown(self)
undo_monkey_patch()
@@ -91,6 +99,17 @@
cu.execute('INSERT Card X: X title "C4: Ze external card", X wikiid "zzz"')
self.aff1 = cu.execute('INSERT Affaire X: X ref "AFFREF"')[0][0]
cnx2.commit()
+ for uri, config in [('extern', EXTERN_SOURCE_CFG),
+ ('extern-multi', '''
+pyro-ns-id = extern-multi
+cubicweb-user = admin
+cubicweb-password = gingkow
+mapping-file = extern_mapping.py
+''')]:
+ self.request().create_entity('CWSource', name=unicode(uri),
+ type=u'pyrorql',
+ config=unicode(config))
+ self.commit()
# trigger discovery
self.sexecute('Card X')
self.sexecute('Affaire X')
@@ -112,11 +131,11 @@
# 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': {'adapter': 'native', 'uri': 'system'},
+ {'source': {'type': 'native', 'uri': 'system'},
'type': u'Card', 'extid': None})
externent = rset.get_entity(3, 0)
metainf = externent.cw_metainformation()
- self.assertEqual(metainf['source'], {'adapter': 'pyrorql', 'base-url': 'http://extern.org/', 'uri': 'extern'})
+ self.assertEqual(metainf['source'], {'type': 'pyrorql', 'base-url': 'http://extern.org/', 'uri': 'extern'})
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',
@@ -184,7 +203,7 @@
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})
+ {'x': affeid, 'u': self.session.user.eid})
self.assertEqual(len(rset), 1)
def test_sort_func(self):
@@ -270,7 +289,6 @@
def test_not_relation(self):
states = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN'))
- self.session.user.clear_all_caches()
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',