From 93526b9cd0e8d5e374366f767e97bef805c5da6b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9dric=20Krier?= Date: Sun, 24 Jun 2018 16:54:46 +0200 Subject: [PATCH] Check rule after indirect fields and modification The rule may depend on indirect field written latter so it must be checked after the storage of those fields. On write, the rule must be checked before the modification but also after to ensure users do not generate records that they can not modify after. issue7403 review53301002 --- CHANGELOG | 2 + trytond/model/modelsql.py | 95 ++++------ trytond/tests/__init__.py | 2 + trytond/tests/rule.py | 16 ++ trytond/tests/test_rule.py | 349 +++++++++++++++++++++++++++++++++++++ 5 files changed, 404 insertions(+), 60 deletions(-) create mode 100644 trytond/tests/rule.py create mode 100644 trytond/tests/test_rule.py diff --git a/CHANGELOG b/CHANGELOG index dc984f361d..8cbe60799d 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,5 @@ +* Check Rule also after modification +* Check Rule after indirect fields * Allow to define view id in switch client action * Do not validate Function fields even with setter * Simplify the creation of dependencies graph diff --git a/trytond/model/modelsql.py b/trytond/model/modelsql.py index 147abc03f2..ede0ffa8e4 100644 --- a/trytond/model/modelsql.py +++ b/trytond/model/modelsql.py @@ -538,7 +538,6 @@ class ModelSQL(ModelStorage): cursor = transaction.connection.cursor() pool = Pool() Translation = pool.get('ir.translation') - Rule = pool.get('ir.rule') super(ModelSQL, cls).create(vlist) @@ -608,21 +607,6 @@ class ModelSQL(ModelStorage): exception, values, transaction=transaction) raise - domain = Rule.domain_get(cls.__name__, mode='create') - if domain: - tables = {None: (table, None)} - tables, expression = cls.search_domain( - domain, active_test=False, tables=tables) - from_ = convert_from(None, tables) - for sub_ids in grouped_slice(new_ids): - sub_ids = list(sub_ids) - red_sql = reduce_ids(table.id, sub_ids) - - cursor.execute(*from_.select(table.id, - where=red_sql & expression)) - if len(cursor.fetchall()) != len(sub_ids): - cls.raise_user_error('access_error', cls.__name__) - transaction.create_records.setdefault(cls.__name__, set()).update(new_ids) @@ -660,6 +644,7 @@ class ModelSQL(ModelStorage): field_names = cls._fields.keys() cls._update_mptt(field_names, [new_ids] * len(field_names)) + cls.__check_domain_rule(new_ids, 'create') records = cls.browse(new_ids) for sub_records in grouped_slice(records, cache_size()): cls._validate(sub_records) @@ -929,7 +914,6 @@ class ModelSQL(ModelStorage): pool = Pool() Translation = pool.get('ir.translation') Config = pool.get('ir.configuration') - Rule = pool.get('ir.rule') assert not len(args) % 2 # Remove possible duplicates from all records @@ -946,6 +930,7 @@ class ModelSQL(ModelStorage): table = cls.__table__() cls.__check_timestamp(all_ids) + cls.__check_domain_rule(all_ids, 'write', nodomain='write_error') fields_to_set = {} actions = iter((records, values) + args) @@ -970,31 +955,8 @@ class ModelSQL(ModelStorage): columns.append(Column(table, fname)) update_values.append(field.sql_format(value)) - domain = Rule.domain_get(cls.__name__, mode='write') - tables = {None: (table, None)} - if domain: - tables, dom_exp = cls.search_domain( - domain, active_test=False, tables=tables) - from_ = convert_from(None, tables) for sub_ids in grouped_slice(ids): - sub_ids = list(sub_ids) red_sql = reduce_ids(table.id, sub_ids) - where = red_sql - if domain: - where &= dom_exp - cursor.execute(*from_.select(table.id, where=where)) - rowcount = cursor.rowcount - if rowcount == -1 or rowcount is None: - rowcount = len(cursor.fetchall()) - if not rowcount == len({}.fromkeys(sub_ids)): - if domain: - cursor.execute(*table.select(table.id, where=red_sql)) - rowcount = cursor.rowcount - if rowcount == -1 or rowcount is None: - rowcount = len(cursor.fetchall()) - if rowcount == len({}.fromkeys(sub_ids)): - cls.raise_user_error('access_error', cls.__name__) - cls.raise_user_error('write_error', cls.__name__) try: cursor.execute(*table.update(columns, update_values, where=red_sql)) @@ -1027,8 +989,11 @@ class ModelSQL(ModelStorage): field.set(cls, fname, *fargs) cls._insert_history(all_ids) + + cls.__check_domain_rule(all_ids, 'write') for sub_records in grouped_slice(all_records, cache_size()): cls._validate(sub_records, field_names=all_field_names) + cls.trigger_write(trigger_eligibles) @classmethod @@ -1039,7 +1004,6 @@ class ModelSQL(ModelStorage): cursor = transaction.connection.cursor() pool = Pool() Translation = pool.get('ir.translation') - Rule = pool.get('ir.rule') ids = map(int, records) if not ids: @@ -1054,6 +1018,7 @@ class ModelSQL(ModelStorage): ids.remove(del_id) cls.__check_timestamp(ids) + cls.__check_domain_rule(ids, 'delete') has_translation = False tree_ids = {} @@ -1092,25 +1057,6 @@ class ModelSQL(ModelStorage): foreign_keys_tocheck.append((model, field_name)) transaction.delete.setdefault(cls.__name__, set()).update(ids) - - domain = Rule.domain_get(cls.__name__, mode='delete') - - if domain: - tables = {None: (table, None)} - tables, dom_exp = cls.search_domain( - domain, active_test=False, tables=tables) - from_ = convert_from(None, tables) - for sub_ids in grouped_slice(ids): - sub_ids = list(sub_ids) - red_sql = reduce_ids(table.id, sub_ids) - cursor.execute(*from_.select(table.id, - where=red_sql & dom_exp)) - rowcount = cursor.rowcount - if rowcount == -1 or rowcount is None: - rowcount = len(cursor.fetchall()) - if not rowcount == len({}.fromkeys(sub_ids)): - cls.raise_user_error('access_error', cls.__name__) - cls.trigger_delete(records) def get_related_records(Model, field_name, sub_ids): @@ -1179,6 +1125,35 @@ class ModelSQL(ModelStorage): cls._update_mptt(tree_ids.keys(), tree_ids.values()) + @classmethod + def __check_domain_rule(cls, ids, mode, nodomain=None): + pool = Pool() + Rule = pool.get('ir.rule') + table = cls.__table__() + cursor = Transaction().connection.cursor() + + domain = Rule.domain_get(cls.__name__, mode=mode) + tables = {None: (table, None)} + if domain or nodomain: + if domain: + tables, dom_exp = cls.search_domain( + domain, active_test=False, tables=tables) + from_ = convert_from(None, tables) + for sub_ids in grouped_slice(ids): + sub_ids = list(set(sub_ids)) + where = reduce_ids(table.id, sub_ids) + if domain: + where &= dom_exp + cursor.execute(*from_.select(table.id, where=where)) + rowcount = cursor.rowcount + if rowcount == -1 or rowcount is None: + rowcount = len(cursor.fetchall()) + if rowcount != len(sub_ids): + if domain: + cls.raise_user_error('access_error', cls.__name__) + else: + cls.raise_user_error(nodomain, cls.__name__) + @classmethod def search(cls, domain, offset=0, limit=None, order=None, count=False, query=False): diff --git a/trytond/tests/__init__.py b/trytond/tests/__init__.py index dcf500ac32..f3e8812809 100644 --- a/trytond/tests/__init__.py +++ b/trytond/tests/__init__.py @@ -43,6 +43,7 @@ from . import field_one2one from . import multivalue from . import mixin from . import tree +from . import rule def register(): @@ -126,6 +127,7 @@ def register(): multivalue.register('tests') mixin.register('tests') tree.register('tests') + rule.register('tests') if pkg_resources is not None: entry_points = pkg_resources.iter_entry_points('trytond.tests') diff --git a/trytond/tests/rule.py b/trytond/tests/rule.py new file mode 100644 index 0000000000..32ff686b5e --- /dev/null +++ b/trytond/tests/rule.py @@ -0,0 +1,16 @@ +# 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 +from trytond.pool import Pool + + +class TestRule(ModelSQL): + "Test Rule" + __name__ = 'test.rule' + field = fields.Char("Field") + + +def register(module): + Pool.register( + TestRule, + module=module, type_='model') diff --git a/trytond/tests/test_rule.py b/trytond/tests/test_rule.py new file mode 100644 index 0000000000..ccd9498997 --- /dev/null +++ b/trytond/tests/test_rule.py @@ -0,0 +1,349 @@ +# This file is part of Tryton. The COPYRIGHT file at the top level of +# this repository contains the full copyright notices and license terms. +import json +import unittest + +from trytond.exceptions import UserError +from trytond.pool import Pool +from trytond.tests.test_tryton import activate_module, with_transaction + + +class ModelRuleTestCase(unittest.TestCase): + "Test Model Rule" + + @classmethod + def setUpClass(cls): + activate_module('tests') + + @with_transaction() + def test_perm_create_without_rule(self): + "Test create without rule" + pool = Pool() + TestRule = pool.get('test.rule') + + test, = TestRule.create([{}]) + + @with_transaction() + def test_perm_create_with_rule(self): + "Test create with rule" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': False, + 'perm_create': True, + 'perm_write': False, + 'perm_delete': False, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + + test, = TestRule.create([{'field': 'bar'}]) + + @with_transaction() + def test_perm_create_with_rule_fail(self): + "Test create with rule fail" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': False, + 'perm_create': True, + 'perm_write': False, + 'perm_delete': False, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + + with self.assertRaises(UserError): + test, = TestRule.create([{'field': 'foo'}]) + + @with_transaction() + def test_perm_write_without_rule(self): + "Test write without rule" + pool = Pool() + TestRule = pool.get('test.rule') + + test, = TestRule.create([{}]) + + TestRule.write([test], {'field': 'foo'}) + + @with_transaction() + def test_perm_write_with_rule(self): + "Test write with rule" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': False, + 'perm_create': False, + 'perm_write': True, + 'perm_delete': False, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + test, = TestRule.create([{'field': 'test'}]) + + TestRule.write([test], {'field': 'bar'}) + + @with_transaction() + def test_perm_write_with_rule_fail_before(self): + "Test write with rule fail before" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': False, + 'perm_create': False, + 'perm_write': True, + 'perm_delete': False, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + test, = TestRule.create([{'field': 'foo'}]) + + with self.assertRaises(UserError): + TestRule.write([test], {'field': 'bar'}) + + @with_transaction() + def test_perm_write_with_rule_fail_after(self): + "Test write with rule fail after" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': False, + 'perm_create': False, + 'perm_write': True, + 'perm_delete': False, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + test, = TestRule.create([{'field': 'bar'}]) + + with self.assertRaises(UserError): + TestRule.write([test], {'field': 'foo'}) + + @with_transaction() + def test_perm_delete_without_rule(self): + "Test delete without rule" + pool = Pool() + TestRule = pool.get('test.rule') + + test, = TestRule.create([{}]) + + TestRule.delete([test]) + + @with_transaction() + def test_perm_delete_with_rule(self): + "Test delete with rule" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': False, + 'perm_create': False, + 'perm_write': False, + 'perm_delete': True, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + test, = TestRule.create([{'field': 'bar'}]) + + TestRule.delete([test]) + + @with_transaction() + def test_perm_delete_with_rule_fail(self): + "Test delete with rule fail" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': False, + 'perm_create': False, + 'perm_write': False, + 'perm_delete': True, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + test, = TestRule.create([{'field': 'foo'}]) + + with self.assertRaises(UserError): + TestRule.delete([test]) + + @with_transaction() + def test_perm_read_without_rule(self): + "Test read without rule" + pool = Pool() + TestRule = pool.get('test.rule') + + test, = TestRule.create([{'field': 'foo'}]) + + TestRule.read([test.id]) + + @with_transaction() + def test_perm_read_with_rule(self): + "Test read with rule" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': True, + 'perm_create': False, + 'perm_write': False, + 'perm_delete': False, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + test, = TestRule.create([{'field': 'bar'}]) + + TestRule.read([test.id]) + + @with_transaction() + def test_perm_read_with_rule_fail(self): + "Test read with rule fail" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': True, + 'perm_create': False, + 'perm_write': False, + 'perm_delete': False, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + test, = TestRule.create([{'field': 'foo'}]) + + with self.assertRaises(UserError): + TestRule.read([test.id]) + + @with_transaction() + def test_search_without_rule(self): + "Test search without rule" + pool = Pool() + TestRule = pool.get('test.rule') + + test, = TestRule.create([{'field': 'foo'}]) + + self.assertListEqual(TestRule.search([]), [test]) + + @with_transaction() + def test_search_with_rule(self): + "Test search with rule" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': True, + 'perm_create': False, + 'perm_write': False, + 'perm_delete': False, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + test, = TestRule.create([{'field': 'bar'}]) + + self.assertListEqual(TestRule.search([]), [test]) + + @with_transaction() + def test_search_with_rule_match(self): + "Test search with rule match" + pool = Pool() + TestRule = pool.get('test.rule') + RuleGroup = pool.get('ir.rule.group') + Model = pool.get('ir.model') + + model, = Model.search([('model', '=', 'test.rule')]) + rule_group, = RuleGroup.create([{ + 'model': model.id, + 'global_p': True, + 'perm_read': True, + 'perm_create': False, + 'perm_write': False, + 'perm_delete': False, + 'rules': [('create', [{ + 'domain': json.dumps( + [('field', '!=', 'foo')]), + }])], + }]) + test, = TestRule.create([{'field': 'foo'}]) + + self.assertListEqual(TestRule.search([]), []) + + +def suite(): + suite_ = unittest.TestSuite() + suite_.addTests( + unittest.TestLoader().loadTestsFromTestCase(ModelRuleTestCase)) + return suite_