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_