cubicweb/hooks/test/unittest_hooks.py
changeset 11057 0b59724cb3f2
parent 11034 75d752e6daf7
child 11366 80dec361a5d0
equal deleted inserted replaced
11052:058bb3dc685f 11057:0b59724cb3f2
       
     1 # -*- coding: utf-8 -*-
       
     2 # copyright 2003-2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
       
     3 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
       
     4 #
       
     5 # This file is part of CubicWeb.
       
     6 #
       
     7 # CubicWeb is free software: you can redistribute it and/or modify it under the
       
     8 # terms of the GNU Lesser General Public License as published by the Free
       
     9 # Software Foundation, either version 2.1 of the License, or (at your option)
       
    10 # any later version.
       
    11 #
       
    12 # CubicWeb is distributed in the hope that it will be useful, but WITHOUT
       
    13 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
       
    14 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
       
    15 # details.
       
    16 #
       
    17 # You should have received a copy of the GNU Lesser General Public License along
       
    18 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
       
    19 """functional tests for core hooks
       
    20 
       
    21 Note:
       
    22   syncschema.py hooks are mostly tested in server/test/unittest_migrations.py
       
    23 """
       
    24 
       
    25 from datetime import datetime
       
    26 
       
    27 from six import text_type
       
    28 
       
    29 from pytz import utc
       
    30 from cubicweb import ValidationError, AuthenticationError, BadConnectionId
       
    31 from cubicweb.devtools.testlib import CubicWebTC
       
    32 
       
    33 
       
    34 class CoreHooksTC(CubicWebTC):
       
    35 
       
    36     def test_inlined(self):
       
    37         with self.admin_access.repo_cnx() as cnx:
       
    38             self.assertEqual(self.repo.schema['sender'].inlined, True)
       
    39             cnx.execute('INSERT EmailAddress X: X address "toto@logilab.fr", X alias "hop"')
       
    40             cnx.execute('INSERT EmailPart X: X content_format "text/plain", X ordernum 1, '
       
    41                         'X content "this is a test"')
       
    42             eeid = cnx.execute('INSERT Email X: X messageid "<1234>", X subject "test", '
       
    43                                'X sender Y, X recipients Y, X parts P '
       
    44                                'WHERE Y is EmailAddress, P is EmailPart')[0][0]
       
    45             cnx.execute('SET X sender Y WHERE X is Email, Y is EmailAddress')
       
    46             rset = cnx.execute('Any S WHERE X sender S, X eid %s' % eeid)
       
    47             self.assertEqual(len(rset), 1)
       
    48 
       
    49     def test_symmetric(self):
       
    50         with self.admin_access.repo_cnx() as cnx:
       
    51             u1 = self.create_user(cnx, u'1')
       
    52             u2 = self.create_user(cnx, u'2')
       
    53             u3 = self.create_user(cnx, u'3')
       
    54             ga = cnx.create_entity('CWGroup', name=u'A')
       
    55             gb = cnx.create_entity('CWGroup', name=u'B')
       
    56             u1.cw_set(friend=u2)
       
    57             u2.cw_set(friend=u3)
       
    58             ga.cw_set(friend=gb)
       
    59             ga.cw_set(friend=u1)
       
    60             cnx.commit()
       
    61             for l1, l2 in ((u'1', u'2'),
       
    62                            (u'2', u'3')):
       
    63                 self.assertTrue(cnx.execute('Any U1,U2 WHERE U1 friend U2, U1 login %(l1)s, U2 login %(l2)s',
       
    64                                             {'l1': l1, 'l2': l2}))
       
    65                 self.assertTrue(cnx.execute('Any U1,U2 WHERE U2 friend U1, U1 login %(l1)s, U2 login %(l2)s',
       
    66                                             {'l1': l1, 'l2': l2}))
       
    67             self.assertTrue(cnx.execute('Any GA,GB WHERE GA friend GB, GA name "A", GB name "B"'))
       
    68             self.assertTrue(cnx.execute('Any GA,GB WHERE GB friend GA, GA name "A", GB name "B"'))
       
    69             self.assertTrue(cnx.execute('Any GA,U1 WHERE GA friend U1, GA name "A", U1 login "1"'))
       
    70             self.assertTrue(cnx.execute('Any GA,U1 WHERE U1 friend GA, GA name "A", U1 login "1"'))
       
    71             self.assertFalse(cnx.execute('Any GA,U WHERE GA friend U, GA name "A", U login "2"'))
       
    72             for l1, l2 in ((u'1', u'3'),
       
    73                            (u'3', u'1')):
       
    74                 self.assertFalse(cnx.execute('Any U1,U2 WHERE U1 friend U2, U1 login %(l1)s, U2 login %(l2)s',
       
    75                                              {'l1': l1, 'l2': l2}))
       
    76                 self.assertFalse(cnx.execute('Any U1,U2 WHERE U2 friend U1, U1 login %(l1)s, U2 login %(l2)s',
       
    77                                              {'l1': l1, 'l2': l2}))
       
    78 
       
    79     def test_html_tidy_hook(self):
       
    80         with self.admin_access.client_cnx() as cnx:
       
    81             entity = cnx.create_entity('Workflow', name=u'wf1',
       
    82                                        description_format=u'text/html',
       
    83                                        description=u'yo')
       
    84             self.assertEqual(u'yo', entity.description)
       
    85             entity = cnx.create_entity('Workflow', name=u'wf2',
       
    86                                        description_format=u'text/html',
       
    87                                        description=u'<b>yo')
       
    88             self.assertEqual(u'<b>yo</b>', entity.description)
       
    89             entity = cnx.create_entity('Workflow', name=u'wf3',
       
    90                                        description_format=u'text/html',
       
    91                                        description=u'<b>yo</b>')
       
    92             self.assertEqual(u'<b>yo</b>', entity.description)
       
    93             entity = cnx.create_entity('Workflow', name=u'wf4',
       
    94                                        description_format=u'text/html',
       
    95                                        description=u'<b>R&D</b>')
       
    96             self.assertEqual(u'<b>R&amp;D</b>', entity.description, )
       
    97             entity = cnx.create_entity('Workflow', name=u'wf5',
       
    98                                        description_format=u'text/html',
       
    99                                        description=u"<div>c&apos;est <b>l'ét&eacute;")
       
   100             self.assertEqual(u"<div>c'est <b>l'été</b></div>", entity.description)
       
   101 
       
   102     def test_nonregr_html_tidy_hook_no_update(self):
       
   103         with self.admin_access.client_cnx() as cnx:
       
   104             entity = cnx.create_entity('Workflow', name=u'wf1',
       
   105                                        description_format=u'text/html',
       
   106                                        description=u'yo')
       
   107             entity.cw_set(name=u'wf2')
       
   108             self.assertEqual(entity.description, u'yo')
       
   109             entity.cw_set(description=u'R&D<p>yo')
       
   110             self.assertEqual(entity.description, u'R&amp;D<p>yo</p>')
       
   111 
       
   112     def test_metadata_cwuri(self):
       
   113         with self.admin_access.repo_cnx() as cnx:
       
   114             entity = cnx.create_entity('Workflow', name=u'wf1')
       
   115             self.assertEqual(entity.cwuri, self.repo.config['base-url'] + str(entity.eid))
       
   116 
       
   117     def test_metadata_creation_modification_date(self):
       
   118         with self.admin_access.repo_cnx() as cnx:
       
   119             _now = datetime.now(utc)
       
   120             entity = cnx.create_entity('Workflow', name=u'wf1')
       
   121             self.assertEqual((entity.creation_date - _now).seconds, 0)
       
   122             self.assertEqual((entity.modification_date - _now).seconds, 0)
       
   123 
       
   124     def test_metadata_created_by(self):
       
   125         with self.admin_access.repo_cnx() as cnx:
       
   126             entity = cnx.create_entity('Bookmark', title=u'wf1', path=u'/view')
       
   127             cnx.commit() # fire operations
       
   128             self.assertEqual(len(entity.created_by), 1) # make sure we have only one creator
       
   129             self.assertEqual(entity.created_by[0].eid, cnx.user.eid)
       
   130 
       
   131     def test_metadata_owned_by(self):
       
   132         with self.admin_access.repo_cnx() as cnx:
       
   133             entity = cnx.create_entity('Bookmark', title=u'wf1', path=u'/view')
       
   134             cnx.commit() # fire operations
       
   135             self.assertEqual(len(entity.owned_by), 1) # make sure we have only one owner
       
   136             self.assertEqual(entity.owned_by[0].eid, cnx.user.eid)
       
   137 
       
   138     def test_user_login_stripped(self):
       
   139         with self.admin_access.repo_cnx() as cnx:
       
   140             u = self.create_user(cnx, '  joe  ')
       
   141             tname = cnx.execute('Any L WHERE E login L, E eid %(e)s',
       
   142                                 {'e': u.eid})[0][0]
       
   143             self.assertEqual(tname, 'joe')
       
   144             cnx.execute('SET X login " jijoe " WHERE X eid %(x)s', {'x': u.eid})
       
   145             tname = cnx.execute('Any L WHERE E login L, E eid %(e)s',
       
   146                                 {'e': u.eid})[0][0]
       
   147             self.assertEqual(tname, 'jijoe')
       
   148 
       
   149 
       
   150 
       
   151 class UserGroupHooksTC(CubicWebTC):
       
   152 
       
   153     def test_user_group_synchronization(self):
       
   154         with self.admin_access.repo_cnx() as cnx:
       
   155             user = cnx.user
       
   156             self.assertEqual(user.groups, set(('managers',)))
       
   157             cnx.execute('SET X in_group G WHERE X eid %s, G name "guests"' % user.eid)
       
   158             self.assertEqual(user.groups, set(('managers',)))
       
   159             cnx.commit()
       
   160             self.assertEqual(user.groups, set(('managers', 'guests')))
       
   161             cnx.execute('DELETE X in_group G WHERE X eid %s, G name "guests"' % user.eid)
       
   162             self.assertEqual(user.groups, set(('managers', 'guests')))
       
   163             cnx.commit()
       
   164             self.assertEqual(user.groups, set(('managers',)))
       
   165 
       
   166     def test_user_composite_owner(self):
       
   167         with self.admin_access.repo_cnx() as cnx:
       
   168             self.create_user(cnx, 'toto').eid
       
   169             # composite of euser should be owned by the euser regardless of who created it
       
   170             cnx.execute('INSERT EmailAddress X: X address "toto@logilab.fr", U use_email X '
       
   171                          'WHERE U login "toto"')
       
   172             cnx.commit()
       
   173             self.assertEqual(cnx.execute('Any A WHERE X owned_by U, U use_email X,'
       
   174                                            'U login "toto", X address A')[0][0],
       
   175                               'toto@logilab.fr')
       
   176 
       
   177     def test_user_composite_no_owner_on_deleted_entity(self):
       
   178         with self.admin_access.repo_cnx() as cnx:
       
   179             u = self.create_user(cnx, 'toto').eid
       
   180             cnx.commit()
       
   181             e = cnx.create_entity('EmailAddress', address=u'toto@logilab.fr', reverse_use_email=u)
       
   182             e.cw_delete()
       
   183             cnx.commit()
       
   184             self.assertFalse(cnx.system_sql(
       
   185                 'SELECT * FROM owned_by_relation '
       
   186                 'WHERE eid_from NOT IN (SELECT eid FROM entities)').fetchall())
       
   187 
       
   188     def test_no_created_by_on_deleted_entity(self):
       
   189         with self.admin_access.repo_cnx() as cnx:
       
   190             eid = cnx.execute('INSERT EmailAddress X: X address "toto@logilab.fr"')[0][0]
       
   191             cnx.execute('DELETE EmailAddress X WHERE X eid %s' % eid)
       
   192             cnx.commit()
       
   193             self.assertFalse(cnx.execute('Any X WHERE X created_by Y, X eid >= %(x)s', {'x': eid}))
       
   194 
       
   195 
       
   196 
       
   197 class SchemaHooksTC(CubicWebTC):
       
   198 
       
   199     def test_duplicate_etype_error(self):
       
   200         with self.admin_access.repo_cnx() as cnx:
       
   201             # check we can't add a CWEType or CWRType entity if it already exists one
       
   202             # with the same name
       
   203             self.assertRaises(ValidationError,
       
   204                               cnx.execute, 'INSERT CWEType X: X name "CWUser"')
       
   205             cnx.rollback()
       
   206             self.assertRaises(ValidationError,
       
   207                               cnx.execute, 'INSERT CWRType X: X name "in_group"')
       
   208 
       
   209     def test_validation_unique_constraint(self):
       
   210         with self.admin_access.repo_cnx() as cnx:
       
   211             with self.assertRaises(ValidationError) as cm:
       
   212                 cnx.execute('INSERT CWUser X: X login "admin"')
       
   213             ex = cm.exception
       
   214             ex.translate(text_type)
       
   215             self.assertIsInstance(ex.entity, int)
       
   216             self.assertEqual(ex.errors, {'login-subject': 'the value "admin" is already used, use another one'})
       
   217 
       
   218 
       
   219 if __name__ == '__main__':
       
   220     from logilab.common.testlib import unittest_main
       
   221     unittest_main()