# HG changeset patch # User Aurelien Campeas # Date 1401886507 -7200 # Node ID 4c1969c173fa9834b69aafcff03b33af07609f79 # Parent 42a874c666b1ac469978ee178ac069c478c9d61e [hookstests/base] use the new connection api diff -r 42a874c666b1 -r 4c1969c173fa hooks/test/unittest_hooks.py --- a/hooks/test/unittest_hooks.py Wed Jun 04 14:54:53 2014 +0200 +++ b/hooks/test/unittest_hooks.py Wed Jun 04 14:55:07 2014 +0200 @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -# copyright 2003-2013 LOGILAB S.A. (Paris, FRANCE), all rights reserved. +# copyright 2003-2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved. # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr # # This file is part of CubicWeb. @@ -30,177 +30,189 @@ class CoreHooksTC(CubicWebTC): def test_inlined(self): - self.assertEqual(self.repo.schema['sender'].inlined, True) - self.execute('INSERT EmailAddress X: X address "toto@logilab.fr", X alias "hop"') - self.execute('INSERT EmailPart X: X content_format "text/plain", X ordernum 1, X content "this is a test"') - eeid = self.execute('INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y, X parts P ' - 'WHERE Y is EmailAddress, P is EmailPart')[0][0] - self.execute('SET X sender Y WHERE X is Email, Y is EmailAddress') - rset = self.execute('Any S WHERE X sender S, X eid %s' % eeid) - self.assertEqual(len(rset), 1) + with self.admin_access.repo_cnx() as cnx: + self.assertEqual(self.repo.schema['sender'].inlined, True) + cnx.execute('INSERT EmailAddress X: X address "toto@logilab.fr", X alias "hop"') + cnx.execute('INSERT EmailPart X: X content_format "text/plain", X ordernum 1, ' + 'X content "this is a test"') + eeid = cnx.execute('INSERT Email X: X messageid "<1234>", X subject "test", ' + 'X sender Y, X recipients Y, X parts P ' + 'WHERE Y is EmailAddress, P is EmailPart')[0][0] + cnx.execute('SET X sender Y WHERE X is Email, Y is EmailAddress') + rset = cnx.execute('Any S WHERE X sender S, X eid %s' % eeid) + self.assertEqual(len(rset), 1) def test_symmetric(self): - req = self.request() - u1 = self.create_user(req, u'1') - u2 = self.create_user(req, u'2') - u3 = self.create_user(req, u'3') - ga = req.create_entity('CWGroup', name=u'A') - gb = req.create_entity('CWGroup', name=u'B') - u1.cw_set(friend=u2) - u2.cw_set(friend=u3) - ga.cw_set(friend=gb) - ga.cw_set(friend=u1) - self.commit() - req = self.request() - for l1, l2 in ((u'1', u'2'), - (u'2', u'3')): - self.assertTrue(req.execute('Any U1,U2 WHERE U1 friend U2, U1 login %(l1)s, U2 login %(l2)s', - {'l1': l1, 'l2': l2})) - self.assertTrue(req.execute('Any U1,U2 WHERE U2 friend U1, U1 login %(l1)s, U2 login %(l2)s', - {'l1': l1, 'l2': l2})) - self.assertTrue(req.execute('Any GA,GB WHERE GA friend GB, GA name "A", GB name "B"')) - self.assertTrue(req.execute('Any GA,GB WHERE GB friend GA, GA name "A", GB name "B"')) - self.assertTrue(req.execute('Any GA,U1 WHERE GA friend U1, GA name "A", U1 login "1"')) - self.assertTrue(req.execute('Any GA,U1 WHERE U1 friend GA, GA name "A", U1 login "1"')) - self.assertFalse(req.execute('Any GA,U WHERE GA friend U, GA name "A", U login "2"')) - for l1, l2 in ((u'1', u'3'), - (u'3', u'1')): - self.assertFalse(req.execute('Any U1,U2 WHERE U1 friend U2, U1 login %(l1)s, U2 login %(l2)s', - {'l1': l1, 'l2': l2})) - self.assertFalse(req.execute('Any U1,U2 WHERE U2 friend U1, U1 login %(l1)s, U2 login %(l2)s', - {'l1': l1, 'l2': l2})) + with self.admin_access.repo_cnx() as cnx: + u1 = self.create_user(cnx, u'1') + u2 = self.create_user(cnx, u'2') + u3 = self.create_user(cnx, u'3') + ga = cnx.create_entity('CWGroup', name=u'A') + gb = cnx.create_entity('CWGroup', name=u'B') + u1.cw_set(friend=u2) + u2.cw_set(friend=u3) + ga.cw_set(friend=gb) + ga.cw_set(friend=u1) + cnx.commit() + for l1, l2 in ((u'1', u'2'), + (u'2', u'3')): + self.assertTrue(cnx.execute('Any U1,U2 WHERE U1 friend U2, U1 login %(l1)s, U2 login %(l2)s', + {'l1': l1, 'l2': l2})) + self.assertTrue(cnx.execute('Any U1,U2 WHERE U2 friend U1, U1 login %(l1)s, U2 login %(l2)s', + {'l1': l1, 'l2': l2})) + self.assertTrue(cnx.execute('Any GA,GB WHERE GA friend GB, GA name "A", GB name "B"')) + self.assertTrue(cnx.execute('Any GA,GB WHERE GB friend GA, GA name "A", GB name "B"')) + self.assertTrue(cnx.execute('Any GA,U1 WHERE GA friend U1, GA name "A", U1 login "1"')) + self.assertTrue(cnx.execute('Any GA,U1 WHERE U1 friend GA, GA name "A", U1 login "1"')) + self.assertFalse(cnx.execute('Any GA,U WHERE GA friend U, GA name "A", U login "2"')) + for l1, l2 in ((u'1', u'3'), + (u'3', u'1')): + self.assertFalse(cnx.execute('Any U1,U2 WHERE U1 friend U2, U1 login %(l1)s, U2 login %(l2)s', + {'l1': l1, 'l2': l2})) + self.assertFalse(cnx.execute('Any U1,U2 WHERE U2 friend U1, U1 login %(l1)s, U2 login %(l2)s', + {'l1': l1, 'l2': l2})) def test_html_tidy_hook(self): - req = self.request() - entity = req.create_entity('Workflow', name=u'wf1', - description_format=u'text/html', - description=u'yo') - self.assertEqual(entity.description, u'yo') - entity = req.create_entity('Workflow', name=u'wf2', - description_format=u'text/html', - description=u'yo') - self.assertEqual(entity.description, u'yo') - entity = req.create_entity('Workflow', name=u'wf3', - description_format=u'text/html', - description=u'yo') - self.assertEqual(entity.description, u'yo') - entity = req.create_entity('Workflow', name=u'wf4', - description_format=u'text/html', - description=u'R&D') - self.assertEqual(entity.description, u'R&D') - entity = req.create_entity('Workflow', name=u'wf5', - description_format=u'text/html', - description=u"
c'est l'été") - self.assertEqual(entity.description, u"
c'est l'été
") + with self.admin_access.web_request() as req: + entity = req.create_entity('Workflow', name=u'wf1', + description_format=u'text/html', + description=u'yo') + self.assertEqual(entity.description, u'yo') + entity = req.create_entity('Workflow', name=u'wf2', + description_format=u'text/html', + description=u'yo') + self.assertEqual(entity.description, u'yo') + entity = req.create_entity('Workflow', name=u'wf3', + description_format=u'text/html', + description=u'yo') + self.assertEqual(entity.description, u'yo') + entity = req.create_entity('Workflow', name=u'wf4', + description_format=u'text/html', + description=u'R&D') + self.assertEqual(entity.description, u'R&D') + entity = req.create_entity('Workflow', name=u'wf5', + description_format=u'text/html', + description=u"
c'est l'été") + self.assertEqual(entity.description, u"
c'est l'été
") def test_nonregr_html_tidy_hook_no_update(self): - entity = self.request().create_entity('Workflow', name=u'wf1', - description_format=u'text/html', - description=u'yo') - entity.cw_set(name=u'wf2') - self.assertEqual(entity.description, u'yo') - entity.cw_set(description=u'R&D

yo') - self.assertEqual(entity.description, u'R&D

yo

') + with self.admin_access.web_request() as req: + entity = req.create_entity('Workflow', name=u'wf1', + description_format=u'text/html', + description=u'yo') + entity.cw_set(name=u'wf2') + self.assertEqual(entity.description, u'yo') + entity.cw_set(description=u'R&D

yo') + self.assertEqual(entity.description, u'R&D

yo

') def test_metadata_cwuri(self): - entity = self.request().create_entity('Workflow', name=u'wf1') - self.assertEqual(entity.cwuri, self.repo.config['base-url'] + str(entity.eid)) + with self.admin_access.repo_cnx() as cnx: + entity = cnx.create_entity('Workflow', name=u'wf1') + self.assertEqual(entity.cwuri, self.repo.config['base-url'] + str(entity.eid)) def test_metadata_creation_modification_date(self): - _now = datetime.now() - entity = self.request().create_entity('Workflow', name=u'wf1') - self.assertEqual((entity.creation_date - _now).seconds, 0) - self.assertEqual((entity.modification_date - _now).seconds, 0) + with self.admin_access.repo_cnx() as cnx: + _now = datetime.now() + entity = cnx.create_entity('Workflow', name=u'wf1') + self.assertEqual((entity.creation_date - _now).seconds, 0) + self.assertEqual((entity.modification_date - _now).seconds, 0) def test_metadata_created_by(self): - entity = self.request().create_entity('Bookmark', title=u'wf1', path=u'/view') - self.commit() # fire operations - self.assertEqual(len(entity.created_by), 1) # make sure we have only one creator - self.assertEqual(entity.created_by[0].eid, self.session.user.eid) + with self.admin_access.repo_cnx() as cnx: + entity = cnx.create_entity('Bookmark', title=u'wf1', path=u'/view') + cnx.commit() # fire operations + self.assertEqual(len(entity.created_by), 1) # make sure we have only one creator + self.assertEqual(entity.created_by[0].eid, self.session.user.eid) def test_metadata_owned_by(self): - entity = self.request().create_entity('Bookmark', title=u'wf1', path=u'/view') - self.commit() # fire operations - self.assertEqual(len(entity.owned_by), 1) # make sure we have only one owner - self.assertEqual(entity.owned_by[0].eid, self.session.user.eid) + with self.admin_access.repo_cnx() as cnx: + entity = cnx.create_entity('Bookmark', title=u'wf1', path=u'/view') + cnx.commit() # fire operations + self.assertEqual(len(entity.owned_by), 1) # make sure we have only one owner + self.assertEqual(entity.owned_by[0].eid, self.session.user.eid) def test_user_login_stripped(self): - req = self.request() - u = self.create_user(req, ' joe ') - tname = self.execute('Any L WHERE E login L, E eid %(e)s', - {'e': u.eid})[0][0] - self.assertEqual(tname, 'joe') - self.execute('SET X login " jijoe " WHERE X eid %(x)s', {'x': u.eid}) - tname = self.execute('Any L WHERE E login L, E eid %(e)s', - {'e': u.eid})[0][0] - self.assertEqual(tname, 'jijoe') + with self.admin_access.repo_cnx() as cnx: + u = self.create_user(cnx, ' joe ') + tname = cnx.execute('Any L WHERE E login L, E eid %(e)s', + {'e': u.eid})[0][0] + self.assertEqual(tname, 'joe') + cnx.execute('SET X login " jijoe " WHERE X eid %(x)s', {'x': u.eid}) + tname = cnx.execute('Any L WHERE E login L, E eid %(e)s', + {'e': u.eid})[0][0] + self.assertEqual(tname, 'jijoe') class UserGroupHooksTC(CubicWebTC): def test_user_synchronization(self): - req = self.request() - self.create_user(req, 'toto', password='hop', commit=False) - self.assertRaises(AuthenticationError, - self.repo.connect, u'toto', password='hop') - self.commit() - cnxid = self.repo.connect(u'toto', password='hop') - self.assertNotEqual(cnxid, self.session.id) - self.execute('DELETE CWUser X WHERE X login "toto"') - self.repo.execute(cnxid, 'State X') - self.commit() - self.assertRaises(BadConnectionId, - self.repo.execute, cnxid, 'State X') + with self.admin_access.repo_cnx() as cnx: + self.create_user(cnx, 'toto', password='hop', commit=False) + self.assertRaises(AuthenticationError, + self.repo.connect, u'toto', password='hop') + cnx.commit() + cnxid = self.repo.connect(u'toto', password='hop') + self.assertNotEqual(cnxid, self.session.id) + cnx.execute('DELETE CWUser X WHERE X login "toto"') + self.repo.execute(cnxid, 'State X') + cnx.commit() + self.assertRaises(BadConnectionId, + self.repo.execute, cnxid, 'State X') def test_user_group_synchronization(self): - user = self.session.user - self.assertEqual(user.groups, set(('managers',))) - self.execute('SET X in_group G WHERE X eid %s, G name "guests"' % user.eid) - self.assertEqual(user.groups, set(('managers',))) - self.commit() - self.assertEqual(user.groups, set(('managers', 'guests'))) - self.execute('DELETE X in_group G WHERE X eid %s, G name "guests"' % user.eid) - self.assertEqual(user.groups, set(('managers', 'guests'))) - self.commit() - self.assertEqual(user.groups, set(('managers',))) + with self.admin_access.repo_cnx() as cnx: + user = cnx.user + self.assertEqual(user.groups, set(('managers',))) + cnx.execute('SET X in_group G WHERE X eid %s, G name "guests"' % user.eid) + self.assertEqual(user.groups, set(('managers',))) + cnx.commit() + self.assertEqual(user.groups, set(('managers', 'guests'))) + cnx.execute('DELETE X in_group G WHERE X eid %s, G name "guests"' % user.eid) + self.assertEqual(user.groups, set(('managers', 'guests'))) + cnx.commit() + self.assertEqual(user.groups, set(('managers',))) def test_user_composite_owner(self): - req = self.request() - ueid = self.create_user(req, 'toto').eid - # composite of euser should be owned by the euser regardless of who created it - self.execute('INSERT EmailAddress X: X address "toto@logilab.fr", U use_email X ' - 'WHERE U login "toto"') - self.commit() - self.assertEqual(self.execute('Any A WHERE X owned_by U, U use_email X,' - 'U login "toto", X address A')[0][0], - 'toto@logilab.fr') + with self.admin_access.repo_cnx() as cnx: + self.create_user(cnx, 'toto').eid + # composite of euser should be owned by the euser regardless of who created it + cnx.execute('INSERT EmailAddress X: X address "toto@logilab.fr", U use_email X ' + 'WHERE U login "toto"') + cnx.commit() + self.assertEqual(cnx.execute('Any A WHERE X owned_by U, U use_email X,' + 'U login "toto", X address A')[0][0], + 'toto@logilab.fr') def test_no_created_by_on_deleted_entity(self): - eid = self.execute('INSERT EmailAddress X: X address "toto@logilab.fr"')[0][0] - self.execute('DELETE EmailAddress X WHERE X eid %s' % eid) - self.commit() - self.assertFalse(self.execute('Any X WHERE X created_by Y, X eid >= %(x)s', {'x': eid})) + with self.admin_access.repo_cnx() as cnx: + eid = cnx.execute('INSERT EmailAddress X: X address "toto@logilab.fr"')[0][0] + cnx.execute('DELETE EmailAddress X WHERE X eid %s' % eid) + cnx.commit() + self.assertFalse(cnx.execute('Any X WHERE X created_by Y, X eid >= %(x)s', {'x': eid})) class SchemaHooksTC(CubicWebTC): def test_duplicate_etype_error(self): - # check we can't add a CWEType or CWRType entity if it already exists one - # with the same name - self.assertRaises(ValidationError, - self.execute, 'INSERT CWEType X: X name "CWUser"') - self.assertRaises(ValidationError, - self.execute, 'INSERT CWRType X: X name "in_group"') + with self.admin_access.repo_cnx() as cnx: + # check we can't add a CWEType or CWRType entity if it already exists one + # with the same name + self.assertRaises(ValidationError, + cnx.execute, 'INSERT CWEType X: X name "CWUser"') + cnx.rollback() + self.assertRaises(ValidationError, + cnx.execute, 'INSERT CWRType X: X name "in_group"') def test_validation_unique_constraint(self): - with self.assertRaises(ValidationError) as cm: - self.execute('INSERT CWUser X: X login "admin"') - ex = cm.exception - ex.translate(unicode) - self.assertIsInstance(ex.entity, int) - self.assertEqual(ex.errors, {'login-subject': 'the value "admin" is already used, use another one'}) + with self.admin_access.repo_cnx() as cnx: + with self.assertRaises(ValidationError) as cm: + cnx.execute('INSERT CWUser X: X login "admin"') + ex = cm.exception + ex.translate(unicode) + self.assertIsInstance(ex.entity, int) + self.assertEqual(ex.errors, {'login-subject': 'the value "admin" is already used, use another one'}) if __name__ == '__main__':