Coverage for ingadhoc-odoo-saas-adhoc / saas_provider_upgrade / tests / test_execution_mode.py: 100%
100 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 20:33 +0000
1from contextlib import contextmanager
2from unittest.mock import MagicMock, PropertyMock, patch
4from odoo import Command
5from odoo.exceptions import UserError
7from .test_provider_upgrade_base import TestProviderUpgradeCommon
10class TestExecutionMode(TestProviderUpgradeCommon):
11 """Test execution modes for upgrade lines (odooly vs odoo_shell)."""
13 @contextmanager
14 def mock_eval_context(self, eval_context, upgrade_line=None):
15 """Mock context manager to avoid database connections."""
16 yield
18 @classmethod
19 def setUpClass(cls):
20 super().setUpClass()
22 # Create test request
23 cls.test_request = cls.env["helpdesk.ticket.upgrade.request"]._create_request(cls.upgrade_ticket.id, "test")
25 # Create upgrade line with odooly mode (default)
26 cls.ul_odooly = cls.env["saas.upgrade.line"].create(
27 {
28 "name": "[Test] UL with odooly mode",
29 "upgrade_type_ids": [Command.link(cls.upgrade_type.id)],
30 "type": "4_post",
31 "state": "approved",
32 "adhoc_product_id": cls.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id,
33 }
34 )
35 ul_odooly_script = cls.env["saas.upgrade.line.script"].create(
36 {
37 "upgrade_line_id": cls.ul_odooly.id,
38 "version": 1,
39 "script": "result = 'odooly executed'",
40 "execution_mode": "odooly",
41 }
42 )
43 cls.ul_odooly.actual_script_id = ul_odooly_script
45 # Create upgrade line with odoo_shell mode
46 cls.ul_odoo_shell = cls.env["saas.upgrade.line"].create(
47 {
48 "name": "[Test] UL with odoo_shell mode",
49 "upgrade_type_ids": [Command.link(cls.upgrade_type.id)],
50 "type": "4_post",
51 "state": "approved",
52 "adhoc_product_id": cls.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id,
53 }
54 )
55 ul_odoo_shell_script = cls.env["saas.upgrade.line.script"].create(
56 {
57 "upgrade_line_id": cls.ul_odoo_shell.id,
58 "version": 1,
59 "script": "result = 'odoo_shell executed'",
60 "execution_mode": "odoo_shell",
61 }
62 )
63 cls.ul_odoo_shell.actual_script_id = ul_odoo_shell_script
65 def test_execution_mode_default_is_odooly(self):
66 """Test that default execution_mode is odooly."""
67 ul = self.env["saas.upgrade.line"].create(
68 {
69 "name": "[Test] Default execution mode",
70 "upgrade_type_ids": [Command.link(self.upgrade_type.id)],
71 "type": "4_post",
72 "adhoc_product_id": self.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id,
73 }
74 )
75 self.assertEqual(ul.dev_execution_mode, "odooly")
76 self.assertFalse(ul.execution_mode)
78 def test_run_odooly_with_breaks(self):
79 """Test that _run_odooly handles breaks correctly."""
80 ul = self.env["saas.upgrade.line"].create(
81 {
82 "name": "[Test] UL with breaks",
83 "upgrade_type_ids": [Command.link(self.upgrade_type.id)],
84 "type": "4_post",
85 "state": "approved",
86 "adhoc_product_id": self.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id,
87 }
88 )
89 request = self.test_request.with_context(test_dev=True)
90 run_info = self.env["saas.upgrade.line.request.run"].create(
91 {
92 "upgrade_line_id": ul.id,
93 "request_id": request.id,
94 }
95 )
97 # Mock _get_eval_context and suppress _commit_progress to avoid DB commits
98 with patch.object(type(request), "_get_eval_context", self.mock_eval_context):
99 with patch.object(type(self.env["ir.cron"]), "_commit_progress", return_value=None):
100 # In test_dev mode the method returns a formatted Markup
101 # notification. Assert the expected text is present.
102 run_info.start(None)
103 result = ul.with_context(test_dev=True)._run_odooly(request, run_info, "breaks = 'Test break message'")
104 self.assertIn("Test break message", str(result))
106 def test_run_odooly_with_breaks_not_test_dev(self):
107 """Test that _run_odooly raises UserError on breaks when not in test_dev."""
108 ul = self.env["saas.upgrade.line"].create(
109 {
110 "name": "[Test] UL with breaks no test_dev",
111 "upgrade_type_ids": [Command.link(self.upgrade_type.id)],
112 "type": "4_post",
113 "state": "approved",
114 "adhoc_product_id": self.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id,
115 }
116 )
117 request = self.test_request
118 run_info = self.env["saas.upgrade.line.request.run"].create(
119 {
120 "upgrade_line_id": ul.id,
121 "request_id": request.id,
122 }
123 )
125 # Mock _get_eval_context and suppress _commit_progress to avoid DB commits
126 script = "breaks = 'Test break message'"
127 with patch.object(type(request), "_get_eval_context", self.mock_eval_context):
128 with patch.object(type(self.env["ir.cron"]), "_commit_progress", return_value=None):
129 # Without test_dev, breaks should raise UserError
130 with self.assertRaises(UserError) as ctx:
131 run_info.start(None)
132 ul._run_odooly(request, run_info, script)
133 self.assertIn("Test break message", str(ctx.exception))
135 def test_run_odooly_with_result(self):
136 """Test that _run_odooly returns result correctly."""
137 ul = self.env["saas.upgrade.line"].create(
138 {
139 "name": "[Test] UL with result",
140 "upgrade_type_ids": [Command.link(self.upgrade_type.id)],
141 "type": "4_post",
142 "state": "approved",
143 "adhoc_product_id": self.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id,
144 }
145 )
146 request = self.test_request.with_context(test_dev=True)
147 run_info = self.env["saas.upgrade.line.request.run"].create(
148 {
149 "upgrade_line_id": ul.id,
150 "request_id": request.id,
151 }
152 )
154 # Mock _get_eval_context and suppress _commit_progress to avoid DB commits
155 script = "result = 'Test result message'"
156 with patch.object(type(request), "_get_eval_context", self.mock_eval_context):
157 with patch.object(type(self.env["ir.cron"]), "_commit_progress", return_value=None):
158 run_info.start(None)
159 result = ul.with_context(test_dev=True)._run_odooly(request, run_info, script)
160 self.assertIn("Test result message", str(result))
162 def test_run_odoo_shell_with_upgraded_db(self):
163 """Test that _run_odoo_shell schedules task on upgraded database for post type."""
164 ul = self.env["saas.upgrade.line"].create(
165 {
166 "name": "[Test] UL odoo_shell post",
167 "upgrade_type_ids": [Command.link(self.upgrade_type.id)],
168 "type": "4_post",
169 "state": "approved",
170 "adhoc_product_id": self.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id,
171 }
172 )
173 request = self.test_request
174 run_info = self.env["saas.upgrade.line.request.run"].create(
175 {
176 "upgrade_line_id": ul.id,
177 "request_id": request.id,
178 }
179 )
181 mock_db = MagicMock()
182 script = "result = 'odoo_shell executed'"
184 # Mock upgraded_database_id and _schedule_task
185 with patch.object(type(request), "upgraded_database_id", PropertyMock(return_value=mock_db)):
186 with patch.object(mock_db, "_schedule_task") as mock_schedule:
187 ul._run_odoo_shell(request, run_info, script)
188 mock_schedule.assert_called_once_with(
189 "scriptupgrade",
190 run_info_id=run_info.id,
191 script=script,
192 state_on_success=mock_db.state,
193 )
195 def test_run_odoo_shell_with_original_db(self):
196 """Test that _run_odoo_shell schedules task on original database for pre type."""
197 ul = self.env["saas.upgrade.line"].create(
198 {
199 "name": "[Test] UL odoo_shell pre",
200 "upgrade_type_ids": [Command.link(self.upgrade_type.id)],
201 "type": "2_pre",
202 "state": "approved",
203 "adhoc_product_id": self.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id,
204 }
205 )
206 request = self.test_request
207 run_info = self.env["saas.upgrade.line.request.run"].create(
208 {
209 "upgrade_line_id": ul.id,
210 "request_id": request.id,
211 }
212 )
214 mock_db = MagicMock()
215 script = "result = 'odoo_shell executed'"
217 # Mock original_database_id and _schedule_task
218 with patch.object(type(request), "original_database_id", PropertyMock(return_value=mock_db)):
219 with patch.object(mock_db, "_schedule_task") as mock_schedule:
220 ul._run_odoo_shell(request, run_info, script)
221 mock_schedule.assert_called_once_with(
222 "scriptupgrade",
223 run_info_id=run_info.id,
224 script=script,
225 state_on_success=mock_db.state,
226 )
228 def test_run_odoo_shell_no_db_raises_error(self):
229 """Test that _run_odoo_shell raises UserError when no database is available."""
230 ul = self.env["saas.upgrade.line"].create(
231 {
232 "name": "[Test] UL odoo_shell no db",
233 "upgrade_type_ids": [Command.link(self.upgrade_type.id)],
234 "type": "4_post",
235 "state": "approved",
236 "adhoc_product_id": self.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id,
237 }
238 )
239 request = self.test_request
240 run_info = self.env["saas.upgrade.line.request.run"].create(
241 {
242 "upgrade_line_id": ul.id,
243 "request_id": request.id,
244 }
245 )
247 script = "result = 'odoo_shell executed'"
248 # Mock to have no upgraded_db
249 with patch.object(type(request), "upgraded_database_id", PropertyMock(return_value=None)):
250 with self.assertRaises(Exception) as ctx:
251 ul._run_odoo_shell(request, run_info, script)
252 self.assertIn("There is no database to run the job!", str(ctx.exception))
254 def test_action_copy_script_also_copies_dev_execution_mode(self):
255 """action_copy_script should copy both dev_script and dev_execution_mode from the actual script."""
256 ul = self.ul_odoo_shell
257 # Confirm the actual script has odoo_shell mode
258 self.assertEqual(ul.execution_mode, "odoo_shell")
260 ul.dev_script = False
261 ul.dev_execution_mode = "odooly" # reset to default before the copy
263 ul.action_copy_script()
265 self.assertEqual(ul.dev_script, ul.script, "dev_script must match the actual script after copy")
266 self.assertEqual(ul.dev_execution_mode, "odoo_shell", "dev_execution_mode must be copied from execution_mode")
268 def test_action_approve_script_uses_dev_execution_mode(self):
269 """When approving a dev script, the created script should keep `dev_execution_mode`."""
270 # Reuse an existing upgrade line that already has an actual script
271 ul = self.upgrade_line_02
273 # Set a dev script and change dev_execution_mode to a non-default value
274 ul.dev_script = "x = 42"
275 ul.dev_execution_mode = "odoo_shell"
277 # Approve the script (mock _validate_ops to bypass request requirement)
278 with patch.object(type(ul), "_validate_ops"):
279 ul.with_context(skip_test=True).action_approve_script()
281 self.assertFalse(ul.dev_script)
282 self.assertEqual(ul.script, "x = 42")
283 # The actual_script_id should have been updated to the new script
284 self.assertTrue(ul.actual_script_id)
285 self.assertEqual(ul.actual_script_id.execution_mode, "odoo_shell")