From 2b4098f1d27d92b6d0debbf2d3fa7fc26b260f36 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9dric=20Krier?= Date: Sat, 22 Jan 2011 23:56:13 +0100 Subject: [PATCH] Add model field access --- CHANGELOG | 1 + trytond/ir/model.py | 165 ++++++- trytond/ir/model.xml | 57 +++ trytond/model/browse.py | 17 +- trytond/model/model.py | 6 +- trytond/model/modelstorage.py | 10 + trytond/model/modelview.py | 36 ++ trytond/test/__init__.py | 1 + trytond/test/access.py | 14 + trytond/tests/test_access.py | 794 ++++++++++++++++++++++++++++++++++ trytond/tests/test_tryton.py | 2 + 11 files changed, 1076 insertions(+), 27 deletions(-) create mode 100644 trytond/test/access.py create mode 100644 trytond/tests/test_access.py diff --git a/CHANGELOG b/CHANGELOG index 2f8941a9ad..ed865f4eb4 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,4 @@ +* Add model field access * Add loading attribute on fields * Remove priority attribute on fields * Model doesn't convert anymore ids for inherited methods diff --git a/trytond/ir/model.py b/trytond/ir/model.py index f41a518e5f..bb36f8df80 100644 --- a/trytond/ir/model.py +++ b/trytond/ir/model.py @@ -235,38 +235,36 @@ class ModelAccess(ModelSQL, ModelView): @Cache('ir_model_access.check') def check(self, model_name, mode='read', raise_exception=True): + ''' + Check access for model_name + + :param model_name: the model name + :param mode: 'read', 'write', 'create' or 'delete' + :param raise_exception: raise an exception if the test failed + + :return: a boolean + ''' assert mode in ['read', 'write', 'create', 'delete'], \ 'Invalid access mode for security' - model_obj = self.pool.get(model_name) - if hasattr(model_obj, 'table_query') \ - and model_obj.table_query(): - return False if Transaction().user == 0: return True + ir_model_obj = self.pool.get('ir.model') user_group_obj = self.pool.get('res.user-res.group') cursor = Transaction().cursor - cursor.execute('SELECT MAX(CASE WHEN a.perm_'+mode+' THEN 1 else 0 END) ' - 'FROM ir_model_access a ' - 'JOIN "' + ir_model_obj._table + '" m ' - 'ON (a.model = m.id) ' - 'JOIN "' + user_group_obj._table + '" gu ' - 'ON (gu.gid = a."group") ' - 'WHERE m.model = %s AND gu.uid = %s', - (model_name, Transaction().user,)) - row = cursor.fetchall() - if row[0][0] is None: - cursor.execute('SELECT ' \ - 'MAX(CASE WHEN perm_' + mode + ' THEN 1 else 0 END) ' \ - 'FROM ir_model_access a ' \ - 'JOIN ir_model m ' \ - 'ON (a.model = m.id) ' \ - 'WHERE a."group" IS NULL AND m.model = %s', (model_name,)) - row = cursor.fetchall() - if row[0][0] is None: - return True - if not row[0][0]: + cursor.execute('SELECT MAX(CASE WHEN a.perm_%s THEN 1 ELSE 0 END) ' + 'FROM "%s" AS a ' + 'JOIN "%s" AS m ' + 'ON (a.model = m.id) ' + 'LEFT JOIN "%s" AS gu ' + 'ON (gu.gid = a."group") ' + 'WHERE m.model = %%s AND (gu.uid = %%s OR a."group" IS NULL)' + % (mode, self._table, ir_model_obj._table, + user_group_obj._table), + (model_name, Transaction().user)) + access, = cursor.fetchone() + if not access and access is not None: if raise_exception: self.raise_user_error(mode, model_name) else: @@ -309,6 +307,125 @@ class ModelAccess(ModelSQL, ModelView): ModelAccess() +class ModelFieldAccess(ModelSQL, ModelView): + "Model Field Access" + _name = 'ir.model.field.access' + _description = __doc__ + _rec_name = 'field' + field = fields.Many2One('ir.model.field', 'Field', required=True, + ondelete='CASCADE') + group = fields.Many2One('res.group', 'Group', ondelete='CASCADE') + perm_read = fields.Boolean('Read Access') + perm_write = fields.Boolean('Write Access') + description = fields.Text('Description') + + def __init__(self): + super(ModelFieldAccess, self).__init__() + self._sql_constraints += [ + ('field_group_uniq', 'UNIQUE("field", "group")', + 'Only one record by field and group is allowed!'), + ] + self._error_messages.update({ + 'read': 'You can not read the field! (%s.%s)', + 'write': 'You can not write on the field! (%s.%s)', + }) + + def check_xml_record(self, ids, values): + return True + + def default_perm_read(self): + return False + + def default_perm_write(self): + return False + + @Cache('ir_model_field_access.check') + def check(self, model_name, fields, mode='read', raise_exception=True, + access=False): + ''' + Check access for fields on model_name. + + :param model_name: the model name + :param fields: a list of fields + :param mode: 'read' or 'write' + :param raise_exception: raise an exception if the test failed + :param access: return a dictionary with access right instead of boolean + + :return: a boolean + ''' + assert mode in ('read', 'write'), 'Invalid access mode' + if Transaction().user == 0: + if access: + return dict((x, True) for x in fields) + return True + + ir_model_obj = self.pool.get('ir.model') + ir_model_field_obj = self.pool.get('ir.model.field') + user_group_obj = self.pool.get('res.user-res.group') + + cursor = Transaction().cursor + + cursor.execute('SELECT f.name, ' + 'MAX(CASE WHEN a.perm_%s THEN 1 ELSE 0 END) ' + 'FROM "%s" AS a ' + 'JOIN "%s" AS f ' + 'ON (a.field = f.id) ' + 'JOIN "%s" AS m ' + 'ON (f.model = m.id) ' + 'LEFT JOIN "%s" AS gu ' + 'ON (gu.gid = a."group") ' + 'WHERE m.model = %%s AND (gu.uid = %%s OR a."group" IS NULL) ' + 'GROUP BY f.name' + % (mode, self._table, ir_model_field_obj._table, + ir_model_obj._table, user_group_obj._table), + (model_name, Transaction().user)) + accesses = dict(cursor.fetchall()) + if access: + return accesses + for field in fields: + if not accesses.get(field, True): + if raise_exception: + self.raise_user_error(mode, (model_name, field)) + else: + return False + return True + + def write(self, ids, vals): + res = super(ModelFieldAccess, self).write(ids, vals) + # Restart the cache + self.check.reset() + for _, model in self.pool.iterobject(): + try: + model.fields_view_get.reset() + except Exception: + pass + return res + + def create(self, vals): + res = super(ModelFieldAccess, self).create(vals) + # Restart the cache + self.check.reset() + for _, model in self.pool.iterobject(): + try: + model.fields_view_get.reset() + except Exception: + pass + return res + + def delete(self, ids): + res = super(ModelFieldAccess, self).delete(ids) + # Restart the cache + self.check.reset() + for _, model in self.pool.iterobject(): + try: + model.fields_view_get.reset() + except Exception: + pass + return res + +ModelFieldAccess() + + class ModelData(ModelSQL, ModelView): "Model data" _name = 'ir.model.data' diff --git a/trytond/ir/model.xml b/trytond/ir/model.xml index 3b9eecb29e..56784aa845 100644 --- a/trytond/ir/model.xml +++ b/trytond/ir/model.xml @@ -190,6 +190,63 @@ this repository contains the full copyright notices and license terms. --> + + ir.model.field.access + tree + + + + + + + + ]]> + + + + + ir.model.field.access + form + + + + + + Fields Access + ir.model.field.access + form + + + + + + + + + + + + + Graph ir.model diff --git a/trytond/model/browse.py b/trytond/model/browse.py index 1a074d8ee4..a23ac30eb4 100644 --- a/trytond/model/browse.py +++ b/trytond/model/browse.py @@ -90,13 +90,26 @@ class BrowseRecord(object): % (name, self._model._name)) if col.loading == 'eager': + field_access_obj = self._model.pool.get('ir.model.field.access') + fread_accesses = {} + for inherit_name in self._model._inherits: + inherit_obj = self._model.pool.get(inherit_name) + fread_accesses.update(field_access_obj.check(inherit_name, + inherit_obj._columns.keys(), 'read', access=True)) + fread_accesses.update(field_access_obj.check(self._model._name, + self._model._columns.keys(), 'read', access=True)) + to_remove = set(x for x, y in fread_accesses.iteritems() + if not y and x != name) + ffields = dict((fname, field) for fname, (_, _, field) in self._model._inherit_fields.iteritems() if field.loading == 'eager' - and fname not in self._model._columns) + and fname not in self._model._columns + and fname not in to_remove) ffields.update(dict((fname, field) for fname, field in self._model._columns.iteritems() - if field.loading == 'eager')) + if field.loading == 'eager' + and fname not in to_remove)) else: ffields = {name: col} diff --git a/trytond/model/model.py b/trytond/model/model.py index 699a57a68f..cc9538feaa 100644 --- a/trytond/model/model.py +++ b/trytond/model/model.py @@ -619,6 +619,8 @@ class Model(object): res = {} translation_obj = self.pool.get('ir.translation') model_access_obj = self.pool.get('ir.model.access') + field_access_obj = self.pool.get('ir.model.field.access') + for parent in self._inherits: res.update(self.pool.get(parent).fields_get(fields_names)) write_access = model_access_obj.check(self._name, 'write', @@ -648,6 +650,8 @@ class Model(object): encoder = PYSONEncoder() + fwrite_accesses = field_access_obj.check(self._name, fields_names or + self._columns.keys(), 'write', access=True) for field in (x for x in self._columns.keys() if ((not fields_names) or x in fields_names)): res[field] = {'type': self._columns[field]._type} @@ -671,7 +675,7 @@ class Model(object): if getattr(self._columns[field], arg, None) != None: res[field][arg] = copy.copy(getattr(self._columns[field], arg)) - if not write_access: + if not write_access or not fwrite_accesses.get(field, True): res[field]['readonly'] = True if res[field].get('states') and \ 'readonly' in res[field]['states']: diff --git a/trytond/model/modelstorage.py b/trytond/model/modelstorage.py index 2d52984b11..5ce8e3bdb3 100644 --- a/trytond/model/modelstorage.py +++ b/trytond/model/modelstorage.py @@ -82,7 +82,11 @@ class ModelStorage(Model): :return: the id of the created record ''' model_access_obj = self.pool.get('ir.model.access') + model_field_access_obj = self.pool.get('ir.model.field.access') + model_access_obj.check(self._name, 'create') + model_field_access_obj.check(self._name, + [x for x in values if x in self._columns], 'write') self.__clean_xxx2many_cache() return False @@ -113,8 +117,11 @@ class ModelStorage(Model): and fields value as value. The list will not be in the same order. ''' model_access_obj = self.pool.get('ir.model.access') + model_field_access_obj = self.pool.get('ir.model.field.access') model_access_obj.check(self._name, 'read') + model_field_access_obj.check(self._name, + fields_names or self._columns.keys(), 'read') if isinstance(ids, (int, long)): return {} return [] @@ -129,8 +136,11 @@ class ModelStorage(Model): :return: True if succeed ''' model_access_obj = self.pool.get('ir.model.access') + model_field_access_obj = self.pool.get('ir.model.field.access') model_access_obj.check(self._name, 'write') + model_field_access_obj.check(self._name, + [x for x in values if x in self._columns], 'write') if not self.check_xml_record(ids, values): self.raise_user_error('write_xml_record', error_description='xml_record_desc') diff --git a/trytond/model/modelview.py b/trytond/model/modelview.py index a4292bf484..ec89bd1f84 100644 --- a/trytond/model/modelview.py +++ b/trytond/model/modelview.py @@ -256,9 +256,45 @@ class ModelView(Model): return value def _view_look_dom_arch(self, tree, type): + field_access_obj = self.pool.get('ir.model.field.access') + fields_width = {} tree_root = tree.getroottree().getroot() + # Find field without read access + fread_accesses = field_access_obj.check(self._name, + self._columns.keys(), 'read', access=True) + fields_to_remove = list(x for x, y in fread_accesses.iteritems() + if not y) + for name, field in self._columns.iteritems(): + for field_to_remove in fields_to_remove: + if field_to_remove in field.depends: + fields_to_remove.append(name) + + # Find field inherited without read access + for inherit_name in self._inherits: + inherit_obj = self.pool.get(inherit_name) + fread_accesses = field_access_obj.check(inherit_obj._name, + inherit_obj._columns.keys(), 'read', access=True) + fields_to_remove += list(x for x, y in fread_accesses.iteritems() + if not y and x not in self._columns.keys()) + for name, field in inherit_obj._columns.iteritems(): + for field_to_remove in fields_to_remove: + if field_to_remove in field.depends: + fields_to_remove.append(name) + + # Remove field without read access + for field in fields_to_remove: + for element in tree.xpath( + '//field[@name="%s"] | //label[@name="%s"]' + % (field, field)): + if type == 'form': + element.tag = 'label' + element.attrib.clear() + elif type == 'tree': + parent = element.getparent() + parent.remove(element) + if type == 'tree': viewtreewidth_obj = self.pool.get('ir.ui.view_tree_width') viewtreewidth_ids = viewtreewidth_obj.search([ diff --git a/trytond/test/__init__.py b/trytond/test/__init__.py index 0af27ec012..0325085a89 100644 --- a/trytond/test/__init__.py +++ b/trytond/test/__init__.py @@ -7,3 +7,4 @@ from mptt import * from import_data import * from export_data import * from trigger import * +from access import * diff --git a/trytond/test/access.py b/trytond/test/access.py new file mode 100644 index 0000000000..9a64c16e45 --- /dev/null +++ b/trytond/test/access.py @@ -0,0 +1,14 @@ +#This file is part of Tryton. The COPYRIGHT file at the top level of +#this repository contains the full copyright notices and license terms. +from trytond.model import ModelSQL, fields + + +class TestAccess(ModelSQL): + 'Test Access' + _name = 'test.access' + _description = __doc__ + + field1 = fields.Char('Field 1') + field2 = fields.Char('Field 2') + +TestAccess() diff --git a/trytond/tests/test_access.py b/trytond/tests/test_access.py new file mode 100644 index 0000000000..7a21a2521a --- /dev/null +++ b/trytond/tests/test_access.py @@ -0,0 +1,794 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +#This file is part of Tryton. The COPYRIGHT file at the top level of +#this repository contains the full copyright notices and license terms. +from __future__ import with_statement +import unittest +from trytond.tests.test_tryton import POOL, DB_NAME, USER, CONTEXT, \ + install_module +from trytond.transaction import Transaction + + +class ModelAccessTestCase(unittest.TestCase): + ''' + Test Model Access + ''' + + def setUp(self): + install_module('test') + self.model_access = POOL.get('ir.model.access') + self.test_access = POOL.get('test.access') + self.model = POOL.get('ir.model') + self.group = POOL.get('res.group') + + def test0010perm_read(self): + ''' + Test Read Access + ''' + with Transaction().start(DB_NAME, USER, CONTEXT) as transaction: + model_id, = self.model.search([('model', '=', 'test.access')]) + + test_id = self.test_access.create({}) + + # Without model access + self.test_access.read(test_id) + + # With model access + + # One access allowed for any group + model_access_wo_group_id = self.model_access.create({ + 'model': model_id, + 'group': False, + 'perm_read': True, + }) + self.test_access.read(test_id) + + # One access disallowed for any group + self.model_access.write(model_access_wo_group_id, { + 'perm_read': False, + }) + self.failUnlessRaises(Exception, self.test_access.read, + test_id) + + # Two access rules with one group allowed + group_id = self.group.search([('users', '=', USER)])[0] + model_access_w_group_id = self.model_access.create({ + 'model': model_id, + 'group': group_id, + 'perm_read': True, + }) + + self.test_access.read(test_id) + + # Two access rules with both allowed + self.model_access.write(model_access_wo_group_id, { + 'perm_read': True, + }) + self.test_access.read(test_id) + + # Two access rules with any group allowed + self.model_access.write(model_access_w_group_id, { + 'perm_read': False, + }) + self.test_access.read(test_id) + + # Two access rules with both disallowed + self.model_access.write(model_access_wo_group_id, { + 'perm_read': False, + }) + self.failUnlessRaises(Exception, self.test_access.read, + test_id) + + # One access disallowed for one group + self.model_access.delete(model_access_wo_group_id) + self.failUnlessRaises(Exception, self.test_access.read, + test_id) + + # One access allowed for one group + self.model_access.write(model_access_w_group_id, { + 'perm_read': True, + }) + self.test_access.read(test_id) + + # One access allowed for one other group + group_id = self.group.create({'name': 'Test'}) + self.model_access.write(model_access_w_group_id, { + 'group': group_id, + }) + self.test_access.read(test_id) + + # One access disallowed for one other group + self.model_access.write(model_access_w_group_id, { + 'perm_read': False, + }) + self.test_access.read(test_id) + + transaction.cursor.rollback() + + def test0020perm_write(self): + ''' + Test Write Access + ''' + with Transaction().start(DB_NAME, USER, CONTEXT) as transaction: + model_id, = self.model.search([('model', '=', 'test.access')]) + + test_id = self.test_access.create({}) + + # Without model access + self.test_access.write(test_id, {}) + + # With model access + + # One access allowed for any group + model_access_wo_group_id = self.model_access.create({ + 'model': model_id, + 'group': False, + 'perm_write': True, + }) + self.test_access.write(test_id, {}) + + # One access disallowed for any group + self.model_access.write(model_access_wo_group_id, { + 'perm_write': False, + }) + self.failUnlessRaises(Exception, self.test_access.write, + test_id, {}) + + # Two access rules with one group allowed + group_id = self.group.search([('users', '=', USER)])[0] + model_access_w_group_id = self.model_access.create({ + 'model': model_id, + 'group': group_id, + 'perm_write': True, + }) + + self.test_access.write(test_id, {}) + + # Two access rules with both allowed + self.model_access.write(model_access_wo_group_id, { + 'perm_write': True, + }) + self.test_access.write(test_id, {}) + + # Two access rules with any group allowed + self.model_access.write(model_access_w_group_id, { + 'perm_write': False, + }) + self.test_access.write(test_id, {}) + + # Two access rules with both disallowed + self.model_access.write(model_access_wo_group_id, { + 'perm_write': False, + }) + self.failUnlessRaises(Exception, self.test_access.write, + test_id, {}) + + # One access disallowed for one group + self.model_access.delete(model_access_wo_group_id) + self.failUnlessRaises(Exception, self.test_access.write, + test_id, {}) + + # One access allowed for one group + self.model_access.write(model_access_w_group_id, { + 'perm_write': True, + }) + self.test_access.write(test_id, {}) + + # One access allowed for one other group + group_id = self.group.create({'name': 'Test'}) + self.model_access.write(model_access_w_group_id, { + 'group': group_id, + }) + self.test_access.write(test_id, {}) + + # One access disallowed for one other group + self.model_access.write(model_access_w_group_id, { + 'perm_write': False, + }) + self.test_access.write(test_id, {}) + + transaction.cursor.rollback() + + def test0030perm_create(self): + ''' + Test Create Access + ''' + with Transaction().start(DB_NAME, USER, CONTEXT) as transaction: + model_id, = self.model.search([('model', '=', 'test.access')]) + + # Without model access + test_id = self.test_access.create({}) + + # With model access + + # One access allowed for any group + model_access_wo_group_id = self.model_access.create({ + 'model': model_id, + 'group': False, + 'perm_create': True, + }) + self.test_access.create({}) + + # One access disallowed for any group + self.model_access.write(model_access_wo_group_id, { + 'perm_create': False, + }) + self.failUnlessRaises(Exception, self.test_access.create, {}) + + # Two access rules with one group allowed + group_id = self.group.search([('users', '=', USER)])[0] + model_access_w_group_id = self.model_access.create({ + 'model': model_id, + 'group': group_id, + 'perm_create': True, + }) + + self.test_access.create({}) + + # Two access rules with both allowed + self.model_access.write(model_access_wo_group_id, { + 'perm_create': True, + }) + self.test_access.create({}) + + # Two access rules with any group allowed + self.model_access.write(model_access_w_group_id, { + 'perm_create': False, + }) + self.test_access.create({}) + + # Two access rules with both disallowed + self.model_access.write(model_access_wo_group_id, { + 'perm_create': False, + }) + self.failUnlessRaises(Exception, self.test_access.create, {}) + + # One access disallowed for one group + self.model_access.delete(model_access_wo_group_id) + self.failUnlessRaises(Exception, self.test_access.create, {}) + + # One access allowed for one group + self.model_access.write(model_access_w_group_id, { + 'perm_create': True, + }) + self.test_access.create({}) + + # One access allowed for one other group + group_id = self.group.create({'name': 'Test'}) + self.model_access.write(model_access_w_group_id, { + 'group': group_id, + }) + self.test_access.create({}) + + # One access disallowed for one other group + self.model_access.write(model_access_w_group_id, { + 'perm_create': False, + }) + self.test_access.create({}) + + transaction.cursor.rollback() + + def test0040perm_delete(self): + ''' + Test Delete Access + ''' + with Transaction().start(DB_NAME, USER, CONTEXT) as transaction: + model_id, = self.model.search([('model', '=', 'test.access')]) + + test_ids = [self.test_access.create({}) for x in range(11)] + + # Without model access + test_id = self.test_access.delete(test_ids.pop()) + + # With model access + + # One access allowed for any group + model_access_wo_group_id = self.model_access.create({ + 'model': model_id, + 'group': False, + 'perm_delete': True, + }) + self.test_access.delete(test_ids.pop()) + + # One access disallowed for any group + self.model_access.write(model_access_wo_group_id, { + 'perm_delete': False, + }) + self.failUnlessRaises(Exception, self.test_access.delete, + test_ids.pop()) + + # Two access rules with one group allowed + group_id = self.group.search([('users', '=', USER)])[0] + model_access_w_group_id = self.model_access.create({ + 'model': model_id, + 'group': group_id, + 'perm_delete': True, + }) + + self.test_access.delete(test_ids.pop()) + + # Two access rules with both allowed + self.model_access.write(model_access_wo_group_id, { + 'perm_delete': True, + }) + self.test_access.delete(test_ids.pop()) + + # Two access rules with any group allowed + self.model_access.write(model_access_w_group_id, { + 'perm_delete': False, + }) + self.test_access.delete(test_ids.pop()) + + # Two access rules with both disallowed + self.model_access.write(model_access_wo_group_id, { + 'perm_delete': False, + }) + self.failUnlessRaises(Exception, self.test_access.delete, + test_ids.pop()) + + # One access disallowed for one group + self.model_access.delete(model_access_wo_group_id) + self.failUnlessRaises(Exception, self.test_access.delete, + test_ids.pop()) + + # One access allowed for one group + self.model_access.write(model_access_w_group_id, { + 'perm_delete': True, + }) + self.test_access.delete(test_ids.pop()) + + # One access allowed for one other group + group_id = self.group.create({'name': 'Test'}) + self.model_access.write(model_access_w_group_id, { + 'group': group_id, + }) + self.test_access.delete(test_ids.pop()) + + # One access disallowed for one other group + self.model_access.write(model_access_w_group_id, { + 'perm_delete': False, + }) + self.test_access.delete(test_ids.pop()) + + transaction.cursor.rollback() + + +class ModelFieldAccessTestCase(unittest.TestCase): + ''' + Test Model Field Access + ''' + + def setUp(self): + install_module('test') + self.field_access = POOL.get('ir.model.field.access') + self.test_access = POOL.get('test.access') + self.field = POOL.get('ir.model.field') + self.group = POOL.get('res.group') + + def test0010perm_read(self): + ''' + Test Read Access + ''' + with Transaction().start(DB_NAME, USER, CONTEXT) as transaction: + field1_id, = self.field.search([ + ('model.model', '=', 'test.access'), + ('name', '=', 'field1'), + ]) + field2_id, = self.field.search([ + ('model.model', '=', 'test.access'), + ('name', '=', 'field2'), + ]) + + test_id = self.test_access.create({ + 'field1': 'ham', + 'field2': 'spam', + }) + + # Without field access + self.test_access.read(test_id, ['field1']) + self.test_access.read(test_id, ['field2']) + self.test_access.read(test_id) + test_access = self.test_access.browse(test_id) + test_access.field1 + test_access.field2 + transaction.cursor.cache.clear() + + # With field access + + # One access allowed for any group + field_access_wo_group_id = self.field_access.create({ + 'field': field1_id, + 'group': False, + 'perm_read': True, + }) + self.test_access.read(test_id, ['field1']) + self.test_access.read(test_id, ['field2']) + self.test_access.read(test_id) + test_access = self.test_access.browse(test_id) + test_access.field1 + test_access.field2 + transaction.cursor.cache.clear() + + # One access disallowed for any group + self.field_access.write(field_access_wo_group_id, { + 'perm_read': False, + }) + + self.failUnlessRaises(Exception, self.test_access.read, test_id, + ['field1']) + self.test_access.read(test_id, ['field2']) + self.failUnlessRaises(Exception, self.test_access.read, test_id) + test_access = self.test_access.browse(test_id) + self.failUnlessRaises(Exception, getattr, test_access, 'field1') + test_access.field2 + transaction.cursor.cache.clear() + + # Two access rules with one group allowed + group_id = self.group.search([('users', '=', USER)])[0] + field_access_w_group_id = self.field_access.create({ + 'field': field1_id, + 'group': group_id, + 'perm_read': True, + }) + + self.test_access.read(test_id, ['field1']) + self.test_access.read(test_id, ['field2']) + self.test_access.read(test_id) + test_access = self.test_access.browse(test_id) + test_access.field1 + test_access.field2 + transaction.cursor.cache.clear() + + # Two access rules with both allowed + self.field_access.write(field_access_wo_group_id, { + 'perm_read': True, + }) + self.test_access.read(test_id, ['field1']) + self.test_access.read(test_id, ['field2']) + self.test_access.read(test_id) + test_access = self.test_access.browse(test_id) + test_access.field1 + test_access.field2 + transaction.cursor.cache.clear() + + # Two access rules with any group allowed + self.field_access.write(field_access_w_group_id, { + 'perm_read': False, + }) + self.test_access.read(test_id, ['field1']) + self.test_access.read(test_id, ['field2']) + self.test_access.read(test_id) + test_access = self.test_access.browse(test_id) + test_access.field1 + test_access.field2 + transaction.cursor.cache.clear() + + # Two access rules with both disallowed + self.field_access.write(field_access_wo_group_id, { + 'perm_read': False, + }) + self.failUnlessRaises(Exception, self.test_access.read, test_id, + ['field1']) + self.test_access.read(test_id, ['field2']) + self.failUnlessRaises(Exception, self.test_access.read, test_id) + test_access = self.test_access.browse(test_id) + self.failUnlessRaises(Exception, getattr, test_access, 'field1') + test_access.field2 + transaction.cursor.cache.clear() + + # One access disallowed for one group + self.field_access.delete(field_access_wo_group_id) + self.failUnlessRaises(Exception, self.test_access.read, test_id, + ['field1']) + self.test_access.read(test_id, ['field2']) + self.failUnlessRaises(Exception, self.test_access.read, test_id) + test_access = self.test_access.browse(test_id) + self.failUnlessRaises(Exception, getattr, test_access, 'field1') + test_access.field2 + transaction.cursor.cache.clear() + + # One access allowed for one group + self.field_access.write(field_access_w_group_id, { + 'perm_read': True, + }) + self.test_access.read(test_id, ['field1']) + self.test_access.read(test_id, ['field2']) + self.test_access.read(test_id) + test_access = self.test_access.browse(test_id) + test_access.field1 + test_access.field2 + transaction.cursor.cache.clear() + + # One access allowed for one other group + group_id = self.group.create({'name': 'Test'}) + self.field_access.write(field_access_w_group_id, { + 'group': group_id, + }) + self.test_access.read(test_id, ['field1']) + self.test_access.read(test_id, ['field2']) + self.test_access.read(test_id) + test_access = self.test_access.browse(test_id) + test_access.field1 + test_access.field2 + transaction.cursor.cache.clear() + + # One access disallowed for one other group + self.field_access.write(field_access_w_group_id, { + 'perm_read': False, + }) + self.test_access.read(test_id, ['field1']) + self.test_access.read(test_id, ['field2']) + self.test_access.read(test_id) + test_access = self.test_access.browse(test_id) + test_access.field1 + test_access.field2 + transaction.cursor.cache.clear() + + # Two access rules on both fields allowed + self.field_access.delete(field_access_w_group_id) + + field_access1 = self.field_access.create({ + 'field': field1_id, + 'group': False, + 'perm_read': True, + }) + field_access2 = self.field_access.create({ + 'field': field2_id, + 'group': False, + 'perm_read': True, + }) + + self.test_access.read(test_id, ['field1']) + self.test_access.read(test_id, ['field2']) + self.test_access.read(test_id) + test_access = self.test_access.browse(test_id) + test_access.field1 + test_access.field2 + transaction.cursor.cache.clear() + + # Two access rules on both fields one allowed and one disallowed + self.field_access.write(field_access2, { + 'perm_read': False, + }) + self.test_access.read(test_id, ['field1']) + self.failUnlessRaises(Exception, self.test_access.read, test_id, + ['field2']) + self.failUnlessRaises(Exception, self.test_access.read, test_id) + test_access = self.test_access.browse(test_id) + test_access.field1 + self.failUnlessRaises(Exception, getattr, test_access, 'field2') + transaction.cursor.cache.clear() + + # Two access rules on both fields disallowed + self.field_access.write(field_access1, { + 'perm_read': False, + }) + self.failUnlessRaises(Exception, self.test_access.read, test_id, + ['field1']) + self.failUnlessRaises(Exception, self.test_access.read, test_id, + ['field2']) + self.failUnlessRaises(Exception, self.test_access.read, test_id) + test_access = self.test_access.browse(test_id) + self.failUnlessRaises(Exception, getattr, test_access, 'field1') + self.failUnlessRaises(Exception, getattr, test_access, 'field2') + transaction.cursor.cache.clear() + + transaction.cursor.rollback() + + def test0010perm_write(self): + ''' + Test Write Access + ''' + with Transaction().start(DB_NAME, USER, CONTEXT) as transaction: + field1_id, = self.field.search([ + ('model.model', '=', 'test.access'), + ('name', '=', 'field1'), + ]) + field2_id, = self.field.search([ + ('model.model', '=', 'test.access'), + ('name', '=', 'field2'), + ]) + + test_id = self.test_access.create({ + 'field1': 'ham', + 'field2': 'spam', + }) + + # Without field access + self.test_access.write(test_id, {}) + self.test_access.write(test_id, {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + + # With field access + + # One access allowed for any group + field_access_wo_group_id = self.field_access.create({ + 'field': field1_id, + 'group': False, + 'perm_write': True, + }) + self.test_access.write(test_id, {}) + self.test_access.write(test_id, {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + self.test_access.write(test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # One access disallowed for any group + self.field_access.write(field_access_wo_group_id, { + 'perm_write': False, + }) + + self.test_access.write(test_id, {}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, + {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # Two access rules with one group allowed + group_id = self.group.search([('users', '=', USER)])[0] + field_access_w_group_id = self.field_access.create({ + 'field': field1_id, + 'group': group_id, + 'perm_write': True, + }) + + self.test_access.write(test_id, {}) + self.test_access.write(test_id, {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + self.test_access.write(test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # Two access rules with both allowed + self.field_access.write(field_access_wo_group_id, { + 'perm_write': True, + }) + self.test_access.write(test_id, {}) + self.test_access.write(test_id, {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + self.test_access.write(test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # Two access rules with any group allowed + self.field_access.write(field_access_w_group_id, { + 'perm_write': False, + }) + self.test_access.write(test_id, {}) + self.test_access.write(test_id, {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + self.test_access.write(test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # Two access rules with both disallowed + self.field_access.write(field_access_wo_group_id, { + 'perm_write': False, + }) + self.test_access.write(test_id, {}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, + {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # One access disallowed for one group + self.field_access.delete(field_access_wo_group_id) + self.test_access.write(test_id, {}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, + {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'ham'}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # One access allowed for one group + self.field_access.write(field_access_w_group_id, { + 'perm_write': True, + }) + self.test_access.write(test_id, {}) + self.test_access.write(test_id, {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + self.test_access.write(test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # One access allowed for one other group + group_id = self.group.create({'name': 'Test'}) + self.field_access.write(field_access_w_group_id, { + 'group': group_id, + }) + self.test_access.write(test_id, {}) + self.test_access.write(test_id, {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + self.test_access.write(test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # One access disallowed for one other group + self.field_access.write(field_access_w_group_id, { + 'perm_write': False, + }) + self.test_access.write(test_id, {}) + self.test_access.write(test_id, {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + self.test_access.write(test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # Two access rules on both fields allowed + self.field_access.delete(field_access_w_group_id) + + field_access1 = self.field_access.create({ + 'field': field1_id, + 'group': False, + 'perm_write': True, + }) + field_access2 = self.field_access.create({ + 'field': field2_id, + 'group': False, + 'perm_write': True, + }) + + self.test_access.write(test_id, {}) + self.test_access.write(test_id, {'field1': 'ham'}) + self.test_access.write(test_id, {'field2': 'spam'}) + self.test_access.write(test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # Two access rules on both fields one allowed and one disallowed + self.field_access.write(field_access2, { + 'perm_write': False, + }) + self.test_access.write(test_id, {}) + self.test_access.write(test_id, {'field1': 'ham'}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, + {'field2': 'spam'}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + # Two access rules on both fields disallowed + self.field_access.write(field_access1, { + 'perm_write': False, + }) + self.test_access.write(test_id, {}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, + {'field1': 'ham'}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, + {'field2': 'spam'}) + self.failUnlessRaises(Exception, self.test_access.write, test_id, { + 'field1': 'ham', + 'field2': 'spam', + }) + + transaction.cursor.rollback() + +def suite(): + suite_ = unittest.TestSuite() + suite_.addTests(unittest.TestLoader( + ).loadTestsFromTestCase(ModelAccessTestCase)) + suite_.addTests(unittest.TestLoader( + ).loadTestsFromTestCase(ModelFieldAccessTestCase)) + return suite_ + +if __name__ == '__main__': + suite = suite() + unittest.TextTestRunner(verbosity=2).run(suite) diff --git a/trytond/tests/test_tryton.py b/trytond/tests/test_tryton.py index df2373f1ef..ba6e8f5e7b 100644 --- a/trytond/tests/test_tryton.py +++ b/trytond/tests/test_tryton.py @@ -194,6 +194,8 @@ def all_suite(): suite_.addTests(test_protocols_datatype.suite()) import trytond.tests.test_sequence as test_sequence suite_.addTests(test_sequence.suite()) + import trytond.tests.test_access as test_access + suite_.addTests(test_access.suite()) return suite_ def modules_suite():