Coverage for ingadhoc-odoo-saas-adhoc / saas_provider_upgrade / tests / test_upgrade_request.py: 99%

231 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-03-09 18:05 +0000

1from unittest.mock import patch 

2 

3from odoo.exceptions import ValidationError 

4 

5from .test_provider_upgrade_base import TestProviderUpgradeCommon 

6 

7 

8class TestProviderUpgradeRequest(TestProviderUpgradeCommon): 

9 @classmethod 

10 def setUpClass(self): 

11 """Prepare shared upgrade ticket, requests and upgrade line linkage.""" 

12 super().setUpClass() 

13 request_model = self.env["helpdesk.ticket.upgrade.request"] 

14 self.test_request = request_model._create_request(self.upgrade_ticket.id, "test") 

15 self.prod_request = request_model._create_request(self.upgrade_ticket.id, "production") 

16 

17 # Ensure the common test upgrade line is linked to the 

18 # actual upgrade type used by the created requests. 

19 # 

20 # After introducing multi-step upgrade paths, the request's 

21 # ``upgrade_type_id`` may differ from ``self.upgrade_type`` 

22 # defined in the base test class. Linking the line here makes 

23 # the script-selection tests exercise the real upgrade type 

24 # used in these requests. 

25 upgrade_type = self.test_request.upgrade_type_id 

26 if upgrade_type: 26 ↛ exitline 26 didn't return from function 'setUpClass' because the condition on line 26 was always true

27 self.upgrade_line_02.write( 

28 { 

29 "upgrade_type_ids": [(4, upgrade_type.id)], 

30 } 

31 ) 

32 

33 def test_request_states(self): 

34 """Basic state/status transitions on an upgrade request.""" 

35 request = self.test_request 

36 # After creation status must be draft 

37 self.assertEqual(request.state, "draft") 

38 # After _running status must be 'running' 

39 request._running() 

40 self.assertEqual(request.status, "running") 

41 # Next state must be validated and status 'on_hold' 

42 request._next_state() 

43 self.assertEqual(request.state, "validated") 

44 self.assertEqual(request.status, "on_hold") 

45 # After _error status must be 'error' 

46 request._error() 

47 self.assertEqual(request.status, "error") 

48 

49 def test_script_ids_compute(self): 

50 """Validate script selection for different UL types and run_type/freeze.""" 

51 test_request = self.test_request 

52 prod_request = self.prod_request 

53 requests = test_request | prod_request 

54 # Request in 'upgraded' state and UL 'post' type 

55 requests.write({"state": "upgraded"}) 

56 prod_request.write({"active_production_freeze": False, "aim": "production"}) 

57 ul = self.upgrade_line_02 

58 ul.write({"type": "4_post", "run_type": False}) 

59 # After compute, in both cases, the UL must be in 'script_ids' 

60 requests._compute_scripts() 

61 self.assertTrue(ul in test_request.script_ids) 

62 self.assertTrue(ul in prod_request.script_ids) 

63 # Set UL 'type' to 'pre' 

64 ul.write({"type": "2_pre"}) 

65 # After compute, in both cases, the UL must NOT be in 'script_ids' 

66 requests._compute_scripts() 

67 self.assertTrue(ul not in test_request.script_ids) 

68 self.assertTrue(ul not in prod_request.script_ids) 

69 # Set UL 'type' to 'post' and 'run_type' to 'production' 

70 ul.write({"type": "4_post", "run_type": "production"}) 

71 # After compute, in the test request, the UL must NOT be in 'script_ids' 

72 # on the other hand, in the production request, the UL must be in 'script_ids' 

73 requests._compute_scripts() 

74 self.assertTrue(ul not in test_request.script_ids) 

75 self.assertTrue(ul in prod_request.script_ids) 

76 # Approve two new script versions to the UL 

77 ul.write({"type": "4_post"}) 

78 self._approve_script(ul) 

79 # Set 'active_production_freeze' request to True 

80 prod_request.write({"active_production_freeze": True}) 

81 ul.write({"run_type": False}) 

82 

83 Request = self.env["helpdesk.ticket.upgrade.request"].with_context( 

84 default_state="draft", from_create_request=True 

85 ) 

86 base_test_request = Request.create( 

87 { 

88 "ticket_id": self.upgrade_ticket.id, 

89 "upgrade_type_id": self.upgrade_type_16_18.id, 

90 "aim": "test", 

91 "state": "done", 

92 } 

93 ) 

94 last_test_request = Request.create( 

95 { 

96 "ticket_id": self.upgrade_ticket.id, 

97 "upgrade_type_id": prod_request.upgrade_type_id.id, 

98 "aim": "test", 

99 "state": "done", 

100 "prev_request_id": base_test_request.id, 

101 } 

102 ) 

103 

104 # After compute, as no test run has registered a version for 

105 # this UL in the base test request, the UL must NOT be in 

106 # 'script_ids'. It should be considered deactivated for this 

107 # production request. 

108 prod_request._compute_scripts() 

109 self.assertTrue(ul not in prod_request.script_ids) 

110 

111 # Now simulate a real test run that registers the version 

112 # used for this UL in the last test request 

113 Run = self.env["saas.upgrade.line.request.run"] 

114 script_v2 = ul.upgrade_line_script_version_ids.filtered(lambda s: s.version == 2) 

115 self.assertTrue(script_v2, "Expected to have at least a script version 2 for the UL") 

116 run = Run.create( 

117 { 

118 "upgrade_line_id": ul.id, 

119 "request_id": last_test_request.id, 

120 "state": "running", 

121 "script_id": script_v2.id, 

122 } 

123 ) 

124 run.finish() 

125 

126 # After registering the version in the base test request, 

127 # recomputing scripts must bring the UL back into the 

128 # production request scripts under freeze. 

129 prod_request._compute_scripts() 

130 # Now, the UL must be in the production request 

131 self.assertTrue(ul in prod_request.script_ids) 

132 # Also, if we deactivate 'active_production_freeze' the UL must be in the production request 

133 requests.write({"active_production_freeze": False}) 

134 requests._compute_scripts() 

135 # Now, the UL must be in the production request 

136 self.assertTrue(ul in prod_request.script_ids) 

137 

138 def test_get_script_version(self): 

139 """Ensure _get_script_version honors freeze map and dev mode.""" 

140 prod_request = self.prod_request 

141 test_request = self.test_request 

142 # Set 'active_production_freeze' to True, approve some script version on the UL 

143 # and add the UL and a version (2) to use to the '4_post' inner map 

144 prod_request.write({"active_production_freeze": True, "aim": "production"}) 

145 ul = self.upgrade_line_02 

146 ul.write({"type": "4_post"}) 

147 self._approve_script(ul) 

148 

149 Request = self.env["helpdesk.ticket.upgrade.request"].with_context( 

150 default_state="draft", from_create_request=True 

151 ) 

152 base_test_request = Request.create( 

153 { 

154 "ticket_id": self.upgrade_ticket.id, 

155 "upgrade_type_id": self.upgrade_type_16_18.id, 

156 "aim": "test", 

157 "state": "done", 

158 } 

159 ) 

160 last_test_request = Request.create( 

161 { 

162 "ticket_id": self.upgrade_ticket.id, 

163 # Again, match the production request's upgrade 

164 # type for the hop whose versions will be reused. 

165 "upgrade_type_id": prod_request.upgrade_type_id.id, 

166 "aim": "test", 

167 "state": "done", 

168 "prev_request_id": base_test_request.id, 

169 } 

170 ) 

171 

172 # Register version 2 for this UL in the base test request 

173 # using the same helper used by real runs. 

174 Run = self.env["saas.upgrade.line.request.run"] 

175 script_v2 = ul.upgrade_line_script_version_ids.filtered(lambda s: s.version == 2) 

176 self.assertTrue(script_v2, "Expected to have at least a script version 2 for the UL") 

177 ul.write({"run_type": False}) 

178 run = Run.create( 

179 { 

180 "upgrade_line_id": ul.id, 

181 "request_id": last_test_request.id, 

182 "state": "running", 

183 "script_id": script_v2.id, 

184 } 

185 ) 

186 run.finish() 

187 

188 script_version = ul.with_context(ticket_request_id=prod_request.id)._get_script_version() 

189 # The script version and the code to execute must be: 2, "x = 1" 

190 self.assertEqual(script_version.version, 2) 

191 self.assertEqual(script_version.script, "x = 1") 

192 # If we set 'active_production_freeze' to False, the script version and the code to execute must 

193 # be the last one: 3, "x = 2" 

194 prod_request.write({"active_production_freeze": False}) 

195 script_version = ul.with_context(ticket_request_id=prod_request.id)._get_script_version() 

196 self.assertEqual(script_version.version, 3) 

197 self.assertEqual(script_version.script, "x = 2") 

198 # Also, the version and the code to execute in the test request be the last one: 3, "x = 2" 

199 script_version = ul.with_context(ticket_request_id=test_request.id)._get_script_version() 

200 self.assertEqual(script_version.version, 3) 

201 self.assertEqual(script_version.script, "x = 2") 

202 # If we set a value to 'dev_script' and use 'test_dev' in the context on the test request 

203 # the returned record must be a new (unsaved) record with the dev_script content 

204 ul.dev_script = "test = 1" 

205 script_version = ul.with_context(test_dev=True, ticket_request_id=test_request.id)._get_script_version() 

206 self.assertFalse(script_version.id) # New record not saved to DB 

207 self.assertEqual(script_version.script, "test = 1") 

208 

209 def test_compute_scripts_with_no_previous_test_requests(self): 

210 """Ensure that when there are no previous test requests and production freeze is active, 

211 `_compute_scripts` does not raise but excludes scripts that weren't tested. 

212 """ 

213 prod_request = self.prod_request 

214 ul = self.upgrade_line_02 

215 # Ensure UL is approved and ready to be considered 

216 ul.write({"type": "4_post", "run_type": False, "state": "approved"}) 

217 # Put request in upgraded state and production aim with active freeze 

218 prod_request.write({"state": "upgraded", "active_production_freeze": True, "aim": "production"}) 

219 # This should not raise even without previous test requests 

220 prod_request._compute_scripts() 

221 # UL should NOT be in script_ids because there are no frozen versions for it 

222 # (no previous test requests means no versions were recorded) 

223 self.assertFalse(ul in prod_request.script_ids) 

224 # But if we disable production freeze, the UL should be included 

225 prod_request.write({"active_production_freeze": False}) 

226 prod_request._compute_scripts() 

227 self.assertTrue(ul in prod_request.script_ids) 

228 

229 @classmethod 

230 def _approve_script(self, ul): 

231 """Create two approved script versions for a given UL.""" 

232 # Approve two scripts for testing (mock _validate_ops to bypass request requirement) 

233 ul.dev_script = "x = 1" 

234 with patch.object(type(ul), "_validate_ops"): 

235 ul.with_context(skip_test=True).action_approve_script() 

236 ul.dev_script = "x = 2" 

237 with patch.object(type(ul), "_validate_ops"): 

238 ul.with_context(skip_test=True).action_approve_script() 

239 

240 # Test re-enabled to ensure the action returns the full chain of requests 

241 def test_action_open_chained_requests_includes_whole_chain(self): 

242 """The action to open chained requests must include all linked requests.""" 

243 Request = self.env["helpdesk.ticket.upgrade.request"].with_context( 

244 default_state="draft", from_create_request=True 

245 ) 

246 req1 = Request.create( 

247 { 

248 "ticket_id": self.upgrade_ticket.id, 

249 "upgrade_type_id": self.upgrade_type.id, 

250 "original_database_id": self.upgrade_production_database.id, 

251 } 

252 ) 

253 req2 = Request.create( 

254 { 

255 "ticket_id": self.upgrade_ticket.id, 

256 "upgrade_type_id": self.upgrade_type.id, 

257 "prev_request_id": req1.id, 

258 } 

259 ) 

260 req3 = Request.create( 

261 { 

262 "ticket_id": self.upgrade_ticket.id, 

263 "upgrade_type_id": self.upgrade_type.id, 

264 "prev_request_id": req2.id, 

265 } 

266 ) 

267 

268 # Invalidate cache to ensure computed fields are refreshed 

269 (req1 | req2 | req3).invalidate_recordset() 

270 

271 # Call the action from the middle of the chain; it should still traverse both ends 

272 action = req2.action_open_chained_requests() 

273 

274 self.assertIsInstance(action, dict) 

275 self.assertIn("domain", action) 

276 

277 domain = action["domain"] 

278 self.assertEqual(len(domain), 1) 

279 self.assertEqual(domain[0][0], "id") 

280 self.assertEqual(domain[0][1], "in") 

281 self.assertCountEqual(domain[0][2], [req1.id, req2.id, req3.id]) 

282 

283 # Test added to validate the new version constraint on original vs target database 

284 def test_check_databases_version_raises_when_original_not_lower(self): 

285 """_check_databases_version must forbid using an original DB in the same target group. 

286 

287 When original and target share the same version-group (same sequence), 

288 the constraint must raise a ValidationError. 

289 """ 

290 UpgradeType = self.env["saas.upgrade.type"] 

291 SaasDatabase = self.env["saas.database"] 

292 

293 # Use an existing version group so the database version and group are consistent 

294 test_group = self.env.ref("saas_provider.odoo_version_group_16") 

295 

296 # Reuse database-type settings from the base upgrade type 

297 upgrade_type = UpgradeType.create( 

298 { 

299 "name": "Test downgrade or same version", 

300 "target_odoo_version_group_id": test_group.id, 

301 "from_major_version_ids": [(6, 0, [test_group.id])], 

302 "original_database_type_id": self.upgrade_type.original_database_type_id.id, 

303 "intermediate_database_type_id": self.upgrade_type.intermediate_database_type_id.id, 

304 "upgraded_database_type_id": self.upgrade_type.upgraded_database_type_id.id, 

305 "production_backup_database_type_id": self.upgrade_type.production_backup_database_type_id.id, 

306 } 

307 ) 

308 

309 # Use a fresh analytic account to avoid max-databases-per-account constraints 

310 test_analytic_account = self.upgrade_analytic_account.copy({"name": "upgrade-analytic-for-version-constraint"}) 

311 database = SaasDatabase.create( 

312 { 

313 "analytic_account_id": test_analytic_account.id, 

314 "database_type_id": self.env.ref("saas_provider.saas_database_type_production").id, 

315 "odoo_version_id": self.env.ref("saas_provider.odoo_version_16").id, 

316 "odoo_version_group_id": test_group.id, 

317 } 

318 ) 

319 with self.assertRaises(ValidationError): 

320 self.env["helpdesk.ticket.upgrade.request"].with_context( 

321 default_state="draft", from_create_request=True 

322 ).create( 

323 { 

324 "ticket_id": self.upgrade_ticket.id, 

325 "upgrade_type_id": upgrade_type.id, 

326 "original_database_id": database.id, 

327 } 

328 ) 

329 

330 # Test added to validate multi-step upgrade path calculation between version groups 

331 def test_calculate_upgrade_steps_multiple_hops(self): 

332 """_calculate_upgrade_steps must return a chain of upgrade types. 

333 

334 We generate three artificial version groups and two upgrade types 

335 (A -> B and B -> C). Asking for steps from A to C must return 

336 both types in the correct order. 

337 """ 

338 VersionGroup = self.env["saas.odoo.version.group"] 

339 UpgradeType = self.env["saas.upgrade.type"] 

340 # Reuse an existing major version to satisfy model constraints 

341 major_version = self.env.ref("saas_provider.saas_odoo_major_version_16") 

342 vg_a = VersionGroup.create({"name": "A", "sequence": 3, "major_version_id": major_version.id}) 

343 vg_b = VersionGroup.create({"name": "B", "sequence": 2, "major_version_id": major_version.id}) 

344 vg_c = VersionGroup.create({"name": "C", "sequence": 1, "major_version_id": major_version.id}) 

345 

346 # Reuse database-type settings from the base upgrade type 

347 common_type_vals = { 

348 "original_database_type_id": self.upgrade_type.original_database_type_id.id, 

349 "intermediate_database_type_id": self.upgrade_type.intermediate_database_type_id.id, 

350 "upgraded_database_type_id": self.upgrade_type.upgraded_database_type_id.id, 

351 "production_backup_database_type_id": self.upgrade_type.production_backup_database_type_id.id, 

352 } 

353 ut_ab = UpgradeType.create( 

354 { 

355 "name": "A->B", 

356 "target_odoo_version_group_id": vg_b.id, 

357 "from_major_version_ids": [(6, 0, [vg_a.id])], 

358 **common_type_vals, 

359 } 

360 ) 

361 ut_bc = UpgradeType.create( 

362 { 

363 "name": "B->C", 

364 "target_odoo_version_group_id": vg_c.id, 

365 "from_major_version_ids": [(6, 0, [vg_b.id])], 

366 **common_type_vals, 

367 } 

368 ) 

369 steps = self.env["saas.upgrade.type"].calculate_upgrade_steps(vg_a, vg_c) 

370 # Order matters: the method must return the exact union in sequence 

371 self.assertEqual(steps, ut_ab | ut_bc) 

372 

373 # Tests for _process_cancel_upgrade_request and _cancel method on runs 

374 # Validates commit b9100226: cancel running upgrade line request runs when request is cancelled 

375 def test_process_cancel_upgrade_request_cancels_running_runs(self): 

376 """_process_cancel_upgrade_request must cancel runs in running/pending/waiting states. 

377 

378 When an upgrade request is cancelled, all associated runs that are in 

379 'running', 'pending', or 'waiting' state should be marked as error. 

380 """ 

381 Run = self.env["saas.upgrade.line.request.run"] 

382 UpgradeLine = self.env["saas.upgrade.line"] 

383 request = self.test_request 

384 

385 # Create multiple upgrade lines to avoid unique constraint violation 

386 upgrade_lines = UpgradeLine.browse() 

387 for i, state in enumerate(["running", "pending", "waiting", "done", "error"]): 

388 upgrade_lines |= UpgradeLine.create( 

389 { 

390 "name": f"[Test Cancel] UL for state {state}", 

391 "upgrade_type_ids": [(4, self.upgrade_type.id)], 

392 "type": "0_on_create", 

393 "state": "approved", 

394 "run_type": "test", 

395 "description": f"Test UL {i}", 

396 "adhoc_product_id": self.env.ref( 

397 "saas_provider_adhoc.adhoc_product_actualizacion_actualizacion" 

398 ).id, 

399 } 

400 ) 

401 

402 # Create runs in different states using different upgrade lines 

403 run_running = Run.create( 

404 { 

405 "upgrade_line_id": upgrade_lines[0].id, 

406 "request_id": request.id, 

407 "state": "running", 

408 } 

409 ) 

410 run_pending = Run.create( 

411 { 

412 "upgrade_line_id": upgrade_lines[1].id, 

413 "request_id": request.id, 

414 "state": "pending", 

415 } 

416 ) 

417 run_waiting = Run.create( 

418 { 

419 "upgrade_line_id": upgrade_lines[2].id, 

420 "request_id": request.id, 

421 "state": "waiting", 

422 } 

423 ) 

424 run_done = Run.create( 

425 { 

426 "upgrade_line_id": upgrade_lines[3].id, 

427 "request_id": request.id, 

428 "state": "done", 

429 } 

430 ) 

431 run_error = Run.create( 

432 { 

433 "upgrade_line_id": upgrade_lines[4].id, 

434 "request_id": request.id, 

435 "state": "error", 

436 } 

437 ) 

438 

439 # Cancel the upgrade request 

440 request._process_cancel_upgrade_request() 

441 

442 # Verify that running/pending/waiting runs are now in canceled state 

443 self.assertEqual(run_running.state, "canceled", "Running run should be cancelled") 

444 self.assertEqual(run_pending.state, "canceled", "Pending run should be cancelled") 

445 self.assertEqual(run_waiting.state, "canceled", "Waiting run should be cancelled") 

446 

447 # Verify that already finished runs are not affected 

448 self.assertEqual(run_done.state, "done", "Done run should remain unchanged") 

449 self.assertEqual(run_error.state, "error", "Error run should remain unchanged") 

450 

451 def test_process_cancel_upgrade_request_no_runs_to_cancel(self): 

452 """_process_cancel_upgrade_request should handle requests with no active runs gracefully.""" 

453 Run = self.env["saas.upgrade.line.request.run"] 

454 UpgradeLine = self.env["saas.upgrade.line"] 

455 request = self.test_request 

456 

457 # Create a dedicated upgrade line for this test 

458 upgrade_line = UpgradeLine.create( 

459 { 

460 "name": "[Test Cancel] UL for no runs test", 

461 "upgrade_type_ids": [(4, self.upgrade_type.id)], 

462 "type": "0_on_create", 

463 "state": "approved", 

464 "run_type": "test", 

465 "description": "Test UL", 

466 "adhoc_product_id": self.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id, 

467 } 

468 ) 

469 

470 # Create only finished runs 

471 run_done = Run.create( 

472 { 

473 "upgrade_line_id": upgrade_line.id, 

474 "request_id": request.id, 

475 "state": "done", 

476 } 

477 ) 

478 

479 # Cancel the upgrade request - should not raise 

480 request._process_cancel_upgrade_request() 

481 

482 # Verify the done run is unchanged 

483 self.assertEqual(run_done.state, "done", "Done run should remain unchanged") 

484 

485 def test_run_cancel_sets_error_state(self): 

486 """_cancel method on runs should set state to error and set end_date.""" 

487 Run = self.env["saas.upgrade.line.request.run"] 

488 UpgradeLine = self.env["saas.upgrade.line"] 

489 request = self.test_request 

490 

491 # Create a dedicated upgrade line for this test 

492 upgrade_line = UpgradeLine.create( 

493 { 

494 "name": "[Test Cancel] UL for cancel state test", 

495 "upgrade_type_ids": [(4, self.upgrade_type.id)], 

496 "type": "0_on_create", 

497 "state": "approved", 

498 "run_type": "test", 

499 "description": "Test UL", 

500 "adhoc_product_id": self.env.ref("saas_provider_adhoc.adhoc_product_actualizacion_actualizacion").id, 

501 } 

502 ) 

503 

504 run = Run.create( 

505 { 

506 "upgrade_line_id": upgrade_line.id, 

507 "request_id": request.id, 

508 "state": "running", 

509 } 

510 ) 

511 

512 # Call _cancel directly 

513 run._cancel() 

514 

515 # Verify state is canceled and end_date is set 

516 self.assertEqual(run.state, "canceled", "Run should be in canceled state after cancel") 

517 self.assertTrue(run.end_date, "End date should be set after cancel") 

518 

519 def test_run_cancel_multiple_runs(self): 

520 """_cancel method should work on multiple runs at once.""" 

521 Run = self.env["saas.upgrade.line.request.run"] 

522 UpgradeLine = self.env["saas.upgrade.line"] 

523 request = self.test_request 

524 

525 # Create multiple upgrade lines 

526 runs = Run.browse() 

527 for i, state in enumerate(["running", "pending", "waiting"]): 

528 upgrade_line = UpgradeLine.create( 

529 { 

530 "name": f"[Test Cancel] UL multiple {i}", 

531 "upgrade_type_ids": [(4, self.upgrade_type.id)], 

532 "type": "0_on_create", 

533 "state": "approved", 

534 "run_type": "test", 

535 "description": f"Test UL {i}", 

536 "adhoc_product_id": self.env.ref( 

537 "saas_provider_adhoc.adhoc_product_actualizacion_actualizacion" 

538 ).id, 

539 } 

540 ) 

541 runs |= Run.create( 

542 { 

543 "upgrade_line_id": upgrade_line.id, 

544 "request_id": request.id, 

545 "state": state, 

546 } 

547 ) 

548 

549 # Call _cancel on all runs 

550 runs._cancel() 

551 

552 original_states = {run: run.state for run in runs} 

553 runs._cancel() 

554 for run in runs: 

555 self.assertEqual( 

556 run.state, "canceled", f"Run originally in state {original_states[run]} should be cancelled" 

557 ) 

558 

559 def test_stop_at_automatic_pause(self): 

560 """Test that stop_at pauses automatic execution and notifies technician.""" 

561 # Create a technician and assign to ticket 

562 technician = self.env["res.users"].create( 

563 { 

564 "name": "Test Technician", 

565 "login": "test.technician", 

566 "email": "test.technician@example.com", 

567 "group_ids": [(6, 0, [self.env.ref("base.group_user").id])], 

568 } 

569 ) 

570 self.upgrade_ticket.assigned_technician_id = technician 

571 

572 # Create automatic request with stop_at='validated' 

573 # _create_request returns the last request in the chain 

574 last_request = self.env["helpdesk.ticket.upgrade.request"]._create_request( 

575 self.upgrade_ticket.id, 

576 aim="test", 

577 automatic=True, 

578 stop_at="validated", 

579 ) 

580 

581 # Get the first request in the chain (which has stop_at and stop_at_user_id set) 

582 request = last_request 

583 while request.prev_request_id: 

584 request = request.prev_request_id 

585 

586 # Verify initial state 

587 self.assertTrue(request.automatic, "Request should be automatic") 

588 self.assertEqual(request.stop_at, "validated", "stop_at should be set to 'validated'") 

589 self.assertEqual(request.stop_at_user_id, self.env.user, "stop_at_user_id should be set to current user") 

590 self.assertEqual(request.state, "draft", "Initial state should be 'draft'") 

591 

592 # Advance to 'validated' - this should trigger the pause 

593 request._running() 

594 messages_before = len(request.message_ids) 

595 request._next_state() # draft -> validated (STOP HERE) 

596 

597 # Verify the pause was triggered 

598 self.assertEqual(request.state, "validated", "Request should have advanced to 'validated'") 

599 self.assertFalse(request.automatic, "automatic should be set to False to pause execution") 

600 self.assertFalse(request.stop_at, "stop_at should be cleared after pausing") 

601 self.assertEqual(request.status, "on_hold", "status should be 'on_hold'") 

602 

603 # Verify notification was posted 

604 messages_after = len(request.message_ids) 

605 self.assertGreater(messages_after, messages_before, "A message should have been posted") 

606 

607 def test_stop_at_without_automatic(self): 

608 """Test that stop_at does not trigger pause for non-automatic requests.""" 

609 # Create a manual request (automatic=False) with stop_at 

610 # This shouldn't normally happen due to validation in _create_request, but test the logic 

611 request = ( 

612 self.env["helpdesk.ticket.upgrade.request"] 

613 .with_context(from_create_request=True) 

614 .create( 

615 { 

616 "ticket_id": self.upgrade_ticket.id, 

617 "upgrade_type_id": self.upgrade_type.id, 

618 "aim": "test", 

619 "automatic": False, 

620 "stop_at": "upgraded", # Set manually for test 

621 } 

622 ) 

623 ) 

624 

625 # Advance through states to 'upgraded' 

626 request.write({"state": "upgrading"}) 

627 request._running() 

628 messages_before = len(request.message_ids) 

629 request._next_state() # upgrading -> upgraded 

630 

631 # Verify NO pause was triggered (automatic is False) 

632 self.assertEqual(request.state, "upgraded") 

633 self.assertFalse(request.automatic, "automatic should still be False") 

634 self.assertEqual(request.stop_at, "upgraded", "stop_at should NOT be cleared (no pause)") 

635 

636 # Verify no notification was posted 

637 messages_after = len(request.message_ids) 

638 self.assertEqual(messages_after, messages_before, "No message should have been posted") 

639 

640 def test_stop_at_without_stop_at_set(self): 

641 """Test that _next_state works normally without stop_at set.""" 

642 # Create automatic request WITHOUT stop_at 

643 request = self.env["helpdesk.ticket.upgrade.request"]._create_request( 

644 self.upgrade_ticket.id, 

645 aim="test", 

646 automatic=True, 

647 stop_at=False, 

648 ) 

649 

650 # Advance through states normally 

651 request._running() 

652 request._next_state() # draft -> validated 

653 

654 # Verify normal behavior (no pause) 

655 self.assertEqual(request.state, "validated") 

656 self.assertTrue(request.automatic, "automatic should still be True") 

657 self.assertFalse(request.stop_at, "stop_at should remain False") 

658 self.assertEqual(request.status, "on_hold", "status should be 'on_hold' as normal")