odoo16/addons/test_base_automation/tests/test_flow.py

# # -*- coding: utf-8 -*-
# # Part of Odoo. See LICENSE file for full copyright and licensing details.

from unittest.mock import patch
import sys

from odoo.addons.base.tests.common import TransactionCaseWithUserDemo
from odoo.tests import common, tagged
from odoo.exceptions import AccessError


@tagged('post_install', '-at_install')
class BaseAutomationTest(TransactionCaseWithUserDemo):

    def setUp(self):
        super(BaseAutomationTest, self).setUp()
        self.user_root = self.env.ref('base.user_root')
        self.user_admin = self.env.ref('base.user_admin')

        self.test_mail_template_automation = self.env['mail.template'].create({
            'name': 'Template Automation',
            'model_id': self.env.ref('test_base_automation.model_base_automation_lead_test').id,
            'body_html': """<div>Email automation</div>""",
        })

        self.res_partner_1 = self.env['res.partner'].create({'name': 'My Partner'})
        self.env['base.automation'].create([
            {
                'name': 'Base Automation: test rule on create',
                'model_id': self.env.ref('test_base_automation.model_base_automation_lead_test').id,
                'state': 'code',
                'code': "records.write({'user_id': %s})" % (self.user_demo.id),
                'trigger': 'on_create',
                'active': True,
                'filter_domain': "[('state', '=', 'draft')]",
            }, {
                'name': 'Base Automation: test rule on write',
                'model_id': self.env.ref('test_base_automation.model_base_automation_lead_test').id,
                'state': 'code',
                'code': "records.write({'user_id': %s})" % (self.user_demo.id),
                'trigger': 'on_write',
                'active': True,
                'filter_domain': "[('state', '=', 'done')]",
                'filter_pre_domain': "[('state', '=', 'open')]",
            }, {
                'name': 'Base Automation: test rule on recompute',
                'model_id': self.env.ref('test_base_automation.model_base_automation_lead_test').id,
                'state': 'code',
                'code': "records.write({'user_id': %s})" % (self.user_demo.id),
                'trigger': 'on_write',
                'active': True,
                'filter_domain': "[('employee', '=', True)]",
            }, {
                'name': 'Base Automation: test recursive rule',
                'model_id': self.env.ref('test_base_automation.model_base_automation_lead_test').id,
                'state': 'code',
                'code': """
record = model.browse(env.context['active_id'])
if 'partner_id' in env.context['old_values'][record.id]:
    record.write({'state': 'draft'})""",
                'trigger': 'on_write',
                'active': True,
            }, {
                'name': 'Base Automation: test rule on secondary model',
                'model_id': self.env.ref('test_base_automation.model_base_automation_line_test').id,
                'state': 'code',
                'code': "records.write({'user_id': %s})" % (self.user_demo.id),
                'trigger': 'on_create',
                'active': True,
            }, {
                'name': 'Base Automation: test rule on write check context',
                'model_id': self.env.ref('test_base_automation.model_base_automation_lead_test').id,
                'state': 'code',
                'code': """
record = model.browse(env.context['active_id'])
if 'user_id' in env.context['old_values'][record.id]:
    record.write({'is_assigned_to_admin': (record.user_id.id == 1)})""",
                'trigger': 'on_write',
                'active': True,
            }, {
                'name': 'Base Automation: test rule with trigger',
                'model_id': self.env.ref('test_base_automation.model_base_automation_lead_test').id,
                'trigger_field_ids': [(4, self.env.ref('test_base_automation.field_base_automation_lead_test__state').id)],
                'state': 'code',
                'code': """
record = model.browse(env.context['active_id'])
record['name'] = record.name + 'X'""",
                'trigger': 'on_write',
                'active': True,
            }, {
                'name': 'Base Automation: test send an email',
                'mail_post_method': 'email',
                'model_id': self.env.ref('test_base_automation.model_base_automation_lead_test').id,
                'template_id': self.test_mail_template_automation.id,
                'trigger_field_ids': [(4, self.env.ref('test_base_automation.field_base_automation_lead_test__deadline').id)],
                'state': 'mail_post',
                'code': """
record = model.browse(env.context['active_id'])
record['name'] = record.name + 'X'""",
                'trigger': 'on_write',
                'active': True,
                'filter_domain': "[('deadline', '!=', False)]",
                'filter_pre_domain': "[('deadline', '=', False)]",
            }
        ])

    def tearDown(self):
        super().tearDown()
        self.env['base.automation']._unregister_hook()

    def create_lead(self, **kwargs):
        vals = {
            'name': "Lead Test",
            'user_id': self.user_root.id,
        }
        vals.update(kwargs)
        return self.env['base.automation.lead.test'].create(vals)

    def test_00_check_to_state_open_pre(self):
        """
        Check that a new record (with state = open) doesn't change its responsible
        when there is a precondition filter which check that the state is open.
        """
        lead = self.create_lead(state='open')
        self.assertEqual(lead.state, 'open')
        self.assertEqual(lead.user_id, self.user_root, "Responsible should not change on creation of Lead with state 'open'.")

    def test_01_check_to_state_draft_post(self):
        """
        Check that a new record changes its responsible when there is a postcondition
        filter which check that the state is draft.
        """
        lead = self.create_lead()
        self.assertEqual(lead.state, 'draft', "Lead state should be 'draft'")
        self.assertEqual(lead.user_id, self.user_demo, "Responsible should be change on creation of Lead with state 'draft'.")

    def test_02_check_from_draft_to_done_with_steps(self):
        """
        A new record is created and goes from states 'open' to 'done' via the
        other states (open, pending and cancel). We have a rule with:
         - precondition: the record is in "open"
         - postcondition: that the record is "done".
        If the state goes from 'open' to 'done' the responsible is changed.
        If those two conditions aren't verified, the responsible remains the same.
        """
        lead = self.create_lead(state='open')
        self.assertEqual(lead.state, 'open', "Lead state should be 'open'")
        self.assertEqual(lead.user_id, self.user_root, "Responsible should not change on creation of Lead with state 'open'.")
        # change state to pending and check that responsible has not changed
        lead.write({'state': 'pending'})
        self.assertEqual(lead.state, 'pending', "Lead state should be 'pending'")
        self.assertEqual(lead.user_id, self.user_root, "Responsible should not change on creation of Lead with state from 'draft' to 'open'.")
        # change state to done and check that responsible has not changed
        lead.write({'state': 'done'})
        self.assertEqual(lead.state, 'done', "Lead state should be 'done'")
        self.assertEqual(lead.user_id, self.user_root, "Responsible should not chang on creation of Lead with state from 'pending' to 'done'.")

    def test_03_check_from_draft_to_done_without_steps(self):
        """
        A new record is created and goes from states 'open' to 'done' via the
        other states (open, pending and cancel). We have a rule with:
         - precondition: the record is in "open"
         - postcondition: that the record is "done".
        If the state goes from 'open' to 'done' the responsible is changed.
        If those two conditions aren't verified, the responsible remains the same.
        """
        lead = self.create_lead(state='open')
        self.assertEqual(lead.state, 'open', "Lead state should be 'open'")
        self.assertEqual(lead.user_id, self.user_root, "Responsible should not change on creation of Lead with state 'open'.")
        # change state to done and check that responsible has changed
        lead.write({'state': 'done'})
        self.assertEqual(lead.state, 'done', "Lead state should be 'done'")
        self.assertEqual(lead.user_id, self.user_demo, "Responsible should be change on write of Lead with state from 'open' to 'done'.")

    def test_10_recomputed_field(self):
        """
        Check that a rule is executed whenever a field is recomputed after a
        change on another model.
        """
        partner = self.res_partner_1
        partner.write({'employee': False})
        lead = self.create_lead(state='open', partner_id=partner.id)
        self.assertFalse(lead.employee, "Customer field should updated to False")
        self.assertEqual(lead.user_id, self.user_root, "Responsible should not change on creation of Lead with state from 'draft' to 'open'.")
        # change partner, recompute on lead should trigger the rule
        partner.write({'employee': True})
        self.env.flush_all()
        self.assertTrue(lead.employee, "Customer field should updated to True")
        self.assertEqual(lead.user_id, self.user_demo, "Responsible should be change on write of Lead when Customer becomes True.")

    def test_11_recomputed_field(self):
        """
        Check that a rule is executed whenever a field is recomputed and the
        context contains the target field
        """
        partner = self.res_partner_1
        lead = self.create_lead(state='draft', partner_id=partner.id)
        self.assertFalse(lead.deadline, 'There should not be a deadline defined')
        # change priority and user; this triggers deadline recomputation, and
        # the server action should set the boolean field to True
        lead.write({'priority': True, 'user_id': self.user_root.id})
        self.assertTrue(lead.deadline, 'Deadline should be defined')
        self.assertTrue(lead.is_assigned_to_admin, 'Lead should be assigned to admin')

    def test_11b_recomputed_field(self):
        mail_automation = self.env['base.automation'].search([('name', '=', 'Base Automation: test send an email')])
        send_mail_count = 0

        def _patched_get_actions(*args, **kwargs):
            obj = args[0]
            if '__action_done' not in obj._context:
                obj = obj.with_context(__action_done={})
            return mail_automation.with_env(obj.env)

        def _patched_send_mail(*args, **kwargs):
            nonlocal send_mail_count
            send_mail_count += 1

        patchers = [
            patch('odoo.addons.base_automation.models.base_automation.BaseAutomation._get_actions', _patched_get_actions),
            patch('odoo.addons.mail.models.mail_template.MailTemplate.send_mail', _patched_send_mail),
        ]

        self.startPatcher(patchers[0])

        lead = self.create_lead()
        self.assertFalse(lead.priority)
        self.assertFalse(lead.deadline)

        self.startPatcher(patchers[1])

        lead.write({'priority': True})

        self.assertTrue(lead.priority)
        self.assertTrue(lead.deadline)


        self.assertEqual(send_mail_count, 1)

    def test_12_recursive(self):
        """ Check that a rule is executed recursively by a secondary change. """
        lead = self.create_lead(state='open')
        self.assertEqual(lead.state, 'open')
        self.assertEqual(lead.user_id, self.user_root)
        # change partner; this should trigger the rule that modifies the state
        partner = self.res_partner_1
        lead.write({'partner_id': partner.id})
        self.assertEqual(lead.state, 'draft')

    def test_20_direct_line(self):
        """
        Check that a rule is executed after creating a line record.
        """
        line = self.env['base.automation.line.test'].create({'name': "Line"})
        self.assertEqual(line.user_id, self.user_demo)

    def test_20_indirect_line(self):
        """
        Check that creating a lead with a line executes rules on both records.
        """
        lead = self.create_lead(line_ids=[(0, 0, {'name': "Line"})])
        self.assertEqual(lead.state, 'draft', "Lead state should be 'draft'")
        self.assertEqual(lead.user_id, self.user_demo, "Responsible should change on creation of Lead test line.")
        self.assertEqual(len(lead.line_ids), 1, "New test line is not created")
        self.assertEqual(lead.line_ids.user_id, self.user_demo, "Responsible should be change on creation of Lead test line.")

    def test_21_trigger_fields(self):
        """
        Check that the rule with trigger is executed only once per pertinent update.
        """
        lead = self.create_lead(name="X")
        lead.priority = True
        partner1 = self.res_partner_1
        lead.partner_id = partner1.id
        self.assertEqual(lead.name, 'X', "No update until now.")

        lead.state = 'open'
        self.assertEqual(lead.name, 'XX', "One update should have happened.")
        lead.state = 'done'
        self.assertEqual(lead.name, 'XXX', "One update should have happened.")
        lead.state = 'done'
        self.assertEqual(lead.name, 'XXX', "No update should have happened.")
        lead.state = 'cancel'
        self.assertEqual(lead.name, 'XXXX', "One update should have happened.")

        # change the rule to trigger on partner_id
        rule = self.env['base.automation'].search([('name', '=', 'Base Automation: test rule with trigger')])
        rule.write({'trigger_field_ids':  [(6, 0, [self.env.ref('test_base_automation.field_base_automation_lead_test__partner_id').id])]})

        partner2 = self.env['res.partner'].create({'name': 'A new partner'})
        lead.name = 'X'
        lead.state = 'open'
        self.assertEqual(lead.name, 'X', "No update should have happened.")
        lead.partner_id = partner2
        self.assertEqual(lead.name, 'XX', "One update should have happened.")
        lead.partner_id = partner2
        self.assertEqual(lead.name, 'XX', "No update should have happened.")
        lead.partner_id = partner1
        self.assertEqual(lead.name, 'XXX', "One update should have happened.")

    def test_30_modelwithoutaccess(self):
        """
        Ensure a domain on a M2O without user access doesn't fail.
        We create a base automation with a filter on a model the user haven't access to
        - create a group
        - restrict acl to this group and set only admin in it
        - create base.automation with a filter
        - create a record in the restricted model in admin
        - create a record in the non restricted model in demo
        """
        Model = self.env['base.automation.link.test']
        Comodel = self.env['base.automation.linked.test']

        access = self.env.ref("test_base_automation.access_base_automation_linked_test")
        access.group_id = self.env['res.groups'].create({
            'name': "Access to base.automation.linked.test",
            "users": [(6, 0, [self.user_admin.id,])]
        })

        # sanity check: user demo has no access to the comodel of 'linked_id'
        with self.assertRaises(AccessError):
            Comodel.with_user(self.user_demo).check_access_rights('read')

        # check base automation with filter that performs Comodel.search()
        self.env['base.automation'].create({
            'name': 'test no access',
            'model_id': self.env['ir.model']._get_id("base.automation.link.test"),
            'trigger': 'on_create_or_write',
            'filter_pre_domain': "[('linked_id.another_field', '=', 'something')]",
            'state': 'code',
            'active': True,
            'code': "action = [rec.name for rec in records]"
        })
        Comodel.create([
            {'name': 'a first record', 'another_field': 'something'},
            {'name': 'another record', 'another_field': 'something different'},
        ])
        rec1 = Model.create({'name': 'a record'})
        rec1.write({'name': 'a first record'})
        rec2 = Model.with_user(self.user_demo).create({'name': 'another record'})
        rec2.write({'name': 'another value'})

        # check base automation with filter that performs Comodel.name_search()
        self.env['base.automation'].create({
            'name': 'test no name access',
            'model_id': self.env['ir.model']._get_id("base.automation.link.test"),
            'trigger': 'on_create_or_write',
            'filter_pre_domain': "[('linked_id', '=', 'whatever')]",
            'state': 'code',
            'active': True,
            'code': "action = [rec.name for rec in records]"
        })
        rec3 = Model.create({'name': 'a random record'})
        rec3.write({'name': 'a first record'})
        rec4 = Model.with_user(self.user_demo).create({'name': 'again another record'})
        rec4.write({'name': 'another value'})


@common.tagged('post_install', '-at_install')
class TestCompute(common.TransactionCase):
    def test_inversion(self):
        """ If a stored field B depends on A, an update to the trigger for A
        should trigger the recomputaton of A, then B.

        However if a search() is performed during the computation of A
        ??? and _order is affected ??? a flush will be triggered, forcing the
        computation of B, based on the previous A.

        This happens if a rule has has a non-empty filter_pre_domain, even if
        it's an empty list (``'[]'`` as opposed to ``False``).
        """
        company1 = self.env['res.partner'].create({
            'name': "Gorofy",
            'is_company': True,
        })
        company2 = self.env['res.partner'].create({
            'name': "Awiclo",
            'is_company': True
        })
        r = self.env['res.partner'].create({
            'name': 'Bob',
            'is_company': False,
            'parent_id': company1.id
        })
        self.assertEqual(r.display_name, 'Gorofy, Bob')
        r.parent_id = company2
        self.assertEqual(r.display_name, 'Awiclo, Bob')

        self.env['base.automation'].create({
            'name': "test rule",
            'filter_pre_domain': False,
            'trigger': 'on_create_or_write',
            'state': 'code', # no-op action
            'model_id': self.env.ref('base.model_res_partner').id,
        })
        r.parent_id = company1
        self.assertEqual(r.display_name, 'Gorofy, Bob')

        self.env['base.automation'].create({
            'name': "test rule",
            'filter_pre_domain': '[]',
            'trigger': 'on_create_or_write',
            'state': 'code', # no-op action
            'model_id': self.env.ref('base.model_res_partner').id,
        })
        r.parent_id = company2
        self.assertEqual(r.display_name, 'Awiclo, Bob')

    def test_recursion(self):
        project = self.env['test_base_automation.project'].create({})

        # this action is executed every time a task is assigned to project
        self.env['base.automation'].create({
            'name': 'dummy',
            'model_id': self.env['ir.model']._get_id('test_base_automation.task'),
            'state': 'code',
            'trigger': 'on_create_or_write',
            'filter_domain': repr([('project_id', '=', project.id)]),
        })

        # create one task in project with 10 subtasks; all the subtasks are
        # automatically assigned to project, too
        task = self.env['test_base_automation.task'].create({'project_id': project.id})
        subtasks = task.create([{'parent_id': task.id} for _ in range(10)])
        subtasks.flush_model()

        # This test checks what happens when a stored recursive computed field
        # is marked to compute on many records, and automated actions are
        # triggered depending on that field.  In this case, we trigger the
        # recomputation of 'project_id' on 'subtasks' by deleting their parent
        # task.
        #
        # An issue occurs when the domain of automated actions is evaluated by
        # method search(), because the latter flushes the fields to search on,
        # which are also the ones being recomputed.  Combined with the fact
        # that recursive fields are not computed in batch, this leads to a huge
        # amount of recursive calls between the automated action and flush().
        #
        # The execution of task.unlink() looks like this:
        # - mark 'project_id' to compute on subtasks
        # - delete task
        # - flush()
        #   - recompute 'project_id' on subtask1
        #     - call compute on subtask1
        #     - in action, search([('id', 'in', subtask1.ids), ('project_id', '=', pid)])
        #       - flush(['id', 'project_id'])
        #         - recompute 'project_id' on subtask2
        #           - call compute on subtask2
        #           - in action search([('id', 'in', subtask2.ids), ('project_id', '=', pid)])
        #             - flush(['id', 'project_id'])
        #               - recompute 'project_id' on subtask3
        #                 - call compute on subtask3
        #                 - in action, search([('id', 'in', subtask3.ids), ('project_id', '=', pid)])
        #                   - flush(['id', 'project_id'])
        #                     - recompute 'project_id' on subtask4
        #                       ...
        limit = sys.getrecursionlimit()
        try:
            sys.setrecursionlimit(100)
            task.unlink()
        finally:
            sys.setrecursionlimit(limit)