cubicweb: comparison server/test/unittest

equal deleted inserted replaced

-:eccd1885d42e
+:71c143c0ada3
 from cubicweb import BadConnectionId, RepositoryError, ValidationError, UnknownEid, AuthenticationError
 from cubicweb.schema import CubicWebSchema, RQLConstraint
 from cubicweb.dbapi import connect, repo_connect
 from cubicweb.devtools.apptest import RepositoryBasedTC
 from cubicweb.devtools.repotest import tuplify
 from cubicweb.server import repository
 from cubicweb.server.sqlutils import SQL_PREFIX
 # start name server anyway, process will fail if already running
 os.system('pyro-ns >/dev/null 2>/dev/null &')
 class RepositoryTC(RepositoryBasedTC):
 """ singleton providing access to a persistent storage for entities
 and relation
 """
 #     def setUp(self):
 #         pass
 #     def tearDown(self):
 #         self.repo.config.db_perms = True
 #         cnxid = self.repo.connect(*self.default_user_password())
 #         for etype in ('Affaire', 'Note', 'Societe', 'Personne'):
 #             self.repo.execute(cnxid, 'DELETE %s X' % etype)
 (u'Int',),
 (u'Interval',), (u'Password',),
 (u'String',), (u'Time',)])
 finally:
 self.repo._free_pool(pool)
 def test_schema_has_owner(self):
 repo = self.repo
 cnxid = repo.connect(*self.default_user_password())
 self.failIf(repo.execute(cnxid, 'CWEType X WHERE NOT X owned_by U'))
 self.failIf(repo.execute(cnxid, 'CWRType X WHERE NOT X owned_by U'))
 self.failIf(repo.execute(cnxid, 'CWAttribute X WHERE NOT X owned_by U'))
 self.failIf(repo.execute(cnxid, 'CWRelation X WHERE NOT X owned_by U'))
 self.failIf(repo.execute(cnxid, 'CWConstraint X WHERE NOT X owned_by U'))
 self.failIf(repo.execute(cnxid, 'CWConstraintType X WHERE NOT X owned_by U'))
 def test_connect(self):
 login, passwd = self.default_user_password()
 self.assert_(self.repo.connect(login, passwd))
 self.assertRaises(AuthenticationError,
 self.repo.connect, login, 'nimportnawak')
 self.assertRaises(AuthenticationError,
 self.repo.connect, login, None)
 self.assertRaises(AuthenticationError,
 self.repo.connect, None, None)
 def test_execute(self):
 repo = self.repo
 cnxid = repo.connect(*self.default_user_password())
 repo.execute(cnxid, 'Any X')
 repo.execute(cnxid, 'Any X where X is Personne')
 repo.execute(cnxid, 'Any X where X is Personne, X nom ~= "to"')
 repo.execute(cnxid, 'Any X WHERE X has_text %(text)s', {'text': u'\xe7a'})
 repo.close(cnxid)
 def test_login_upassword_accent(self):
 repo = self.repo
 cnxid = repo.connect(*self.default_user_password())
 repo.execute(cnxid, 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s, X in_state S, X in_group G WHERE S name "activated", G name "users"',
 {'login': u"barnab�", 'passwd': u"h�h�h�".encode('UTF8')})
 repo.commit(cnxid)
 repo.close(cnxid)
 self.assert_(repo.connect(u"barnab�", u"h�h�h�".encode('UTF8')))
 def test_invalid_entity_rollback(self):
 repo = self.repo
 cnxid = repo.connect(*self.default_user_password())
+# no group
 repo.execute(cnxid, 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s, X in_state S WHERE S name "activated"',
 {'login': u"tutetute", 'passwd': 'tutetute'})
 self.assertRaises(ValidationError, repo.commit, cnxid)
 rset = repo.execute(cnxid, 'CWUser X WHERE X login "tutetute"')
 self.assertEquals(rset.rowcount, 0)
 def test_close(self):
 repo = self.repo
 cnxid = repo.connect(*self.default_user_password())
 self.assert_(cnxid)
 repo.close(cnxid)
 self.assertRaises(BadConnectionId, repo.execute, cnxid, 'Any X')
 def test_invalid_cnxid(self):
 self.assertRaises(BadConnectionId, self.repo.execute, 0, 'Any X')
 self.assertRaises(BadConnectionId, self.repo.close, None)
 def test_shared_data(self):
 repo = self.repo
 cnxid = repo.connect(*self.default_user_password())
 repo.set_shared_data(cnxid, 'data', 4)
 cnxid2 = repo.connect(*self.default_user_password())
 result = repo.execute(cnxid, "Any U WHERE U in_group G, U login 'admin', G name 'guests'")
 self.assertEquals(result.rowcount, 1)
 repo.rollback(cnxid)
 result = repo.execute(cnxid, "Any U WHERE U in_group G, U login 'admin', G name 'guests'")
 self.assertEquals(result.rowcount, 0, result.rows)
 def test_transaction_base3(self):
 repo = self.repo
 cnxid = repo.connect(*self.default_user_password())
 # rollback state change which trigger TrInfo insertion
 ueid = repo._get_session(cnxid).user.eid
 rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': ueid})
 self.assertEquals(len(rset), 2)
 repo.rollback(cnxid)
 rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': ueid})
 self.assertEquals(len(rset), 1)
 def test_transaction_interleaved(self):
 self.skip('implement me')
 def test_initial_schema(self):
 schema = self.repo.schema
 # check order of attributes is respected
 self.assertListEquals([r.type for r in schema.eschema('CWAttribute').ordered_relations()
 if not r.type in ('eid', 'is', 'is_instance_of', 'identity',
 'creation_date', 'modification_date',
 'owned_by', 'created_by')],
 ['relation_type', 'from_entity', 'to_entity', 'constrained_by',
 'cardinality', 'ordernum',
 'indexed', 'fulltextindexed', 'internationalizable',
 'defaultval', 'description_format', 'description'])
 self.assertEquals(schema.eschema('CWEType').main_attribute(), 'name')
 self.assertEquals(schema.eschema('State').main_attribute(), 'name')
 # execute
 daemon.handleRequests(1.0)
 t.join()
 finally:
 repository.pyro_unregister(self.repo.config)
 def _pyro_client(self, lock):
 cnx = connect(self.repo.config.appid, u'admin', 'gingkow')
 # check we can get the schema
 schema = cnx.get_schema()
 self.assertEquals(schema.__hashmode__, None)
 rset = cnx.cursor().execute('Any U,G WHERE U in_group G')
 def test_internal_api(self):
 repo = self.repo
 cnxid = repo.connect(*self.default_user_password())
 session = repo._get_session(cnxid, setpool=True)
 repo.get_shared_data(cnxid, 'whatever', pop=True)
 self.assertEquals(repo.get_shared_data(cnxid, 'data'), None)
 repo.close(cnxid)
 self.assertRaises(BadConnectionId, repo.set_shared_data, cnxid, 'data', 0)
 self.assertRaises(BadConnectionId, repo.get_shared_data, cnxid, 'data')
 class DataHelpersTC(RepositoryBasedTC):
 def setUp(self):
 """ called before each test from this class """
 cnxid = self.repo.connect(*self.default_user_password())
 self.session = self.repo._sessions[cnxid]
 self.session.set_pool()
 def tearDown(self):
 self.session.rollback()
 def test_create_eid(self):
 self.assert_(self.repo.system_source.create_eid(self.session))
 def test_source_from_eid(self):
 self.assertEquals(self.repo.source_from_eid(1, self.session),
 def test_source_from_eid_raise(self):
 self.assertRaises(UnknownEid, self.repo.source_from_eid, -2, self.session)
 def test_type_from_eid(self):
 self.assertEquals(self.repo.type_from_eid(1, self.session), 'CWGroup')
 def test_type_from_eid_raise(self):
 self.assertRaises(UnknownEid, self.repo.type_from_eid, -2, self.session)
 def test_add_delete_info(self):
 entity = self.repo.vreg.etype_class('Personne')(self.session, None, None)
 entity.eid = -1
 entity.complete = lambda x: None
 self.repo.add_info(self.session, entity, self.repo.sources_by_uri['system'])
 data = cursor.fetchall()
 self.assertEquals(data, [])
 class FTITC(RepositoryBasedTC):
 def test_reindex_and_modified_since(self):
 cursor = self.session.pool['system']
 eidp = self.execute('INSERT Personne X: X nom "toto", X prenom "tutu"')[0][0]
 self.commit()
 ts = datetime.now()
 eid = self.add_entity('EmailAddress', address=u'tutu@logilab.fr').eid
 self.execute('SET X use_email Y WHERE X login "admin", Y eid %(y)s', {'y': eid})
 self.commit()
 rset = self.execute('Any X WHERE X has_text %(t)s', {'t': 'tutu'})
 self.assertEquals(rset.rows, [[self.session.user.eid]])
 class DBInitTC(RepositoryBasedTC):
 def test_versions_inserted(self):
 inserted = [r[0] for r in self.execute('Any K ORDERBY K WHERE P pkey K, P pkey ~= "system.version.%"')]
 self.assertEquals(inserted,
-[u'system.version.basket', u'system.version.comment',
+[u'system.version.basket', u'system.version.card', u'system.version.comment',
 u'system.version.cubicweb', u'system.version.email',
 u'system.version.file', u'system.version.folder',
 u'system.version.tag'])
 class InlineRelHooksTC(RepositoryBasedTC):
 """test relation hooks are called for inlined relations
 """
 def setUp(self):
 RepositoryBasedTC.setUp(self)
 self.hm = self.repo.hm
 self.called = []
 def _before_relation_hook(self, pool, fromeid, rtype, toeid):
 self.called.append((fromeid, rtype, toeid))
 def _after_relation_hook(self, pool, fromeid, rtype, toeid):
 self.called.append((fromeid, rtype, toeid))
 def test_before_add_inline_relation(self):
 """make sure before_<event>_relation hooks are called directly"""
 self.hm.register_hook(self._before_relation_hook,
 'before_add_relation', 'ecrit_par')
 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(self.called, [(eidn, 'ecrit_par', eidp)])
 def test_after_add_inline_relation(self):
 """make sure after_<event>_relation hooks are deferred"""
 self.hm.register_hook(self._after_relation_hook,
 'after_add_relation', 'ecrit_par')
 eidp = self.execute('INSERT Personne X: X nom "toto"')[0][0]
 eidn = self.execute('INSERT Note X: X type "T"')[0][0]
 self.assertEquals(self.called, [])
 self.execute('SET N ecrit_par Y WHERE N type "T", Y nom "toto"')
 self.assertEquals(self.called, [(eidn, 'ecrit_par', eidp,)])
 def test_after_add_inline(self):
 """make sure after_<event>_relation hooks are deferred"""
 self.hm.register_hook(self._after_relation_hook,
 'after_add_relation', 'in_state')
 eidp = self.execute('INSERT CWUser X: X login "toto", X upassword "tutu", X in_state S WHERE S name "activated"')[0][0]
 eids = self.execute('State X WHERE X name "activated"')[0][0]
 self.assertEquals(self.called, [(eidp, 'in_state', eids,)])
 def test_before_delete_inline_relation(self):
 """make sure before_<event>_relation hooks are called directly"""
 self.hm.register_hook(self._before_relation_hook,
 'before_delete_relation', 'ecrit_par')
 eidp = self.execute('INSERT Personne X: X nom "toto"')[0][0]
 self.execute('SET N ecrit_par Y WHERE N type "T", Y nom "toto"')
 self.assertEquals(self.called, [])
 self.execute('DELETE N ecrit_par Y WHERE N type "T", Y nom "toto"')
 self.assertEquals(self.called, [(eidn, 'ecrit_par', eidp,)])
 if __name__ == '__main__':
 unittest_main()

branch	tls-sprint
changeset 1787	71c143c0ada3
parent 1398	5fe84a5f7035
child 1880	293fe4b49e28