cubicweb: comparison server/test/unittest

equal deleted inserted replaced

-:d6fd82a5a4e8
+:a63d7886fcf5
 :organization: Logilab
 :copyright: 2001-2010 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2.
 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
 :license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses
 """
+from __future__ import with_statement
+from __future__ import with_statement
 import os
 import sys
 import threading
 import time
 from yams.constraints import UniqueConstraint
 from cubicweb import (BadConnectionId, RepositoryError, ValidationError,
 UnknownEid, AuthenticationError)
+from cubicweb.selectors import implements
 from cubicweb.schema import CubicWebSchema, RQLConstraint
 from cubicweb.dbapi import connect, multiple_connections_unfix
 from cubicweb.devtools.testlib import CubicWebTC
 from cubicweb.devtools.repotest import tuplify
 from cubicweb.server import repository, hook
 from cubicweb.server.sqlutils import SQL_PREFIX
+from cubicweb.server.hook import Hook
 from cubicweb.server.sources import native
 # start name server anyway, process will fail if already running
 os.system('pyro-ns >/dev/null 2>/dev/null &')
 finally:
 repository.pyro_unregister(self.repo.config)
 from logilab.common import pyro_ext
 pyro_ext._DAEMONS.clear()
 def _pyro_client(self, done):
-cnx = connect(self.repo.config.appid, u'admin', password='gingkow')
+cnx = connect(self.repo.config.appid, u'admin', password='gingkow',
+initlog=False) # don't reset logging configuration
 try:
 # check we can get the schema
 schema = cnx.get_schema()
 self.assertEquals(schema.__hashmode__, None)
 cu = cnx.cursor()
 rset = cu.execute('Any U,G WHERE U in_group G')
 cnx.close()
 done.append(True)
 finally:
-# connect monkey path some method by default, remove them
+# connect monkey patch some method by default, remove them
 multiple_connections_unfix()
 def test_internal_api(self):
 repo = self.repo
 cnxid = repo.connect(self.admlogin, password=self.admpassword)
 {'x': note.eid})
 self.assertEquals(len(rset), 1)
 self.assertEquals(rset.rows[0][0], p2.eid)
+def test_set_attributes_in_before_update(self):
+# local hook
+class DummyBeforeHook(Hook):
+__regid__ = 'dummy-before-hook'
+__select__ = Hook.__select__ & implements('EmailAddress')
+events = ('before_update_entity',)
+def __call__(self):
+# safety belt: avoid potential infinite recursion if the test
+#              fails (i.e. RuntimeError not raised)
+pendings = self._cw.transaction_data.setdefault('pending', set())
+if self.entity.eid not in pendings:
+pendings.add(self.entity.eid)
+self.entity.set_attributes(alias=u'foo')
+with self.temporary_appobjects(DummyBeforeHook):
+req = self.request()
+addr = req.create_entity('EmailAddress', address=u'a@b.fr')
+addr.set_attributes(address=u'a@b.com')
+rset = self.execute('Any A,AA WHERE X eid %(x)s, X address A, X alias AA',
+{'x': addr.eid})
+self.assertEquals(rset.rows, [[u'a@b.com', u'foo']])
+def test_set_attributes_in_before_add(self):
+# local hook
+class DummyBeforeHook(Hook):
+__regid__ = 'dummy-before-hook'
+__select__ = Hook.__select__ & implements('EmailAddress')
+events = ('before_add_entity',)
+def __call__(self):
+# set_attributes is forbidden within before_add_entity()
+self.entity.set_attributes(alias=u'foo')
+with self.temporary_appobjects(DummyBeforeHook):
+req = self.request()
+self.assertRaises(RepositoryError, req.create_entity,
+'EmailAddress', address=u'a@b.fr')
 class DataHelpersTC(CubicWebTC):
 def test_create_eid(self):
 self.session.set_pool()
 self.assert_(self.repo.system_source.create_eid(self.session))
 class FTITC(CubicWebTC):
 def test_reindex_and_modified_since(self):
+self.repo.system_source.multisources_etypes.add('Personne')
 eidp = self.execute('INSERT Personne X: X nom "toto", X prenom "tutu"')[0][0]
 self.commit()
 ts = datetime.now()
 self.assertEquals(len(self.execute('Personne X WHERE X has_text "tutu"')), 1)
 self.session.set_pool()
 class InlineRelHooksTC(CubicWebTC):
 """test relation hooks are called for inlined relations
 """
 def setUp(self):
 CubicWebTC.setUp(self)
-self.hm = self.repo.hm
 CALLED[:] = ()
 def _after_relation_hook(self, pool, fromeid, rtype, toeid):
 self.called.append((fromeid, rtype, toeid))
 events = ('before_add_relation', 'after_add_relation',
 'before_delete_relation', 'after_delete_relation')
 def __call__(self):
 CALLED.append((self.event, self.eidfrom, self.rtype, self.eidto))
-self.hm.register(EcritParHook)
+with self.temporary_appobjects(EcritParHook):
 eidp = self.execute('INSERT Personne X: X nom "toto"')[0][0]
 eidn = self.execute('INSERT Note X: X type "T"')[0][0]
 self.execute('SET N ecrit_par Y WHERE N type "T", Y nom "toto"')
 self.assertEquals(CALLED, [('before_add_relation', eidn, 'ecrit_par', eidp),
 ('after_add_relation', eidn, 'ecrit_par', eidp)])
 CALLED[:] = ()
 self.execute('DELETE N ecrit_par Y WHERE N type "T", Y nom "toto"')
 self.assertEquals(CALLED, [('before_delete_relation', eidn, 'ecrit_par', eidp),
 ('after_delete_relation', eidn, 'ecrit_par', eidp)])
 CALLED[:] = ()
 eidn = self.execute('INSERT Note N: N ecrit_par P WHERE P nom "toto"')[0][0]
 self.assertEquals(CALLED, [('before_add_relation', eidn, 'ecrit_par', eidp),
 ('after_add_relation', eidn, 'ecrit_par', eidp)])
 if __name__ == '__main__':
 unittest_main()

changeset 5121	a63d7886fcf5
parent 5115	2e43ef618d14
child 5126	c869d8e7d166