# -*- coding: utf-8 -*-
# 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.
#
# 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/>.
"""functional tests for core hooks
Note:
syncschema.py hooks are mostly tested in server/test/unittest_migrations.py
"""
from datetime import datetime
from cubicweb import ValidationError, AuthenticationError, BadConnectionId
from cubicweb.devtools.testlib import CubicWebTC
class CoreHooksTC(CubicWebTC):
def test_inlined(self):
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):
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):
with self.admin_access.client_cnx() as cnx:
entity = cnx.create_entity('Workflow', name=u'wf1',
description_format=u'text/html',
description=u'yo')
self.assertEqual(u'yo', entity.description)
entity = cnx.create_entity('Workflow', name=u'wf2',
description_format=u'text/html',
description=u'<b>yo')
self.assertEqual(u'<b>yo</b>', entity.description)
entity = cnx.create_entity('Workflow', name=u'wf3',
description_format=u'text/html',
description=u'<b>yo</b>')
self.assertEqual(u'<b>yo</b>', entity.description)
entity = cnx.create_entity('Workflow', name=u'wf4',
description_format=u'text/html',
description=u'<b>R&D</b>')
self.assertEqual(u'<b>R&D</b>', entity.description, )
entity = cnx.create_entity('Workflow', name=u'wf5',
description_format=u'text/html',
description=u"<div>c'est <b>l'été")
self.assertEqual(u"<div>c'est <b>l'été</b></div>", entity.description)
def test_nonregr_html_tidy_hook_no_update(self):
with self.admin_access.client_cnx() as cnx:
entity = cnx.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<p>yo')
self.assertEqual(entity.description, u'R&D<p>yo</p>')
def test_metadata_cwuri(self):
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):
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):
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, cnx.user.eid)
def test_metadata_owned_by(self):
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, cnx.user.eid)
def test_user_login_stripped(self):
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):
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, cnx.sessionid)
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):
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):
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_user_composite_no_owner_on_deleted_entity(self):
with self.admin_access.repo_cnx() as cnx:
u = self.create_user(cnx, 'toto').eid
cnx.commit()
e = cnx.create_entity('EmailAddress', address=u'toto@logilab.fr', reverse_use_email=u)
e.cw_delete()
cnx.commit()
self.assertFalse(cnx.system_sql(
'SELECT * FROM owned_by_relation '
'WHERE eid_from NOT IN (SELECT eid FROM entities)').fetchall())
def test_no_created_by_on_deleted_entity(self):
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):
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.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__':
from logilab.common.testlib import unittest_main
unittest_main()