tryton-do/tryton
2
0 Fork 0
Code Issues Pull Requests 2 Actions 2 Packages Projects Releases Wiki Activity

Check rule after indirect fields and modification

Browse Source 
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
This commit is contained in:
Cédric Krier committed 2018-06-24 16:54:46 +02:00
1 parent 0d07dea2ac
commit 93526b9cd0
5 files changed
+404 -60
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CHANGELOG
+2
View File
@@ -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
trytond/model/modelsql.py
+35 -60
View File
@@ -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):
trytond/tests/__init__.py
+2
View File
@@ -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')
trytond/tests/rule.py
+16
View File
@@ -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')
trytond/tests/test_rule.py
+349
View File
@@ -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_