cubicweb: comparison server/test/unittest

equal deleted inserted replaced

-:b1f6777fc839
+:8af7c6d86efb
 """functional tests for server'security"""
 import sys
 from logilab.common.testlib import unittest_main, TestCase
+from rql import RQLException
 from cubicweb.devtools.testlib import CubicWebTC
 from cubicweb import Unauthorized, ValidationError, QueryError
 from cubicweb.schema import ERQLExpression
 from cubicweb.server.querier import check_read_access
 class BaseSecurityTC(CubicWebTC):
 def setup_database(self):
 super(BaseSecurityTC, self).setup_database()
-req = self.request()
+self.create_user(self.request(), 'iaminusersgrouponly')
-self.create_user(req, 'iaminusersgrouponly')
-readoriggroups = self.schema['Personne'].permissions['read']
-addoriggroups = self.schema['Personne'].permissions['add']
-def fix_perm():
-self.schema['Personne'].set_action_permissions('read', readoriggroups)
-self.schema['Personne'].set_action_permissions('add', addoriggroups)
-self.addCleanup(fix_perm)
 class LowLevelSecurityFunctionTC(BaseSecurityTC):
 def test_check_read_access(self):
 rql = u'Personne U where U nom "managers"'
 rqlst = self.repo.vreg.rqlhelper.parse(rql).children[0]
-origgroups = self.schema['Personne'].get_groups('read')
+with self.temporary_permissions(Personne={'read': ('users', 'managers')}):
-self.schema['Personne'].set_action_permissions('read', ('users', 'managers'))
+self.repo.vreg.solutions(self.session, rqlst, None)
-self.repo.vreg.solutions(self.session, rqlst, None)
+solution = rqlst.solutions[0]
-solution = rqlst.solutions[0]
+check_read_access(self.session, rqlst, solution, {})
-check_read_access(self.session, rqlst, solution, {})
+with self.login('anon') as cu:
-cnx = self.login('anon')
+self.assertRaises(Unauthorized,
-cu = cnx.cursor()
+check_read_access,
-self.assertRaises(Unauthorized,
+self.session, rqlst, solution, {})
-check_read_access,
+self.assertRaises(Unauthorized, cu.execute, rql)
-self.session, rqlst, solution, {})
-self.assertRaises(Unauthorized, cu.execute, rql)
 def test_upassword_not_selectable(self):
 self.assertRaises(Unauthorized,
 self.execute, 'Any X,P WHERE X is CWUser, X upassword P')
 self.rollback()
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+self.assertRaises(Unauthorized,
-self.assertRaises(Unauthorized,
+cu.execute, 'Any X,P WHERE X is CWUser, X upassword P')
-cu.execute, 'Any X,P WHERE X is CWUser, X upassword P')
 class SecurityRewritingTC(BaseSecurityTC):
 def hijack_source_execute(self):
 def syntax_tree_search(*args, **kwargs):
 def tearDown(self):
 self.repo.system_source.__dict__.pop('syntax_tree_search', None)
 super(SecurityRewritingTC, self).tearDown()
 def test_not_relation_read_security(self):
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly'):
 self.hijack_source_execute()
 self.execute('Any U WHERE NOT A todo_by U, A is Affaire')
 self.assertEqual(self.query[0][1].as_string(),
 'Any U WHERE NOT EXISTS(A todo_by U), A is Affaire')
 self.execute('Any U WHERE NOT EXISTS(A todo_by U), A is Affaire')
 self.assertEqual(self.query[0][1].as_string(),
 'Any U WHERE NOT EXISTS(A todo_by U), A is Affaire')
-cnx.close()
 class SecurityTC(BaseSecurityTC):
 def setUp(self):
 BaseSecurityTC.setUp(self)
 self.execute("INSERT Personne X: X nom 'bidule'")
 self.execute('INSERT CWGroup X: X name "staff"')
 self.commit()
 def test_insert_security(self):
-cnx = self.login('anon')
+with self.login('anon') as cu:
-cu = cnx.cursor()
+cu.execute("INSERT Personne X: X nom 'bidule'")
-cu.execute("INSERT Personne X: X nom 'bidule'")
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
+self.assertEqual(cu.execute('Personne X').rowcount, 1)
-self.assertEqual(cu.execute('Personne X').rowcount, 1)
-cnx.close()
 def test_insert_rql_permission(self):
 # test user can only add une affaire related to a societe he owns
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+cu.execute("INSERT Affaire X: X sujet 'cool'")
-cu.execute("INSERT Affaire X: X sujet 'cool'")
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
 # test nothing has actually been inserted
-self.restore_connection()
 self.assertEqual(self.execute('Affaire X').rowcount, 1)
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+cu.execute("INSERT Affaire X: X sujet 'cool'")
-cu.execute("INSERT Affaire X: X sujet 'cool'")
+cu.execute("INSERT Societe X: X nom 'chouette'")
-cu.execute("INSERT Societe X: X nom 'chouette'")
+cu.execute("SET A concerne S WHERE A sujet 'cool', S nom 'chouette'")
-cu.execute("SET A concerne S WHERE A sujet 'cool', S nom 'chouette'")
+self.commit()
-cnx.commit()
-cnx.close()
 def test_update_security_1(self):
-cnx = self.login('anon')
+with self.login('anon') as cu:
-cu = cnx.cursor()
+# local security check
-# local security check
+cu.execute( "SET X nom 'bidulechouette' WHERE X is Personne")
-cu.execute( "SET X nom 'bidulechouette' WHERE X is Personne")
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
-self.restore_connection()
 self.assertEqual(self.execute('Personne X WHERE X nom "bidulechouette"').rowcount, 0)
 def test_update_security_2(self):
-cnx = self.login('anon')
+with self.temporary_permissions(Personne={'read': ('users', 'managers'),
-cu = cnx.cursor()
+'add': ('guests', 'users', 'managers')}):
-self.repo.schema['Personne'].set_action_permissions('read', ('users', 'managers'))
+with self.login('anon') as cu:
-self.repo.schema['Personne'].set_action_permissions('add', ('guests', 'users', 'managers'))
+self.assertRaises(Unauthorized, cu.execute, "SET X nom 'bidulechouette' WHERE X is Personne")
-self.assertRaises(Unauthorized, cu.execute, "SET X nom 'bidulechouette' WHERE X is Personne")
+self.rollback()
-#self.assertRaises(Unauthorized, cnx.commit)
+# self.assertRaises(Unauthorized, cnx.commit)
 # test nothing has actually been inserted
-self.restore_connection()
 self.assertEqual(self.execute('Personne X WHERE X nom "bidulechouette"').rowcount, 0)
 def test_update_security_3(self):
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+cu.execute("INSERT Personne X: X nom 'biduuule'")
-cu.execute("INSERT Personne X: X nom 'biduuule'")
+cu.execute("INSERT Societe X: X nom 'looogilab'")
-cu.execute("INSERT Societe X: X nom 'looogilab'")
+cu.execute("SET X travaille S WHERE X nom 'biduuule', S nom 'looogilab'")
-cu.execute("SET X travaille S WHERE X nom 'biduuule', S nom 'looogilab'")
-cnx.close()
 def test_update_rql_permission(self):
 self.execute("SET A concerne S WHERE A is Affaire, S is Societe")
 self.commit()
 # test user can only update une affaire related to a societe he owns
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+cu.execute("SET X sujet 'pascool' WHERE X is Affaire")
-cu.execute("SET X sujet 'pascool' WHERE X is Affaire")
+# this won't actually do anything since the selection query won't return anything
-# this won't actually do anything since the selection query won't return anything
+self.commit()
-cnx.commit()
+# to actually get Unauthorized exception, try to update an entity we can read
-# to actually get Unauthorized exception, try to update an entity we can read
+cu.execute("SET X nom 'toto' WHERE X is Societe")
-cu.execute("SET X nom 'toto' WHERE X is Societe")
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
+cu.execute("INSERT Affaire X: X sujet 'pascool'")
-cu.execute("INSERT Affaire X: X sujet 'pascool'")
+cu.execute("INSERT Societe X: X nom 'chouette'")
-cu.execute("INSERT Societe X: X nom 'chouette'")
+cu.execute("SET A concerne S WHERE A sujet 'pascool', S nom 'chouette'")
-cu.execute("SET A concerne S WHERE A sujet 'pascool', S nom 'chouette'")
+cu.execute("SET X sujet 'habahsicestcool' WHERE X sujet 'pascool'")
-cu.execute("SET X sujet 'habahsicestcool' WHERE X sujet 'pascool'")
+self.commit()
-cnx.commit()
-cnx.close()
 def test_delete_security(self):
 # FIXME: sample below fails because we don't detect "owner" can't delete
 # user anyway, and since no user with login == 'bidule' exists, no
 # exception is raised
 #user._groups = {'guests':1}
 #self.assertRaises(Unauthorized,
 #                  self.o.execute, user, "DELETE CWUser X WHERE X login 'bidule'")
 # check local security
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+self.assertRaises(Unauthorized, cu.execute, "DELETE CWGroup Y WHERE Y name 'staff'")
-self.assertRaises(Unauthorized, cu.execute, "DELETE CWGroup Y WHERE Y name 'staff'")
+self.rollback()
-cnx.close()
 def test_delete_rql_permission(self):
 self.execute("SET A concerne S WHERE A is Affaire, S is Societe")
 self.commit()
 # test user can only dele une affaire related to a societe he owns
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+# this won't actually do anything since the selection query won't return anything
-# this won't actually do anything since the selection query won't return anything
+cu.execute("DELETE Affaire X")
-cu.execute("DELETE Affaire X")
+self.commit()
-cnx.commit()
+# to actually get Unauthorized exception, try to delete an entity we can read
-# to actually get Unauthorized exception, try to delete an entity we can read
+self.assertRaises(Unauthorized, cu.execute, "DELETE Societe S")
-self.assertRaises(Unauthorized, cu.execute, "DELETE Societe S")
+self.assertRaises(QueryError, self.commit) # can't commit anymore
-self.assertRaises(QueryError, cnx.commit) # can't commit anymore
+self.rollback() # required after Unauthorized
-cnx.rollback() # required after Unauthorized
+cu.execute("INSERT Affaire X: X sujet 'pascool'")
-cu.execute("INSERT Affaire X: X sujet 'pascool'")
+cu.execute("INSERT Societe X: X nom 'chouette'")
-cu.execute("INSERT Societe X: X nom 'chouette'")
+cu.execute("SET A concerne S WHERE A sujet 'pascool', S nom 'chouette'")
-cu.execute("SET A concerne S WHERE A sujet 'pascool', S nom 'chouette'")
+self.commit()
-cnx.commit()
 ##         # this one should fail since it will try to delete two affaires, one authorized
 ##         # and the other not
 ##         self.assertRaises(Unauthorized, cu.execute, "DELETE Affaire X")
 cu.execute("DELETE Affaire X WHERE X sujet 'pascool'")
-cnx.commit()
+self.commit()
-cnx.close()
 def test_insert_relation_rql_permission(self):
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-session = self.session
+cu.execute("SET A concerne S WHERE A is Affaire, S is Societe")
-cu = cnx.cursor(session)
+# should raise Unauthorized since user don't own S though this won't
-cu.execute("SET A concerne S WHERE A is Affaire, S is Societe")
+# actually do anything since the selection query won't return
-# should raise Unauthorized since user don't own S
+# anything
-# though this won't actually do anything since the selection query won't return anything
+self.commit()
-cnx.commit()
+# to actually get Unauthorized exception, try to insert a relation
-# to actually get Unauthorized exception, try to insert a relation were we can read both entities
+# were we can read both entities
 rset = cu.execute('Personne P')
 self.assertEqual(len(rset), 1)
 ent = rset.get_entity(0, 0)
-session.set_cnxset() # necessary
+self.assertFalse(cu.execute('Any P,S WHERE P travaille S,P is Personne, S is Societe'))
 self.assertRaises(Unauthorized, ent.cw_check_perm, 'update')
 self.assertRaises(Unauthorized,
 cu.execute, "SET P travaille S WHERE P is Personne, S is Societe")
-self.assertRaises(QueryError, cnx.commit) # can't commit anymore
+self.assertRaises(QueryError, self.commit) # can't commit anymore
-cnx.rollback()
+self.rollback()
 # test nothing has actually been inserted:
-self.assertEqual(cu.execute('Any P,S WHERE P travaille S,P is Personne, S is Societe').rowcount, 0)
+self.assertFalse(cu.execute('Any P,S WHERE P travaille S,P is Personne, S is Societe'))
 cu.execute("INSERT Societe X: X nom 'chouette'")
 cu.execute("SET A concerne S WHERE A is Affaire, S nom 'chouette'")
-cnx.commit()
+self.commit()
-cnx.close()
 def test_delete_relation_rql_permission(self):
 self.execute("SET A concerne S WHERE A is Affaire, S is Societe")
 self.commit()
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+# this won't actually do anything since the selection query won't return anything
-# this won't actually do anything since the selection query won't return anything
+cu.execute("DELETE A concerne S")
-cu.execute("DELETE A concerne S")
+self.commit()
-cnx.commit()
 # to actually get Unauthorized exception, try to delete a relation we can read
-self.restore_connection()
 eid = self.execute("INSERT Affaire X: X sujet 'pascool'")[0][0]
 self.execute('SET X owned_by U WHERE X eid %(x)s, U login "iaminusersgrouponly"', {'x': eid})
 self.execute("SET A concerne S WHERE A sujet 'pascool', S is Societe")
 self.commit()
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+self.assertRaises(Unauthorized, cu.execute, "DELETE A concerne S")
-self.assertRaises(Unauthorized, cu.execute, "DELETE A concerne S")
+self.assertRaises(QueryError, self.commit) # can't commit anymore
-self.assertRaises(QueryError, cnx.commit) # can't commit anymore
+self.rollback() # required after Unauthorized
-cnx.rollback() # required after Unauthorized
+cu.execute("INSERT Societe X: X nom 'chouette'")
-cu.execute("INSERT Societe X: X nom 'chouette'")
+cu.execute("SET A concerne S WHERE A is Affaire, S nom 'chouette'")
-cu.execute("SET A concerne S WHERE A is Affaire, S nom 'chouette'")
+self.commit()
-cnx.commit()
+cu.execute("DELETE A concerne S WHERE S nom 'chouette'")
-cu.execute("DELETE A concerne S WHERE S nom 'chouette'")
+self.commit()
-cnx.close()
 def test_user_can_change_its_upassword(self):
 req = self.request()
 ueid = self.create_user(req, 'user').eid
-cnx = self.login('user')
+with self.login('user') as cu:
-cu = cnx.cursor()
+cu.execute('SET X upassword %(passwd)s WHERE X eid %(x)s',
-cu.execute('SET X upassword %(passwd)s WHERE X eid %(x)s',
+{'x': ueid, 'passwd': 'newpwd'})
-{'x': ueid, 'passwd': 'newpwd'})
+self.commit()
-cnx.commit()
-cnx.close()
 cnx = self.login('user', password='newpwd')
 cnx.close()
 def test_user_cant_change_other_upassword(self):
 req = self.request()
 ueid = self.create_user(req, 'otheruser').eid
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+cu.execute('SET X upassword %(passwd)s WHERE X eid %(x)s',
-cu.execute('SET X upassword %(passwd)s WHERE X eid %(x)s',
+{'x': ueid, 'passwd': 'newpwd'})
-{'x': ueid, 'passwd': 'newpwd'})
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
-cnx.close()
 # read security test
 def test_read_base(self):
-self.schema['Personne'].set_action_permissions('read', ('users', 'managers'))
+with self.temporary_permissions(Personne={'read': ('users', 'managers')}):
-cnx = self.login('anon')
+with self.login('anon') as cu:
-cu = cnx.cursor()
+self.assertRaises(Unauthorized,
-self.assertRaises(Unauthorized,
+cu.execute, 'Personne U where U nom "managers"')
-cu.execute, 'Personne U where U nom "managers"')
+self.rollback()
-cnx.close()
 def test_read_erqlexpr_base(self):
 eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
 self.commit()
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+rset = cu.execute('Affaire X')
-rset = cu.execute('Affaire X')
+self.assertEqual(rset.rows, [])
-self.assertEqual(rset.rows, [])
+self.assertRaises(Unauthorized, cu.execute, 'Any X WHERE X eid %(x)s', {'x': eid})
-self.assertRaises(Unauthorized, cu.execute, 'Any X WHERE X eid %(x)s', {'x': eid})
+# cache test
-# cache test
+self.assertRaises(Unauthorized, cu.execute, 'Any X WHERE X eid %(x)s', {'x': eid})
-self.assertRaises(Unauthorized, cu.execute, 'Any X WHERE X eid %(x)s', {'x': eid})
+aff2 = cu.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
-aff2 = cu.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
+soc1 = cu.execute("INSERT Societe X: X nom 'chouette'")[0][0]
-soc1 = cu.execute("INSERT Societe X: X nom 'chouette'")[0][0]
+cu.execute("SET A concerne S WHERE A is Affaire, S is Societe")
-cu.execute("SET A concerne S WHERE A is Affaire, S is Societe")
+self.commit()
-cnx.commit()
+rset = cu.execute('Any X WHERE X eid %(x)s', {'x': aff2})
-rset = cu.execute('Any X WHERE X eid %(x)s', {'x': aff2})
+self.assertEqual(rset.rows, [[aff2]])
-self.assertEqual(rset.rows, [[aff2]])
+# more cache test w/ NOT eid
-# more cache test w/ NOT eid
+rset = cu.execute('Affaire X WHERE NOT X eid %(x)s', {'x': eid})
-rset = cu.execute('Affaire X WHERE NOT X eid %(x)s', {'x': eid})
+self.assertEqual(rset.rows, [[aff2]])
-self.assertEqual(rset.rows, [[aff2]])
+rset = cu.execute('Affaire X WHERE NOT X eid %(x)s', {'x': aff2})
-rset = cu.execute('Affaire X WHERE NOT X eid %(x)s', {'x': aff2})
+self.assertEqual(rset.rows, [])
-self.assertEqual(rset.rows, [])
+# test can't update an attribute of an entity that can't be readen
-# test can't update an attribute of an entity that can't be readen
+self.assertRaises(Unauthorized, cu.execute, 'SET X sujet "hacked" WHERE X eid %(x)s', {'x': eid})
-self.assertRaises(Unauthorized, cu.execute, 'SET X sujet "hacked" WHERE X eid %(x)s', {'x': eid})
+self.rollback()
-cnx.close()
 def test_entity_created_in_transaction(self):
 affschema = self.schema['Affaire']
-origperms = affschema.permissions['read']
+with self.temporary_permissions(Affaire={'read': affschema.permissions['add']}):
-affschema.set_action_permissions('read', affschema.permissions['add'])
+with self.login('iaminusersgrouponly') as cu:
-try:
+aff2 = cu.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
-cnx = self.login('iaminusersgrouponly')
+# entity created in transaction are readable *by eid*
-cu = cnx.cursor()
+self.assertTrue(cu.execute('Any X WHERE X eid %(x)s', {'x':aff2}))
-aff2 = cu.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
+# XXX would be nice if it worked
-# entity created in transaction are readable *by eid*
+rset = cu.execute("Affaire X WHERE X sujet 'cool'")
-self.assertTrue(cu.execute('Any X WHERE X eid %(x)s', {'x':aff2}))
+self.assertEqual(len(rset), 0)
-# XXX would be nice if it worked
+self.assertRaises(Unauthorized, self.commit)
-rset = cu.execute("Affaire X WHERE X sujet 'cool'")
-self.assertEqual(len(rset), 0)
-finally:
-affschema.set_action_permissions('read', origperms)
-cnx.close()
 def test_read_erqlexpr_has_text1(self):
 aff1 = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
 card1 = self.execute("INSERT Card X: X title 'cool'")[0][0]
 self.execute('SET X owned_by U WHERE X eid %(x)s, U login "iaminusersgrouponly"', {'x': card1})
 self.commit()
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+aff2 = cu.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
-aff2 = cu.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
+soc1 = cu.execute("INSERT Societe X: X nom 'chouette'")[0][0]
-soc1 = cu.execute("INSERT Societe X: X nom 'chouette'")[0][0]
+cu.execute("SET A concerne S WHERE A eid %(a)s, S eid %(s)s", {'a': aff2, 's': soc1})
-cu.execute("SET A concerne S WHERE A eid %(a)s, S eid %(s)s", {'a': aff2, 's': soc1})
+self.commit()
-cnx.commit()
+self.assertRaises(Unauthorized, cu.execute, 'Any X WHERE X eid %(x)s', {'x':aff1})
-self.assertRaises(Unauthorized, cu.execute, 'Any X WHERE X eid %(x)s', {'x':aff1})
+self.assertTrue(cu.execute('Any X WHERE X eid %(x)s', {'x':aff2}))
-self.assertTrue(cu.execute('Any X WHERE X eid %(x)s', {'x':aff2}))
+self.assertTrue(cu.execute('Any X WHERE X eid %(x)s', {'x':card1}))
-self.assertTrue(cu.execute('Any X WHERE X eid %(x)s', {'x':card1}))
+rset = cu.execute("Any X WHERE X has_text 'cool'")
-rset = cu.execute("Any X WHERE X has_text 'cool'")
+self.assertEqual(sorted(eid for eid, in rset.rows),
-self.assertEqual(sorted(eid for eid, in rset.rows),
+[card1, aff2])
-[card1, aff2])
+self.rollback()
-cnx.close()
 def test_read_erqlexpr_has_text2(self):
 self.execute("INSERT Personne X: X nom 'bidule'")
 self.execute("INSERT Societe X: X nom 'bidule'")
 self.commit()
-self.schema['Personne'].set_action_permissions('read', ('managers',))
+with self.temporary_permissions(Personne={'read': ('managers',)}):
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+rset = cu.execute('Any N WHERE N has_text "bidule"')
-rset = cu.execute('Any N WHERE N has_text "bidule"')
+self.assertEqual(len(rset.rows), 1, rset.rows)
-self.assertEqual(len(rset.rows), 1, rset.rows)
+rset = cu.execute('Any N WITH N BEING (Any N WHERE N has_text "bidule")')
-rset = cu.execute('Any N WITH N BEING (Any N WHERE N has_text "bidule")')
+self.assertEqual(len(rset.rows), 1, rset.rows)
-self.assertEqual(len(rset.rows), 1, rset.rows)
-cnx.close()
 def test_read_erqlexpr_optional_rel(self):
 self.execute("INSERT Personne X: X nom 'bidule'")
 self.execute("INSERT Societe X: X nom 'bidule'")
 self.commit()
-self.schema['Personne'].set_action_permissions('read', ('managers',))
+with self.temporary_permissions(Personne={'read': ('managers',)}):
-cnx = self.login('anon')
+with self.login('anon') as cu:
-cu = cnx.cursor()
+rset = cu.execute('Any N,U WHERE N has_text "bidule", N owned_by U?')
-rset = cu.execute('Any N,U WHERE N has_text "bidule", N owned_by U?')
+self.assertEqual(len(rset.rows), 1, rset.rows)
-self.assertEqual(len(rset.rows), 1, rset.rows)
-cnx.close()
 def test_read_erqlexpr_aggregat(self):
 self.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
 self.commit()
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+rset = cu.execute('Any COUNT(X) WHERE X is Affaire')
-rset = cu.execute('Any COUNT(X) WHERE X is Affaire')
+self.assertEqual(rset.rows, [[0]])
-self.assertEqual(rset.rows, [[0]])
+aff2 = cu.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
-aff2 = cu.execute("INSERT Affaire X: X sujet 'cool'")[0][0]
+soc1 = cu.execute("INSERT Societe X: X nom 'chouette'")[0][0]
-soc1 = cu.execute("INSERT Societe X: X nom 'chouette'")[0][0]
+cu.execute("SET A concerne S WHERE A is Affaire, S is Societe")
-cu.execute("SET A concerne S WHERE A is Affaire, S is Societe")
+self.commit()
-cnx.commit()
+rset = cu.execute('Any COUNT(X) WHERE X is Affaire')
-rset = cu.execute('Any COUNT(X) WHERE X is Affaire')
+self.assertEqual(rset.rows, [[1]])
-self.assertEqual(rset.rows, [[1]])
+rset = cu.execute('Any ETN, COUNT(X) GROUPBY ETN WHERE X is ET, ET name ETN')
-rset = cu.execute('Any ETN, COUNT(X) GROUPBY ETN WHERE X is ET, ET name ETN')
+values = dict(rset)
-values = dict(rset)
+self.assertEqual(values['Affaire'], 1)
-self.assertEqual(values['Affaire'], 1)
+self.assertEqual(values['Societe'], 2)
-self.assertEqual(values['Societe'], 2)
+rset = cu.execute('Any ETN, COUNT(X) GROUPBY ETN WHERE X is ET, ET name ETN WITH X BEING ((Affaire X) UNION (Societe X))')
-rset = cu.execute('Any ETN, COUNT(X) GROUPBY ETN WHERE X is ET, ET name ETN WITH X BEING ((Affaire X) UNION (Societe X))')
+self.assertEqual(len(rset), 2)
-self.assertEqual(len(rset), 2)
+values = dict(rset)
-values = dict(rset)
+self.assertEqual(values['Affaire'], 1)
-self.assertEqual(values['Affaire'], 1)
+self.assertEqual(values['Societe'], 2)
-self.assertEqual(values['Societe'], 2)
-cnx.close()
 def test_attribute_security(self):
 # only managers should be able to edit the 'test' attribute of Personne entities
 eid = self.execute("INSERT Personne X: X nom 'bidule', X web 'http://www.debian.org', X test TRUE")[0][0]
-self.commit()
 self.execute('SET X test FALSE WHERE X eid %(x)s', {'x': eid})
 self.commit()
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+cu.execute("INSERT Personne X: X nom 'bidule', X web 'http://www.debian.org', X test TRUE")
-cu.execute("INSERT Personne X: X nom 'bidule', X web 'http://www.debian.org', X test TRUE")
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
+cu.execute("INSERT Personne X: X nom 'bidule', X web 'http://www.debian.org', X test FALSE")
-cu.execute("INSERT Personne X: X nom 'bidule', X web 'http://www.debian.org', X test FALSE")
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
+eid = cu.execute("INSERT Personne X: X nom 'bidule', X web 'http://www.debian.org'")[0][0]
-eid = cu.execute("INSERT Personne X: X nom 'bidule', X web 'http://www.debian.org'")[0][0]
+self.commit()
-cnx.commit()
+cu.execute('SET X test FALSE WHERE X eid %(x)s', {'x': eid})
-cu.execute('SET X test FALSE WHERE X eid %(x)s', {'x': eid})
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
+cu.execute('SET X test TRUE WHERE X eid %(x)s', {'x': eid})
-cu.execute('SET X test TRUE WHERE X eid %(x)s', {'x': eid})
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
+cu.execute('SET X web "http://www.logilab.org" WHERE X eid %(x)s', {'x': eid})
-cu.execute('SET X web "http://www.logilab.org" WHERE X eid %(x)s', {'x': eid})
+self.commit()
-cnx.commit()
-cnx.close()
 def test_attribute_security_rqlexpr(self):
 # Note.para attribute editable by managers or if the note is in "todo" state
 note = self.execute("INSERT Note X: X para 'bidule'").get_entity(0, 0)
 self.commit()
 note.cw_adapt_to('IWorkflowable').fire_transition('markasdone')
 self.execute('SET X para "truc" WHERE X eid %(x)s', {'x': note.eid})
 self.commit()
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+cu.execute("SET X para 'chouette' WHERE X eid %(x)s", {'x': note.eid})
-cu.execute("SET X para 'chouette' WHERE X eid %(x)s", {'x': note.eid})
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
+note2 = cu.execute("INSERT Note X: X para 'bidule'").get_entity(0, 0)
-note2 = cu.execute("INSERT Note X: X para 'bidule'").get_entity(0, 0)
+self.commit()
-cnx.commit()
+note2.cw_adapt_to('IWorkflowable').fire_transition('markasdone')
-note2.cw_adapt_to('IWorkflowable').fire_transition('markasdone')
+self.commit()
-cnx.commit()
+self.assertEqual(len(cu.execute('Any X WHERE X in_state S, S name "todo", X eid %(x)s', {'x': note2.eid})),
-self.assertEqual(len(cu.execute('Any X WHERE X in_state S, S name "todo", X eid %(x)s', {'x': note2.eid})),
+0)
-0)
+cu.execute("SET X para 'chouette' WHERE X eid %(x)s", {'x': note2.eid})
-cu.execute("SET X para 'chouette' WHERE X eid %(x)s", {'x': note2.eid})
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
+note2.cw_adapt_to('IWorkflowable').fire_transition('redoit')
-note2.cw_adapt_to('IWorkflowable').fire_transition('redoit')
+self.commit()
-cnx.commit()
+cu.execute("SET X para 'chouette' WHERE X eid %(x)s", {'x': note2.eid})
-cu.execute("SET X para 'chouette' WHERE X eid %(x)s", {'x': note2.eid})
+self.commit()
-cnx.commit()
-cnx.close()
 def test_attribute_read_security(self):
 # anon not allowed to see users'login, but they can see users
-self.repo.schema['CWUser'].set_action_permissions('read', ('guests', 'users', 'managers'))
+login_rdef = self.repo.schema['CWUser'].rdef('login')
-self.repo.schema['CWUser'].rdef('login').set_action_permissions('read', ('users', 'managers'))
+with self.temporary_permissions((login_rdef, {'read': ('users', 'managers')}),
-cnx = self.login('anon')
+CWUser={'read': ('guests', 'users', 'managers')}):
-cu = cnx.cursor()
+with self.login('anon') as cu:
 rset = cu.execute('CWUser X')
 self.assertTrue(rset)
 x = rset.get_entity(0, 0)
 self.assertEqual(x.login, None)
 self.assertTrue(x.creation_date)
 x = rset.get_entity(1, 0)
 x.complete()
 self.assertEqual(x.login, None)
 self.assertTrue(x.creation_date)
-cnx.rollback()
-cnx.close()
 def test_yams_inheritance_and_security_bug(self):
-oldperms = self.schema['Division'].permissions
+with self.temporary_permissions(Division={'read': ('managers', ERQLExpression('X owned_by U'))}):
-try:
+with self.login('iaminusersgrouponly'):
-self.schema['Division'].permissions = {
+querier = self.repo.querier
-'read': ('managers', ERQLExpression('X owned_by U')),
+rqlst = querier.parse('Any X WHERE X is_instance_of Societe')
-'add': ('managers', 'users'),
+querier.solutions(self.session, rqlst, {})
-'update': ('managers', 'owners'),
+querier._annotate(rqlst)
-'delete': ('managers', 'owners')}
+plan = querier.plan_factory(rqlst, {}, self.session)
-self.login('iaminusersgrouponly')
+plan.preprocess(rqlst)
-querier = self.repo.querier
+self.assertEqual(
-rqlst = querier.parse('Any X WHERE X is_instance_of Societe')
+rqlst.as_string(),
-querier.solutions(self.session, rqlst, {})
+'(Any X WHERE X is IN(SubDivision, Societe)) UNION (Any X WHERE X is Division, EXISTS(X owned_by %(B)s))')
-querier._annotate(rqlst)
-plan = querier.plan_factory(rqlst, {}, self.session)
-plan.preprocess(rqlst)
-self.assertEqual(
-rqlst.as_string(),
-'(Any X WHERE X is IN(SubDivision, Societe)) UNION (Any X WHERE X is Division, EXISTS(X owned_by %(B)s))')
-finally:
-self.schema['Division'].permissions = oldperms
 class BaseSchemaSecurityTC(BaseSecurityTC):
 """tests related to the base schema permission configuration"""
 def test_user_can_delete_object_he_created(self):
 # even if some other user have changed object'state
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+# due to security test, affaire has to concerne a societe the user owns
-# due to security test, affaire has to concerne a societe the user owns
+cu.execute('INSERT Societe X: X nom "ARCTIA"')
-cu.execute('INSERT Societe X: X nom "ARCTIA"')
+cu.execute('INSERT Affaire X: X ref "ARCT01", X concerne S WHERE S nom "ARCTIA"')
-cu.execute('INSERT Affaire X: X ref "ARCT01", X concerne S WHERE S nom "ARCTIA"')
+self.commit()
-cnx.commit()
-self.restore_connection()
 affaire = self.execute('Any X WHERE X ref "ARCT01"').get_entity(0, 0)
 affaire.cw_adapt_to('IWorkflowable').fire_transition('abort')
 self.commit()
 self.assertEqual(len(self.execute('TrInfo X WHERE X wf_info_for A, A ref "ARCT01"')),
 1)
 self.assertEqual(len(self.execute('TrInfo X WHERE X wf_info_for A, A ref "ARCT01",'
 'X owned_by U, U login "admin"')),
 1) # TrInfo at the above state change
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+cu.execute('DELETE Affaire X WHERE X ref "ARCT01"')
-cu.execute('DELETE Affaire X WHERE X ref "ARCT01"')
+self.commit()
-cnx.commit()
+self.assertFalse(cu.execute('Affaire X'))
-self.assertFalse(cu.execute('Affaire X'))
-cnx.close()
 def test_users_and_groups_non_readable_by_guests(self):
-cnx = self.login('anon')
+with self.login('anon') as cu:
-anon = cnx.user(self.session)
+anon = cu.connection.user(self.session)
-cu = cnx.cursor()
+# anonymous user can only read itself
-# anonymous user can only read itself
+rset = cu.execute('Any L WHERE X owned_by U, U login L')
-rset = cu.execute('Any L WHERE X owned_by U, U login L')
+self.assertEqual(rset.rows, [['anon']])
-self.assertEqual(rset.rows, [['anon']])
+rset = cu.execute('CWUser X')
-rset = cu.execute('CWUser X')
+self.assertEqual(rset.rows, [[anon.eid]])
-self.assertEqual(rset.rows, [[anon.eid]])
+# anonymous user can read groups (necessary to check allowed transitions for instance)
-# anonymous user can read groups (necessary to check allowed transitions for instance)
+self.assert_(cu.execute('CWGroup X'))
-self.assert_(cu.execute('CWGroup X'))
+# should only be able to read the anonymous user, not another one
-# should only be able to read the anonymous user, not another one
+origuser = self.adminsession.user
-origuser = self.adminsession.user
+self.assertRaises(Unauthorized,
-self.assertRaises(Unauthorized,
+cu.execute, 'CWUser X WHERE X eid %(x)s', {'x': origuser.eid})
-cu.execute, 'CWUser X WHERE X eid %(x)s', {'x': origuser.eid})
+# nothing selected, nothing updated, no exception raised
-# nothing selected, nothing updated, no exception raised
+#self.assertRaises(Unauthorized,
-#self.assertRaises(Unauthorized,
+#                  cu.execute, 'SET X login "toto" WHERE X eid %(x)s',
-#                  cu.execute, 'SET X login "toto" WHERE X eid %(x)s',
+#                  {'x': self.user.eid})
-#                  {'x': self.user.eid})
+rset = cu.execute('CWUser X WHERE X eid %(x)s', {'x': anon.eid})
-rset = cu.execute('CWUser X WHERE X eid %(x)s', {'x': anon.eid})
+self.assertEqual(rset.rows, [[anon.eid]])
-self.assertEqual(rset.rows, [[anon.eid]])
+# but can't modify it
-# but can't modify it
+cu.execute('SET X login "toto" WHERE X eid %(x)s', {'x': anon.eid})
-cu.execute('SET X login "toto" WHERE X eid %(x)s', {'x': anon.eid})
+self.assertRaises(Unauthorized, self.commit)
-self.assertRaises(Unauthorized, cnx.commit)
-cnx.close()
 def test_in_group_relation(self):
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+rql = u"DELETE U in_group G WHERE U login 'admin'"
-rql = u"DELETE U in_group G WHERE U login 'admin'"
+self.assertRaises(Unauthorized, cu.execute, rql)
-self.assertRaises(Unauthorized, cu.execute, rql)
+rql = u"SET U in_group G WHERE U login 'admin', G name 'users'"
-rql = u"SET U in_group G WHERE U login 'admin', G name 'users'"
+self.assertRaises(Unauthorized, cu.execute, rql)
-self.assertRaises(Unauthorized, cu.execute, rql)
+self.rollback()
-cnx.close()
 def test_owned_by(self):
 self.execute("INSERT Personne X: X nom 'bidule'")
 self.commit()
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
-cu = cnx.cursor()
+rql = u"SET X owned_by U WHERE U login 'iaminusersgrouponly', X is Personne"
-rql = u"SET X owned_by U WHERE U login 'iaminusersgrouponly', X is Personne"
+self.assertRaises(Unauthorized, cu.execute, rql)
-self.assertRaises(Unauthorized, cu.execute, rql)
+self.rollback()
-cnx.close()
 def test_bookmarked_by_guests_security(self):
 beid1 = self.execute('INSERT Bookmark B: B path "?vid=manage", B title "manage"')[0][0]
 beid2 = self.execute('INSERT Bookmark B: B path "?vid=index", B title "index", B bookmarked_by U WHERE U login "anon"')[0][0]
 self.commit()
-cnx = self.login('anon')
+with self.login('anon') as cu:
-cu = cnx.cursor()
+anoneid = self.session.user.eid
-anoneid = self.session.user.eid
+self.assertEqual(cu.execute('Any T,P ORDERBY lower(T) WHERE B is Bookmark,B title T,B path P,'
-self.assertEqual(cu.execute('Any T,P ORDERBY lower(T) WHERE B is Bookmark,B title T,B path P,'
+'B bookmarked_by U, U eid %s' % anoneid).rows,
-'B bookmarked_by U, U eid %s' % anoneid).rows,
+[['index', '?vid=index']])
-[['index', '?vid=index']])
+self.assertEqual(cu.execute('Any T,P ORDERBY lower(T) WHERE B is Bookmark,B title T,B path P,'
-self.assertEqual(cu.execute('Any T,P ORDERBY lower(T) WHERE B is Bookmark,B title T,B path P,'
+'B bookmarked_by U, U eid %(x)s', {'x': anoneid}).rows,
-'B bookmarked_by U, U eid %(x)s', {'x': anoneid}).rows,
+[['index', '?vid=index']])
-[['index', '?vid=index']])
+# can read others bookmarks as well
-# can read others bookmarks as well
+self.assertEqual(cu.execute('Any B where B is Bookmark, NOT B bookmarked_by U').rows,
-self.assertEqual(cu.execute('Any B where B is Bookmark, NOT B bookmarked_by U').rows,
+[[beid1]])
-[[beid1]])
+self.assertRaises(Unauthorized, cu.execute,'DELETE B bookmarked_by U')
-self.assertRaises(Unauthorized, cu.execute,'DELETE B bookmarked_by U')
+self.assertRaises(Unauthorized,
-self.assertRaises(Unauthorized,
+cu.execute, 'SET B bookmarked_by U WHERE U eid %(x)s, B eid %(b)s',
-cu.execute, 'SET B bookmarked_by U WHERE U eid %(x)s, B eid %(b)s',
+{'x': anoneid, 'b': beid1})
-{'x': anoneid, 'b': beid1})
+self.rollback()
-cnx.close()
 def test_ambigous_ordered(self):
-cnx = self.login('anon')
+with self.login('anon') as cu:
-cu = cnx.cursor()
+names = [t for t, in cu.execute('Any N ORDERBY lower(N) WHERE X name N')]
-names = [t for t, in cu.execute('Any N ORDERBY lower(N) WHERE X name N')]
+self.assertEqual(names, sorted(names, key=lambda x: x.lower()))
-self.assertEqual(names, sorted(names, key=lambda x: x.lower()))
-cnx.close()
 def test_restrict_is_instance_ok(self):
-from rql import RQLException
 rset = self.execute('Any X WHERE X is_instance_of BaseTransition')
 rqlst = rset.syntax_tree()
 select = rqlst.children[0]
 x = select.get_selected_variables().next()
 self.assertRaises(RQLException, select.add_type_restriction,
 """check a user change in_state without having update permission on the
 subject
 """
 eid = self.execute('INSERT Affaire X: X ref "ARCT01"')[0][0]
 self.commit()
-cnx = self.login('iaminusersgrouponly')
+with self.login('iaminusersgrouponly') as cu:
 session = self.session
 # needed to avoid check_perm error
 session.set_cnxset()
 # needed to remove rql expr granting update perm to the user
-affaire_perms = self.schema['Affaire'].permissions.copy()
+affschema = self.schema['Affaire']
-self.schema['Affaire'].set_action_permissions('update', self.schema['Affaire'].get_groups('update'))
+with self.temporary_permissions(Affaire={'update': affschema.get_groups('update'),
-try:
+'read': ('users',)}):
 self.assertRaises(Unauthorized,
-self.schema['Affaire'].check_perm, session, 'update', eid=eid)
+affschema.check_perm, session, 'update', eid=eid)
-cu = cnx.cursor()
+aff = cu.execute('Any X WHERE X ref "ARCT01"').get_entity(0, 0)
-self.schema['Affaire'].set_action_permissions('read', ('users',))
+aff.cw_adapt_to('IWorkflowable').fire_transition('abort')
-aff = cu.execute('Any X WHERE X ref "ARCT01"').get_entity(0, 0)
+self.commit()
-aff.cw_adapt_to('IWorkflowable').fire_transition('abort')
+# though changing a user state (even logged user) is reserved to managers
-cnx.commit()
+user = self.user(session)
-# though changing a user state (even logged user) is reserved to managers
+session.set_cnxset()
-user = cnx.user(self.session)
+# XXX wether it should raise Unauthorized or ValidationError is not clear
-# XXX wether it should raise Unauthorized or ValidationError is not clear
+# the best would probably ValidationError if the transition doesn't exist
-# the best would probably ValidationError if the transition doesn't exist
+# from the current state but Unauthorized if it exists but user can't pass it
-# from the current state but Unauthorized if it exists but user can't pass it
+self.assertRaises(ValidationError,
-self.assertRaises(ValidationError,
+user.cw_adapt_to('IWorkflowable').fire_transition, 'deactivate')
-user.cw_adapt_to('IWorkflowable').fire_transition, 'deactivate')
+self.rollback() # else will fail on login cm exit
-finally:
-# restore orig perms
-for action, perms in affaire_perms.iteritems():
-self.schema['Affaire'].set_action_permissions(action, perms)
-cnx.close()
 def test_trinfo_security(self):
 aff = self.execute('INSERT Affaire X: X ref "ARCT01"').get_entity(0, 0)
 iworkflowable = aff.cw_adapt_to('IWorkflowable')
 self.commit()

branch	stable
changeset 8461	8af7c6d86efb
parent 8452	1ad42383a9ec
child 8463	a964c40adbe3