oca/oca-technical
2
0
Fork 0
mirror of https://github.com/bringout/oca-technical.git synced 2026-04-22 23:12:08 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Initial commit: OCA Technical packages (595 packages)

Browse source
This commit is contained in:
Ernad Husremovic 2025-08-29 15:43:03 +02:00
commit 2cc02aac6e
24950 changed files with 2318079 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

5
odoo-bringout-oca-automation-automation_oca/automation_oca/tests/__init__.py Normal file
View file

@ -0,0 +1,5 @@
from . import test_automation_action
from . import test_automation_activity
from . import test_automation_base
from . import test_automation_mail
from . import test_automation_security

93
odoo-bringout-oca-automation-automation_oca/automation_oca/tests/common.py Normal file
View file

@ -0,0 +1,93 @@
# Copyright 2024 Dixmit
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from odoo.tests.common import TransactionCase
class AutomationTestCase(TransactionCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.env["automation.configuration"].search([]).toggle_active()
cls.action = cls.env["ir.actions.server"].create(
{
"name": "Demo action",
"state": "code",
"model_id": cls.env.ref("base.model_res_partner").id,
"code": "records.write({'comment': env.context.get('key_value')})",
}
)
cls.activity_type = cls.env["mail.activity.type"].create({"name": "DEMO"})
cls.error_action = cls.env["ir.actions.server"].create(
{
"name": "Demo action",
"state": "code",
"model_id": cls.env.ref("base.model_res_partner").id,
"code": "raise UserError('ERROR')",
}
)
cls.template = cls.env["mail.template"].create(
{
"name": "My template",
"model_id": cls.env.ref("base.model_res_partner").id,
"subject": "Subject",
"partner_to": "{{ object.id }}",
"body_html": 'My template <a href="https://www.twitter.com" /> with link',
}
)
cls.partner_01 = cls.env["res.partner"].create(
{"name": "Demo partner", "comment": "Demo", "email": "test@test.com"}
)
cls.partner_02 = cls.env["res.partner"].create(
{"name": "Demo partner 2", "comment": "Demo", "email": "test@test.com"}
)
cls.configuration = cls.env["automation.configuration"].create(
{
"name": "Test configuration",
"model_id": cls.env.ref("base.model_res_partner").id,
"is_periodic": True,
}
)
@classmethod
def create_server_action(cls, parent_id=False, **kwargs):
return cls.env["automation.configuration.step"].create(
{
"name": "Demo activity",
"parent_id": parent_id,
"configuration_id": cls.configuration.id,
"step_type": "action",
"server_action_id": cls.action.id,
"trigger_type": "after_step" if parent_id else "start",
**kwargs,
}
)
@classmethod
def create_activity_action(cls, parent_id=False, **kwargs):
return cls.env["automation.configuration.step"].create(
{
"name": "Demo activity",
"parent_id": parent_id,
"configuration_id": cls.configuration.id,
"step_type": "activity",
"activity_type_id": cls.activity_type.id,
"trigger_type": "after_step" if parent_id else "start",
**kwargs,
}
)
@classmethod
def create_mail_activity(cls, parent_id=False, trigger_type=False, **kwargs):
return cls.env["automation.configuration.step"].create(
{
"name": "Demo activity",
"parent_id": parent_id,
"configuration_id": cls.configuration.id,
"step_type": "mail",
"mail_template_id": cls.template.id,
"trigger_type": trigger_type
or ("after_step" if parent_id else "start"),
**kwargs,
}
)

235
odoo-bringout-oca-automation-automation_oca/automation_oca/tests/test_automation_action.py Normal file
View file

@ -0,0 +1,235 @@
# Copyright 2024 Dixmit
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from .common import AutomationTestCase
class TestAutomationAction(AutomationTestCase):
def test_activity_immediate_execution(self):
"""
We will check the execution of the tasks and that we cannot execute them again
"""
activity = self.create_server_action(trigger_interval=-1)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertFalse(self.partner_01.comment)
self.assertTrue(self.partner_02.comment)
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual(1, len(record_activity))
self.assertEqual("done", record_activity.state)
def test_activity_execution(self):
"""
We will check the execution of the tasks and that we cannot execute them again
"""
activity = self.create_server_action()
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertTrue(self.partner_01.comment)
self.assertTrue(self.partner_02.comment)
self.env["automation.record.step"]._cron_automation_steps()
self.assertFalse(self.partner_01.comment)
self.assertTrue(self.partner_02.comment)
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual(1, len(record_activity))
self.assertEqual("done", record_activity.state)
self.partner_01.comment = "My comment"
# We check that the action is not executed again
record_activity.run()
self.assertFalse(record_activity.step_actions)
self.assertTrue(self.partner_01.comment)
def test_child_execution_filters(self):
"""
We will create a task that executes two more tasks filtered with and extra task
The child tasks should only be created after the first one is finished.
Also, if one is aborted, the subsuquent tasks will not be created.
TASK 1 ---> TASK 1_1 (only for partner 1) --> TASK 1_1_1
---> TASK 1_2 (only for partner 2) --> TASK 1_2_1
In this case, the task 1_1_1 will only be generated for partner 1 and task 1_2_1
for partner 2
"""
self.configuration.editable_domain = "[('id', 'in', [%s, %s])]" % (
self.partner_01.id,
self.partner_02.id,
)
activity_1 = self.create_server_action()
activity_1_1 = self.create_server_action(
parent_id=activity_1.id, domain="[('id', '=', %s)]" % self.partner_01.id
)
activity_1_2 = self.create_server_action(
parent_id=activity_1.id, domain="[('id', '=', %s)]" % self.partner_02.id
)
activity_1_1_1 = self.create_server_action(parent_id=activity_1_1.id)
activity_1_2_1 = self.create_server_action(parent_id=activity_1_2.id)
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertEqual(
0,
self.env["automation.record.step"].search_count(
[
(
"configuration_step_id",
"in",
(
activity_1_1
| activity_1_2
| activity_1_1_1
| activity_1_2_1
).ids,
)
]
),
)
self.assertTrue(self.partner_01.comment)
self.assertTrue(self.partner_02.comment)
self.env["automation.record.step"]._cron_automation_steps()
self.assertFalse(self.partner_01.comment)
self.assertFalse(self.partner_02.comment)
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_1.id),
("record_id.res_id", "=", self.partner_01.id),
]
),
)
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_2.id),
("record_id.res_id", "=", self.partner_01.id),
]
),
)
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_1.id),
("record_id.res_id", "=", self.partner_02.id),
]
),
)
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_2.id),
("record_id.res_id", "=", self.partner_02.id),
]
),
)
self.assertEqual(
0,
self.env["automation.record.step"].search_count(
[
(
"configuration_step_id",
"in",
(activity_1_1_1 | activity_1_2_1).ids,
)
]
),
)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_1.id),
("record_id.res_id", "=", self.partner_01.id),
("state", "=", "done"),
]
),
)
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_2.id),
("record_id.res_id", "=", self.partner_01.id),
("state", "=", "rejected"),
]
),
)
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_1.id),
("record_id.res_id", "=", self.partner_02.id),
("state", "=", "rejected"),
]
),
)
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_2.id),
("record_id.res_id", "=", self.partner_02.id),
("state", "=", "done"),
]
),
)
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_1_1.id),
("record_id.res_id", "=", self.partner_01.id),
]
),
)
self.assertEqual(
0,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_2_1.id),
("record_id.res_id", "=", self.partner_01.id),
]
),
)
self.assertEqual(
0,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_1_1.id),
("record_id.res_id", "=", self.partner_02.id),
]
),
)
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[
("configuration_step_id", "=", activity_1_2_1.id),
("record_id.res_id", "=", self.partner_02.id),
]
),
)
def test_context(self):
"""
We will check that the context is modified and passed on server actions
"""
self.create_server_action(server_context='{"key_value": "My Value"}')
self.partner_01.comment = False
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertFalse(self.partner_01.comment)
self.env["automation.record.step"]._cron_automation_steps()
self.assertIn("My Value", self.partner_01.comment)

334
odoo-bringout-oca-automation-automation_oca/automation_oca/tests/test_automation_activity.py Normal file
View file

@ -0,0 +1,334 @@
# Copyright 2024 Dixmit
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from odoo.tests import Form, new_test_user
from .common import AutomationTestCase
class TestAutomationActivity(AutomationTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.user = new_test_user(
cls.env,
login="test_user_automation",
groups="base.group_user,base.group_partner_manager",
)
def test_activity_execution(self):
"""
We will check the execution of activity tasks (generation of an activity)
"""
activity = self.create_activity_action()
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertFalse(self.partner_01.activity_ids)
self.env["automation.record.step"]._cron_automation_steps()
self.assertTrue(self.partner_01.activity_ids)
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual(
record_activity, self.partner_01.activity_ids.automation_record_step_id
)
self.assertFalse(record_activity.activity_done_on)
record_activity.invalidate_recordset()
self.assertFalse(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-clock-o"
]
)
self.partner_01.activity_ids.action_feedback()
self.assertTrue(record_activity.activity_done_on)
record_activity.invalidate_recordset()
self.assertTrue(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-clock-o"
]
)
def test_activity_execution_permission(self):
"""
We will check the execution of activity tasks (generation of an activity)
"""
activity = self.create_activity_action()
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertFalse(self.partner_01.activity_ids)
self.env["automation.record.step"]._cron_automation_steps()
self.assertTrue(self.partner_01.activity_ids)
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual(
record_activity, self.partner_01.activity_ids.automation_record_step_id
)
self.assertFalse(record_activity.activity_done_on)
record_activity.invalidate_recordset()
self.assertFalse(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-clock-o"
]
)
self.partner_01.activity_ids.with_user(self.user.id).action_feedback()
self.assertTrue(record_activity.activity_done_on)
record_activity.invalidate_recordset()
self.assertTrue(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-clock-o"
]
)
def test_activity_execution_child(self):
"""
We will check the execution of the child task (activity_done) is only scheduled
after the activity is done
"""
activity = self.create_activity_action()
child_activity = self.create_server_action(
parent_id=activity.id, trigger_type="activity_done"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.env["automation.record.step"]._cron_automation_steps()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual(
record_activity, self.partner_01.activity_ids.automation_record_step_id
)
self.assertFalse(record_activity.activity_done_on)
self.assertFalse(record_child_activity.scheduled_date)
self.partner_01.activity_ids.action_feedback()
self.assertTrue(record_activity.activity_done_on)
self.assertTrue(record_child_activity.scheduled_date)
self.assertFalse(record_child_activity.processed_on)
def test_activity_execution_child_immediate(self):
"""
We will check the execution of the child task (activity_done) is only scheduled
after the activity is done
"""
activity = self.create_activity_action()
child_activity = self.create_server_action(
parent_id=activity.id, trigger_type="activity_done", trigger_interval=-1
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.env["automation.record.step"]._cron_automation_steps()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual(
record_activity, self.partner_01.activity_ids.automation_record_step_id
)
self.assertFalse(record_activity.activity_done_on)
self.assertFalse(record_child_activity.scheduled_date)
self.partner_01.activity_ids.action_feedback()
self.assertTrue(record_activity.activity_done_on)
self.assertTrue(record_child_activity.scheduled_date)
self.assertTrue(record_child_activity.processed_on)
def test_activity_execution_on_cancel(self):
"""
We will check the execution of the child task (activity_done) is only scheduled
after the activity is done
"""
activity = self.create_activity_action()
child_activity = self.create_server_action(
parent_id=activity.id, trigger_type="activity_done"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.env["automation.record.step"]._cron_automation_steps()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual(
record_activity, self.partner_01.activity_ids.automation_record_step_id
)
self.assertFalse(record_activity.activity_done_on)
self.assertFalse(record_child_activity.scheduled_date)
self.partner_01.activity_ids.unlink()
self.assertFalse(record_activity.activity_done_on)
self.assertFalse(record_child_activity.scheduled_date)
self.assertEqual(record_child_activity.state, "rejected")
def test_activity_execution_on_cancel_permission(self):
"""
We will check the execution of the child task (activity_done) is only scheduled
after the activity is done
"""
activity = self.create_activity_action()
child_activity = self.create_server_action(
parent_id=activity.id, trigger_type="activity_done"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.env["automation.record.step"]._cron_automation_steps()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual(
record_activity, self.partner_01.activity_ids.automation_record_step_id
)
self.assertFalse(record_activity.activity_done_on)
self.assertFalse(record_child_activity.scheduled_date)
self.partner_01.activity_ids.with_user(self.user.id).unlink()
self.assertFalse(record_activity.activity_done_on)
self.assertFalse(record_child_activity.scheduled_date)
self.assertEqual(record_child_activity.state, "rejected")
def test_activity_execution_cancel_child(self):
"""
We will check the execution of the child task (activity_cancel) is only scheduled
after the activity is cancel
"""
activity = self.create_activity_action()
child_activity = self.create_server_action(
parent_id=activity.id, trigger_type="activity_cancel"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.env["automation.record.step"]._cron_automation_steps()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual(
record_activity, self.partner_01.activity_ids.automation_record_step_id
)
self.assertFalse(record_activity.activity_cancel_on)
self.assertFalse(record_child_activity.scheduled_date)
self.partner_01.activity_ids.unlink()
self.assertTrue(record_activity.activity_cancel_on)
self.assertTrue(record_child_activity.scheduled_date)
def test_activity_execution_cancel_child_on_done(self):
"""
We will check the execution of the child task (activity_cancel) is not scheduled
after the activity is done
"""
activity = self.create_activity_action()
child_activity = self.create_server_action(
parent_id=activity.id, trigger_type="activity_cancel"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.env["automation.record.step"]._cron_automation_steps()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual(
record_activity, self.partner_01.activity_ids.automation_record_step_id
)
self.assertFalse(record_activity.activity_cancel_on)
self.assertFalse(record_child_activity.scheduled_date)
self.partner_01.activity_ids.action_feedback()
self.assertFalse(record_activity.activity_cancel_on)
self.assertFalse(record_child_activity.scheduled_date)
self.assertEqual(record_child_activity.state, "rejected")
def test_activity_execution_not_done_child_done(self):
"""
We will check the execution of the tasks with activity_not_done is not executed
if it has been done
"""
activity = self.create_activity_action()
child_activity = self.create_server_action(
parent_id=activity.id, trigger_type="activity_not_done"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.env["automation.record.step"]._cron_automation_steps()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual(
record_activity, self.partner_01.activity_ids.automation_record_step_id
)
self.assertFalse(record_activity.activity_done_on)
self.assertTrue(record_child_activity.scheduled_date)
self.partner_01.activity_ids.action_feedback()
self.assertTrue(record_activity.activity_done_on)
self.assertTrue(record_child_activity.scheduled_date)
self.assertEqual("scheduled", record_child_activity.state)
record_child_activity.run()
self.assertEqual("rejected", record_child_activity.state)
def test_activity_execution_not_done_child_not_done(self):
"""
We will check the execution of the tasks with activity_not_done is executed
if it has been not done
"""
activity = self.create_activity_action()
child_activity = self.create_server_action(
parent_id=activity.id, trigger_type="activity_not_done"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.env["automation.record.step"]._cron_automation_steps()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual(
record_activity, self.partner_01.activity_ids.automation_record_step_id
)
self.assertFalse(record_activity.activity_done_on)
self.assertTrue(record_child_activity.scheduled_date)
self.assertEqual("scheduled", record_child_activity.state)
record_child_activity.run()
self.assertEqual("done", record_child_activity.state)
def test_compute_default_values(self):
activity = self.create_server_action()
self.assertFalse(activity.activity_user_id)
with Form(activity) as f:
f.step_type = "activity"
f.activity_type_id = self.activity_type
self.assertTrue(activity.activity_user_id)
with Form(activity) as f:
f.step_type = "action"
f.server_action_id = self.action
self.assertFalse(activity.activity_user_id)

681
odoo-bringout-oca-automation-automation_oca/automation_oca/tests/test_automation_base.py Normal file
View file

@ -0,0 +1,681 @@
# Copyright 2024 Dixmit
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from datetime import datetime
from freezegun import freeze_time
from odoo.exceptions import ValidationError
from odoo.tests import Form
from odoo.tools.safe_eval import safe_eval
from .common import AutomationTestCase
class TestAutomationBase(AutomationTestCase):
def test_no_cron_no_start(self):
"""
We want to check that the system only generates on periodical configurations
"""
self.env["automation.configuration"].cron_automation()
self.assertEqual(
0,
self.env["automation.record"].search_count(
[("configuration_id", "=", self.configuration.id)]
),
)
self.configuration.run_automation()
self.assertEqual(
0,
self.env["automation.record"].search_count(
[("configuration_id", "=", self.configuration.id)]
),
)
def test_no_cron_on_demand(self):
"""
We want to check that the system does not generate using cron
on on demand configurations, but allows manuall execution
"""
self.configuration.is_periodic = False
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertEqual(
0,
self.env["automation.record"].search_count(
[("configuration_id", "=", self.configuration.id)]
),
)
self.configuration.run_automation()
self.assertNotEqual(
0,
self.env["automation.record"].search_count(
[("configuration_id", "=", self.configuration.id)]
),
)
def test_next_execution_date(self):
with freeze_time("2022-01-01"):
self.assertFalse(self.configuration.next_execution_date)
self.env.ref(
"automation_oca.cron_configuration_run"
).nextcall = datetime.now()
self.configuration.start_automation()
self.assertEqual(
self.configuration.next_execution_date, datetime(2022, 1, 1, 0, 0, 0)
)
def test_cron_no_duplicates(self):
"""
We want to check that the records are generated only once, not twice
"""
self.create_server_action()
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
record = self.env["automation.record"].search(
[
("configuration_id", "=", self.configuration.id),
("res_id", "=", self.partner_01.id),
]
)
self.assertEqual(
1,
self.env["automation.record"].search_count(
[
("configuration_id", "=", self.configuration.id),
("res_id", "=", self.partner_01.id),
]
),
)
self.assertEqual(
1,
self.env["automation.record"].search_count(
[
("configuration_id", "=", self.configuration.id),
("res_id", "=", self.partner_02.id),
]
),
)
self.env["automation.configuration"].cron_automation()
self.assertEqual(
1,
self.env["automation.record"].search_count(
[
("configuration_id", "=", self.configuration.id),
("res_id", "=", self.partner_01.id),
]
),
)
self.assertEqual(
1,
self.env["automation.record"].search_count(
[
("configuration_id", "=", self.configuration.id),
("res_id", "=", self.partner_02.id),
]
),
)
record = self.env["automation.record"].search(
[
("configuration_id", "=", self.configuration.id),
("res_id", "=", self.partner_01.id),
]
)
self.assertEqual(
1,
self.env["automation.record.step"].search_count(
[("record_id", "=", record.id)]
),
)
def test_filter(self):
"""
We want to see that the records are only generated for
the records that fulfill the domain
"""
self.create_server_action()
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertEqual(
1,
self.env["automation.record"].search_count(
[
("configuration_id", "=", self.configuration.id),
("res_id", "=", self.partner_01.id),
]
),
)
self.assertEqual(
0,
self.env["automation.record"].search_count(
[
("configuration_id", "=", self.configuration.id),
("res_id", "=", self.partner_02.id),
]
),
)
def test_exception(self):
"""
Check that the error is raised properly and stored the full error
"""
activity = self.create_server_action(server_action_id=self.error_action.id)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
record = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertFalse(record.error_trace)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual(record.state, "error")
self.assertTrue(record.error_trace)
def test_record_resource_information(self):
"""
Check the record computed fields of record
"""
self.create_server_action(server_action_id=self.error_action.id)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
record = self.env["automation.record"].search(
[("configuration_id", "=", self.configuration.id)]
)
self.assertEqual(self.partner_01.display_name, record.display_name)
self.assertEqual(self.partner_01, record.resource_ref)
record.model = "unexistent.model"
self.assertFalse(record.resource_ref)
def test_expiry(self):
"""
Testing that expired actions are not executed
"""
activity = self.create_server_action(expiry=True, trigger_interval=1)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual("scheduled", record_activity.state)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("expired", record_activity.state)
def test_cancel(self):
"""
Testing that cancelled actions are not executed
"""
activity = self.create_server_action()
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual("scheduled", record_activity.state)
record_activity.cancel()
self.assertEqual("cancel", record_activity.state)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("cancel", record_activity.state)
def test_counter(self):
"""
Check the counter function
"""
self.create_server_action(server_action_id=self.error_action.id)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.assertEqual(0, self.configuration.record_count)
self.assertEqual(0, self.configuration.record_test_count)
self.env["automation.configuration"].cron_automation()
self.configuration.invalidate_recordset()
self.assertEqual(1, self.configuration.record_count)
self.assertEqual(0, self.configuration.record_test_count)
def test_start_configuration_twice_exception(self):
"""
Check that we cannot start automation twice
"""
self.configuration.start_automation()
with self.assertRaises(ValidationError):
self.configuration.start_automation()
def test_state_automation_management(self):
"""
Testing the change of state
Draft -> Run -> Stop -> Draft
"""
self.configuration.start_automation()
self.assertEqual(self.configuration.state, "periodic")
self.configuration.done_automation()
self.assertEqual(self.configuration.state, "done")
self.env["automation.configuration"].cron_automation()
self.assertFalse(
self.env["automation.record"].search(
[
("configuration_id", "=", self.configuration.id),
]
)
)
self.configuration.back_to_draft()
self.assertEqual(self.configuration.state, "draft")
def test_graph(self):
"""
Checking the graph results.
We will use 2 parent actions (1 will fail) and a child action of the one ok.
After 2 executions, we should have (1 OK, 0 Errors) for parent and child and
(0 OK, 1 Error) for the failing one.
"""
activity_01 = self.create_server_action()
activity_02 = self.create_server_action(server_action_id=self.error_action.id)
activity_03 = self.create_mail_activity()
child_activity = self.create_server_action(parent_id=activity_01.id)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertEqual(0, self.configuration.activity_mail_count)
self.assertEqual(0, self.configuration.activity_action_count)
self.assertEqual(0, activity_01.graph_done)
self.assertEqual(0, activity_01.graph_error)
self.assertEqual(0, sum(d["y"] for d in activity_01.graph_data["done"]))
self.assertEqual(0, sum(d["y"] for d in activity_01.graph_data["error"]))
self.assertEqual(0, activity_02.graph_done)
self.assertEqual(0, activity_02.graph_error)
self.assertEqual(0, sum(d["y"] for d in activity_02.graph_data["done"]))
self.assertEqual(0, sum(d["y"] for d in activity_02.graph_data["error"]))
self.assertEqual(0, activity_03.graph_done)
self.assertEqual(0, activity_03.graph_error)
self.assertEqual(0, sum(d["y"] for d in activity_03.graph_data["done"]))
self.assertEqual(0, sum(d["y"] for d in activity_03.graph_data["error"]))
self.assertEqual(0, child_activity.graph_done)
self.assertEqual(0, child_activity.graph_error)
self.assertEqual(0, sum(d["y"] for d in child_activity.graph_data["done"]))
self.assertEqual(0, sum(d["y"] for d in child_activity.graph_data["error"]))
self.env["automation.record.step"]._cron_automation_steps()
self.configuration.invalidate_recordset()
self.assertEqual(1, self.configuration.activity_mail_count)
self.assertEqual(1, self.configuration.activity_action_count)
activity_01.invalidate_recordset()
self.assertEqual(1, activity_01.graph_done)
self.assertEqual(0, activity_01.graph_error)
self.assertEqual(1, sum(d["y"] for d in activity_01.graph_data["done"]))
self.assertEqual(0, sum(d["y"] for d in activity_01.graph_data["error"]))
activity_02.invalidate_recordset()
self.assertEqual(0, activity_02.graph_done)
self.assertEqual(1, activity_02.graph_error)
self.assertEqual(0, sum(d["y"] for d in activity_02.graph_data["done"]))
self.assertEqual(1, sum(d["y"] for d in activity_02.graph_data["error"]))
activity_03.invalidate_recordset()
self.assertEqual(1, activity_03.graph_done)
self.assertEqual(0, activity_03.graph_error)
self.assertEqual(1, sum(d["y"] for d in activity_03.graph_data["done"]))
self.assertEqual(0, sum(d["y"] for d in activity_03.graph_data["error"]))
child_activity.invalidate_recordset()
self.assertEqual(0, child_activity.graph_done)
self.assertEqual(0, child_activity.graph_error)
self.assertEqual(0, sum(d["y"] for d in child_activity.graph_data["done"]))
self.assertEqual(0, sum(d["y"] for d in child_activity.graph_data["error"]))
self.env["automation.record.step"]._cron_automation_steps()
self.configuration.invalidate_recordset()
self.assertEqual(1, self.configuration.activity_mail_count)
self.assertEqual(2, self.configuration.activity_action_count)
activity_01.invalidate_recordset()
self.assertEqual(1, activity_01.graph_done)
self.assertEqual(0, activity_01.graph_error)
self.assertEqual(1, sum(d["y"] for d in activity_01.graph_data["done"]))
self.assertEqual(0, sum(d["y"] for d in activity_01.graph_data["error"]))
activity_02.invalidate_recordset()
self.assertEqual(0, activity_02.graph_done)
self.assertEqual(1, activity_02.graph_error)
self.assertEqual(0, sum(d["y"] for d in activity_02.graph_data["done"]))
self.assertEqual(1, sum(d["y"] for d in activity_02.graph_data["error"]))
activity_03.invalidate_recordset()
self.assertEqual(1, activity_03.graph_done)
self.assertEqual(0, activity_03.graph_error)
self.assertEqual(1, sum(d["y"] for d in activity_03.graph_data["done"]))
self.assertEqual(0, sum(d["y"] for d in activity_03.graph_data["error"]))
child_activity.invalidate_recordset()
self.assertEqual(1, child_activity.graph_done)
self.assertEqual(0, child_activity.graph_error)
self.assertEqual(1, sum(d["y"] for d in child_activity.graph_data["done"]))
self.assertEqual(0, sum(d["y"] for d in child_activity.graph_data["error"]))
def test_schedule_date_computation_hours(self):
with freeze_time("2022-01-01"):
activity = self.create_server_action(trigger_interval=1)
self.assertEqual(1, activity.trigger_interval_hours)
self.configuration.editable_domain = (
"[('id', '=', %s)]" % self.partner_01.id
)
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual("scheduled", record_activity.state)
self.assertEqual(
record_activity.scheduled_date, datetime(2022, 1, 1, 1, 0, 0, 0)
)
def test_schedule_date_force(self):
partner_01 = self.env["res.partner"].create(
{
"name": "Demo partner",
"comment": "Demo",
"email": "test@test.com",
"date": "2025-01-01",
}
)
with freeze_time("2024-01-01 00:00:00"):
activity = self.create_server_action(
trigger_date_kind="date",
trigger_date_field_id=self.env["ir.model.fields"]
.search(
[
("name", "=", "date"),
("model", "=", "res.partner"),
]
)
.id,
trigger_interval=1,
trigger_interval_type="days",
)
self.configuration.editable_domain = "[('id', '=', %s)]" % partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual("scheduled", record_activity.state)
self.assertEqual(record_activity.scheduled_date, datetime(2025, 1, 2))
def test_schedule_date_computation_days(self):
with freeze_time("2022-01-01"):
activity = self.create_server_action(
trigger_interval=1, trigger_interval_type="days"
)
self.assertEqual(24, activity.trigger_interval_hours)
self.configuration.editable_domain = (
"[('id', '=', %s)]" % self.partner_01.id
)
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual("scheduled", record_activity.state)
self.assertEqual(
record_activity.scheduled_date, datetime(2022, 1, 2, 0, 0, 0, 0)
)
def test_onchange_activity_trigger_type(self):
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(parent_id=activity.id)
self.assertEqual(child_activity.trigger_type, "after_step")
self.assertTrue(child_activity.parent_id)
with Form(child_activity) as f:
f.trigger_type = "mail_bounce"
self.assertTrue(f.parent_id)
def test_onchange_activity_trigger_type_start(self):
activity = self.create_server_action()
child_activity = self.create_server_action(parent_id=activity.id)
self.assertEqual(child_activity.trigger_type, "after_step")
self.assertTrue(child_activity.parent_id)
with Form(child_activity) as f:
f.trigger_type = "start"
self.assertFalse(f.parent_id)
def test_field_not_field_unicity(self):
self.configuration.editable_domain = (
"[('id', 'in', %s)]" % (self.partner_01 | self.partner_02).ids
)
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertEqual(
2,
len(
self.env["automation.record"].search(
[("configuration_id", "=", self.configuration.id)]
)
),
)
def test_field_field_unicity(self):
self.configuration.editable_domain = (
"[('id', 'in', %s)]" % (self.partner_01 | self.partner_02).ids
)
self.configuration.field_id = self.env.ref("base.field_res_partner__email")
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.assertEqual(
1,
len(
self.env["automation.record"].search(
[("configuration_id", "=", self.configuration.id)]
)
),
)
self.partner_01.email = "t" + self.partner_01.email
self.env["automation.configuration"].cron_automation()
self.assertEqual(
2,
len(
self.env["automation.record"].search(
[("configuration_id", "=", self.configuration.id)]
)
),
)
def test_configuration_filter_domain(self):
domain = [("partner_id", "=", self.partner_01.id)]
self.assertFalse(self.configuration.filter_id)
self.configuration.editable_domain = domain
self.configuration.save_filter()
self.assertTrue(self.configuration.filter_id)
self.assertEqual(self.configuration.model_id, self.configuration.model_id)
domain = [("partner_id", "=", self.partner_02.id)]
self.configuration.invalidate_recordset()
self.assertNotEqual(domain, safe_eval(self.configuration.domain))
self.configuration.filter_id.domain = domain
self.assertEqual(domain, safe_eval(self.configuration.domain))
with Form(self.env["automation.configuration"]) as f:
self.assertFalse(f.filter_domain)
f.name = "My other configuration"
f.filter_id = self.configuration.filter_id
self.assertEqual(f.model_id, self.env.ref("base.model_res_partner"))
self.assertIn(
self.configuration.filter_id,
self.env["automation.filter"].search(f.filter_domain),
)
f.model_id = self.env.ref("base.model_res_users")
self.assertFalse(f.filter_id)
def test_filter_onchange(self):
with Form(self.env["automation.filter"]) as f:
f.name = "My other configuration"
f.model_id = self.env.ref("base.model_res_partner")
f.domain = [("id", "=", 1)]
f.model_id = self.env.ref("base.model_res_users")
self.assertFalse(safe_eval(f.domain))
def test_constrains_mail(self):
activity = self.create_server_action()
with self.assertRaises(ValidationError):
self.create_server_action(parent_id=activity.id, trigger_type="mail_bounce")
def test_constrains_start_with_parent(self):
activity = self.create_server_action()
with self.assertRaises(ValidationError):
self.create_server_action(parent_id=activity.id, trigger_type="start")
def test_constrains_no_start_without_parent(self):
with self.assertRaises(ValidationError):
self.create_server_action(parent_id=False, trigger_type="after_step")
def test_constrains_wrong_context(self):
with self.assertRaises(ValidationError):
self.create_server_action(server_context="{not a json}")
def test_is_test_behavior(self):
"""
We want to ensure that no mails are sent on tests
"""
self.create_server_action()
with Form(
self.env["automation.configuration.test"].with_context(
default_configuration_id=self.configuration.id,
defaul_model=self.configuration.model,
)
) as f:
self.assertTrue(f.resource_ref)
f.resource_ref = "%s,%s" % (self.partner_01._name, self.partner_01.id)
wizard = f.save()
wizard_action = wizard.test_record()
record = self.env[wizard_action["res_model"]].browse(wizard_action["res_id"])
self.assertEqual(self.configuration, record.configuration_id)
self.assertEqual(1, self.configuration.record_test_count)
self.assertEqual(0, self.configuration.record_count)
def test_check_icons(self):
action = self.create_server_action()
mail = self.create_mail_activity()
activity = self.create_activity_action()
self.assertEqual(action.step_icon, "fa fa-cogs")
self.assertEqual(mail.step_icon, "fa fa-envelope")
self.assertEqual(activity.step_icon, "fa fa-clock-o")
def test_trigger_types(self):
action = self.create_server_action()
child = self.create_server_action(parent_id=action.id)
self.assertTrue(action.trigger_type_data["allow_parent"])
self.assertFalse(child.trigger_type_data.get("allow_parent", False))
def test_trigger_childs(self):
action = self.create_server_action()
mail = self.create_mail_activity()
activity = self.create_activity_action()
self.assertEqual(1, len(action.trigger_child_types))
self.assertEqual({"after_step"}, set(action.trigger_child_types.keys()))
self.assertEqual(8, len(mail.trigger_child_types))
self.assertEqual(
{
"after_step",
"mail_open",
"mail_not_open",
"mail_reply",
"mail_not_reply",
"mail_click",
"mail_not_clicked",
"mail_bounce",
},
set(mail.trigger_child_types.keys()),
)
self.assertEqual(4, len(activity.trigger_child_types))
self.assertEqual(
{"after_step", "activity_done", "activity_not_done", "activity_cancel"},
set(activity.trigger_child_types.keys()),
)
def test_search(self):
configuration_2 = self.env["automation.configuration"].create(
{
"name": "Test configuration",
"model_id": self.env.ref("base.model_res_partner").id,
"is_periodic": True,
}
)
self.create_server_action()
self.create_server_action(configuration_id=configuration_2.id)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
configuration_2.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
configuration_2.start_automation()
self.env["automation.configuration"].cron_automation()
record_activity = self.env["automation.record"].search(
[("model", "=", self.partner_01._name), ("res_id", "=", self.partner_01.id)]
)
self.assertEqual(2, len(record_activity))
def test_generation_orphan_record(self):
self.configuration.editable_domain = (
"['|', ('id', '=', %s), ('id', '=', %s)]"
% (self.partner_01.id, self.partner_02.id)
)
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
self.partner_01.unlink()
records = self.env["automation.record"].search(
[("configuration_id", "=", self.configuration.id), ("is_test", "=", False)]
)
self.configuration._compute_record_count()
self.assertEqual(len(records), 2, "Seems like no orphan record was created")
orphan_record_found = any(record.name == "Orphan Record" for record in records)
self.assertTrue(
orphan_record_found, "No record named 'Orphan Record' was found"
)
def test_delete_step_executed(self):
"""
Testing that deleting a step will keep the results of the executed related steps
"""
activity = self.create_server_action()
child_activity = self.create_server_action(parent_id=activity.id)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual("scheduled", record_activity.state)
self.assertFalse(
self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("done", record_activity.state)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("scheduled", record_child_activity.state)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("done", record_child_activity.state)
child_activity.unlink()
child_activity.flush_recordset()
self.assertEqual("action", record_child_activity.step_type)
def test_delete_step_to_execute(self):
"""
Testing that deleting a step will make pending actions related
to be rejected
"""
activity = self.create_server_action()
child_activity = self.create_server_action(parent_id=activity.id)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual("scheduled", record_activity.state)
self.assertFalse(
self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("done", record_activity.state)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("scheduled", record_child_activity.state)
child_activity.unlink()
child_activity.flush_recordset()
self.assertEqual("action", record_child_activity.step_type)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("rejected", record_child_activity.state)

568
odoo-bringout-oca-automation-automation_oca/automation_oca/tests/test_automation_mail.py Normal file
View file

@ -0,0 +1,568 @@
# Copyright 2024 Dixmit
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from odoo import tools
from odoo.tests.common import Form, HttpCase
from odoo.addons.mail.tests.common import MockEmail
from .common import AutomationTestCase
MAIL_TEMPLATE = """Return-Path: <whatever-2a840@postmaster.twitter.com>
To: {to}
cc: {cc}
Received: by mail1.openerp.com (Postfix, from userid 10002)
id 5DF9ABFB2A; Fri, 10 Aug 2012 16:16:39 +0200 (CEST)
From: {email_from}
Subject: {subject}
MIME-Version: 1.0
Content-Type: multipart/alternative;
boundary="----=_Part_4200734_24778174.1344608186754"
Date: Fri, 10 Aug 2012 14:16:26 +0000
Message-ID: {msg_id}
{extra}
------=_Part_4200734_24778174.1344608186754
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: quoted-printable
I would gladly answer to your mass mailing !
--
Your Dear Customer
------=_Part_4200734_24778174.1344608186754
Content-Type: text/html; charset=utf-8
Content-Transfer-Encoding: quoted-printable
<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01//EN" "http://www.w3.org/TR/html4/strict.dtd">
<html>
<head>=20
<meta http-equiv=3D"Content-Type" content=3D"text/html; charset=3Dutf-8" />
</head>=20
<body style=3D"margin: 0; padding: 0; background: #ffffff;-webkit-text-size-adjust: 100%;">=20
<p>I would gladly answer to your mass mailing !</p>
<p>--<br/>
Your Dear Customer
<p>
</body>
</html>
------=_Part_4200734_24778174.1344608186754--
"""
class TestAutomationMail(AutomationTestCase, MockEmail, HttpCase):
def test_activity_execution(self):
"""
We will check the execution of the tasks and that we cannot execute them again
"""
activity = self.create_mail_activity()
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
messages_01 = self.partner_01.message_ids
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
self.assertEqual(1, len(record_activity))
self.assertEqual("done", record_activity.state)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(self.partner_01.message_ids - messages_01)
def test_bounce(self):
"""
Now we will check the execution of scheduled activities"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_bounce"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(record_child_activity)
self.assertFalse(record_child_activity.scheduled_date)
parsed_bounce_values = {
"email_from": "some.email@external.example.com",
"to": "bounce@test.example.com",
"message_id": tools.generate_tracking_message_id("MailTest"),
"bounced_partner": self.env["res.partner"].sudo(),
"bounced_message": self.env["mail.message"].sudo(),
"bounced_email": "",
"bounced_msg_id": [record_activity.message_id],
}
record_activity.invalidate_recordset()
self.assertFalse(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-exclamation-circle"
]
)
self.env["mail.thread"]._routing_handle_bounce(False, parsed_bounce_values)
self.assertEqual("bounce", record_activity.mail_status)
self.assertTrue(record_child_activity.scheduled_date)
record_activity.invalidate_recordset()
self.assertTrue(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-exclamation-circle"
]
)
def test_reply(self):
"""
Now we will check the execution of scheduled activities"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_reply"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(record_child_activity)
self.assertFalse(record_child_activity.scheduled_date)
record_activity.invalidate_recordset()
self.assertFalse(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-reply"
]
)
self.gateway_mail_reply_wrecord(
MAIL_TEMPLATE, self.partner_01, use_in_reply_to=True
)
self.assertEqual("reply", record_activity.mail_status)
self.assertTrue(record_child_activity.scheduled_date)
record_activity.invalidate_recordset()
self.assertTrue(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-reply"
]
)
def test_no_reply(self):
"""
Now we will check the not reply validation. To remember:
if it is not opened, the schedule date of the child task will be false
"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_not_reply"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(record_child_activity)
self.assertFalse(record_child_activity.scheduled_date)
self.url_open(record_activity._get_mail_tracking_url())
self.assertEqual("open", record_activity.mail_status)
self.assertTrue(record_child_activity.scheduled_date)
self.gateway_mail_reply_wrecord(
MAIL_TEMPLATE, self.partner_01, use_in_reply_to=True
)
self.assertEqual("reply", record_activity.mail_status)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("rejected", record_child_activity.state)
def test_open(self):
"""
Now we will check the execution of scheduled activities"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_open"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(record_child_activity)
self.assertFalse(record_child_activity.scheduled_date)
record_activity.invalidate_recordset()
self.assertFalse(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-envelope-open-o"
]
)
self.url_open(record_activity._get_mail_tracking_url())
self.assertEqual("open", record_activity.mail_status)
self.assertTrue(record_child_activity.scheduled_date)
record_activity.invalidate_recordset()
self.assertTrue(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-envelope-open-o"
]
)
def test_open_wrong_code(self):
"""
We wan to ensure that the code is checked on the call
"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_open"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(record_child_activity)
self.assertFalse(record_child_activity.scheduled_date)
self.url_open(
"/automation_oca/track/%s/INVENTED_CODE/blank.gif" % record_activity.id
)
self.assertEqual("sent", record_activity.mail_status)
self.assertFalse(record_child_activity.scheduled_date)
def test_no_open(self):
"""
Now we will check the not open validation when it is not opened (should be executed)
"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_not_open"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(record_child_activity)
self.assertTrue(record_child_activity.scheduled_date)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("done", record_child_activity.state)
def test_no_open_rejected(self):
"""
Now we will check the not open validation when it was already opened (rejection)
"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_not_open"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(record_child_activity)
self.assertTrue(record_child_activity.scheduled_date)
self.url_open(record_activity._get_mail_tracking_url())
self.assertEqual("open", record_activity.mail_status)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("rejected", record_child_activity.state)
def test_click(self):
"""
Now we will check the execution of scheduled activities that should happen
after a click
"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_click"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.env["link.tracker"].search(
[("url", "=", "https://www.twitter.com")]
).unlink()
self.configuration.start_automation()
self.assertEqual(0, self.configuration.click_count)
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.configuration.invalidate_recordset()
self.assertEqual(0, self.configuration.click_count)
self.assertTrue(record_child_activity)
self.assertFalse(record_child_activity.scheduled_date)
self.url_open(record_activity._get_mail_tracking_url())
self.assertEqual("open", record_activity.mail_status)
self.configuration.invalidate_recordset()
self.assertEqual(0, self.configuration.click_count)
self.assertFalse(record_child_activity.scheduled_date)
record_activity.invalidate_recordset()
self.assertFalse(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-hand-pointer-o"
]
)
tracker = self.env["link.tracker"].search(
[("url", "=", "https://www.twitter.com")]
)
self.assertTrue(tracker)
self.url_open(
"/r/%s/au/%s/%s"
% (
tracker.code,
record_activity.id,
record_activity._get_mail_tracking_token(),
)
)
self.assertEqual("open", record_activity.mail_status)
self.assertEqual(
1,
self.env["link.tracker.click"].search_count(
[
("automation_record_step_id", "=", record_activity.id),
("link_id", "=", tracker.id),
]
),
)
record_activity.invalidate_recordset()
self.assertTrue(
[
step
for step in record_activity.step_actions
if step["done"] and step["icon"] == "fa fa-hand-pointer-o"
]
)
self.assertTrue(record_child_activity.scheduled_date)
self.configuration.invalidate_recordset()
self.assertEqual(1, self.configuration.click_count)
# Now we will check that a second click does not generate a second log
self.url_open(
"/r/%s/au/%s/%s"
% (
tracker.code,
record_activity.id,
record_activity._get_mail_tracking_token(),
)
)
self.assertEqual(
1,
self.env["link.tracker.click"].search_count(
[
("automation_record_step_id", "=", record_activity.id),
("link_id", "=", tracker.id),
]
),
)
self.configuration.invalidate_recordset()
self.assertEqual(1, self.configuration.click_count)
def test_click_wrong_url(self):
"""
Now we will check that no log is processed when the clicked url is malformed.
That happens because we add a code information on the URL.
"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_click"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(record_child_activity)
self.assertFalse(record_child_activity.scheduled_date)
tracker = self.env["link.tracker"].search(
[("url", "=", "https://www.twitter.com")]
)
self.assertTrue(tracker)
self.url_open(
"/r/%s/au/%s/1234"
% (
tracker.code,
record_activity.id,
)
)
self.assertEqual("sent", record_activity.mail_status)
self.assertFalse(record_child_activity.scheduled_date)
# Now we check the case where the code is not found
tracker.unlink()
self.url_open(
"/r/%s/au/%s/%s"
% (
tracker.code,
record_activity.id,
record_activity._get_mail_tracking_token(),
)
)
self.assertEqual("sent", record_activity.mail_status)
self.assertFalse(record_child_activity.scheduled_date)
def test_no_click(self):
"""
Checking the not clicked validation when it is not clicked (should be executed)
"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_not_clicked"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(record_child_activity)
self.assertFalse(record_child_activity.scheduled_date)
self.url_open(record_activity._get_mail_tracking_url())
self.assertEqual("open", record_activity.mail_status)
self.assertTrue(record_child_activity.scheduled_date)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("done", record_child_activity.state)
def test_no_click_rejected(self):
"""
Checking the not clicked validation when it was already clicked
"""
activity = self.create_mail_activity()
child_activity = self.create_mail_activity(
parent_id=activity.id, trigger_type="mail_not_clicked"
)
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
self.configuration.start_automation()
self.env["automation.configuration"].cron_automation()
with self.mock_mail_gateway():
self.env["automation.record.step"]._cron_automation_steps()
self.assertSentEmail(self.env.user.partner_id, [self.partner_01])
record_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", activity.id)]
)
record_child_activity = self.env["automation.record.step"].search(
[("configuration_step_id", "=", child_activity.id)]
)
self.assertEqual("sent", record_activity.mail_status)
self.assertTrue(record_child_activity)
self.assertFalse(record_child_activity.scheduled_date)
self.url_open(record_activity._get_mail_tracking_url())
self.assertEqual("open", record_activity.mail_status)
self.assertTrue(record_child_activity.scheduled_date)
tracker = self.env["link.tracker"].search(
[("url", "=", "https://www.twitter.com")]
)
self.url_open(
"/r/%s/au/%s/%s"
% (
tracker.code,
record_activity.id,
record_activity._get_mail_tracking_token(),
)
)
self.env["automation.record.step"]._cron_automation_steps()
self.assertEqual("rejected", record_child_activity.state)
def test_is_test_behavior(self):
"""
We want to ensure that no mails are sent on tests
"""
self.create_mail_activity()
self.configuration.editable_domain = "[('id', '=', %s)]" % self.partner_01.id
with Form(
self.env["automation.configuration.test"].with_context(
default_configuration_id=self.configuration.id,
defaul_model=self.configuration.model,
)
) as f:
self.assertTrue(f.resource_ref)
f.resource_ref = "%s,%s" % (self.partner_01._name, self.partner_01.id)
wizard = f.save()
wizard_action = wizard.test_record()
record = self.env[wizard_action["res_model"]].browse(wizard_action["res_id"])
self.assertTrue(record)
self.assertEqual("scheduled", record.automation_step_ids.state)
self.assertFalse(record.automation_step_ids.mail_status)
with self.mock_mail_gateway():
record.automation_step_ids.run()
self.assertNotSentEmail()
self.assertEqual("sent", record.automation_step_ids.mail_status)

106
odoo-bringout-oca-automation-automation_oca/automation_oca/tests/test_automation_security.py Normal file
View file

@ -0,0 +1,106 @@
# Copyright 2024 Dixmit
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from odoo.tests.common import users
from odoo.addons.mail.tests.common import mail_new_test_user
from .common import AutomationTestCase
class TestAutomationSecurity(AutomationTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
# Removing rules in order to check only what we expect
cls.env["ir.rule"].search(
[("model_id", "=", cls.env.ref("base.model_res_partner").id)]
).toggle_active()
cls.user_automation_01 = mail_new_test_user(
cls.env,
login="user_automation_01",
name="User automation 01",
email="user_automation_01@test.example.com",
company_id=cls.env.user.company_id.id,
notification_type="inbox",
groups="base.group_user,automation_oca.group_automation_user",
)
cls.user_automation_02 = mail_new_test_user(
cls.env,
login="user_automation_02",
name="User automation 01",
email="user_automation_02@test.example.com",
company_id=cls.env.user.company_id.id,
notification_type="inbox",
groups="base.group_user,automation_oca.group_automation_user",
)
cls.group_1 = cls.env["res.groups"].create(
{
"name": "G1",
"users": [(4, cls.user_automation_01.id)],
"rule_groups": [
(
0,
0,
{
"name": "Rule 01",
"model_id": cls.env.ref("base.model_res_partner").id,
"domain_force": "[('id', '!=', %s)]" % cls.partner_01.id,
},
)
],
}
)
cls.group_2 = cls.env["res.groups"].create(
{
"name": "G2",
"users": [(4, cls.user_automation_02.id)],
"rule_groups": [
(
0,
0,
{
"name": "Rule 01",
"model_id": cls.env.ref("base.model_res_partner").id,
"domain_force": "[('id', '!=', %s)]" % cls.partner_02.id,
},
)
],
}
)
cls.configuration.editable_domain = [
("id", "in", (cls.partner_01 | cls.partner_02).ids)
]
cls.configuration.start_automation()
cls.env["automation.configuration"].cron_automation()
@users("user_automation_01")
def test_security_01(self):
record = self.env["automation.record"].search(
[("configuration_id", "=", self.configuration.id)]
)
self.assertEqual(1, len(record))
self.assertEqual(self.partner_02, record.resource_ref)
@users("user_automation_02")
def test_security_02(self):
record = self.env["automation.record"].search(
[("configuration_id", "=", self.configuration.id)]
)
self.assertEqual(1, len(record))
self.assertEqual(self.partner_01, record.resource_ref)
@users("user_automation_01")
def test_security_deleted_record(self):
original_record = self.env["automation.record"].search(
[("configuration_id", "=", self.configuration.id)]
)
self.partner_02.unlink()
record = self.env["automation.record"].search(
[("configuration_id", "=", self.configuration.id)]
)
self.assertTrue(record)
self.assertTrue(record.is_orphan_record)
self.assertTrue(original_record)
self.assertTrue(original_record.read())