From 8b2b00309b19592471e914bd1dbd4993061ec20c Mon Sep 17 00:00:00 2001 From: git_admin Date: Mon, 27 Apr 2026 08:18:25 +0000 Subject: [PATCH] Tower: upload cetmix_tower_server 16.0.3.0.1 (via marketplace) --- .../tests/test_jet_waypoint.py | 1995 +++++++++++++++++ 1 file changed, 1995 insertions(+) create mode 100644 addons/cetmix_tower_server/tests/test_jet_waypoint.py diff --git a/addons/cetmix_tower_server/tests/test_jet_waypoint.py b/addons/cetmix_tower_server/tests/test_jet_waypoint.py new file mode 100644 index 0000000..1f3d022 --- /dev/null +++ b/addons/cetmix_tower_server/tests/test_jet_waypoint.py @@ -0,0 +1,1995 @@ +# Copyright (C) 2024 Cetmix OƜ +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). + +from odoo.exceptions import ValidationError +from odoo.tools import mute_logger + +from .common_jets import TestTowerJetsCommon + + +class TestTowerJetWaypoint(TestTowerJetsCommon): + """ + Test the Jet Waypoint model functionality + """ + + @classmethod + def setUpClass(cls): + super().setUpClass() + + # Create variables for testing + cls.variable_test_1 = cls.Variable.create( + { + "name": "Test Variable 1", + "reference": "test_var_1", + } + ) + cls.variable_test_2 = cls.Variable.create( + { + "name": "Test Variable 2", + "reference": "test_var_2", + } + ) + cls.variable_test_3 = cls.Variable.create( + { + "name": "Test Variable 3", + "reference": "test_var_3", + } + ) + # waypoint_template and waypoint are now inherited from TestTowerJetsCommon + + # Create commands for flight plans + cls.command_success = cls.Command.create( + { + "name": "Command -> Success", + "action": "python_code", + "code": "# Just return default values", + } + ) + cls.command_error = cls.Command.create( + { + "name": "Command -> Error", + "action": "python_code", + "code": "result = {'exit_code': -100, 'message': 'Error'}", + } + ) + cls.command_waypoint_check = cls.Command.create( + { + "name": "Command -> Waypoint Check", + "action": "python_code", + "code": ( + "result = {'exit_code': waypoint.id if waypoint else -1, " + "'message': 'waypoint check'}" + ), + } + ) + + # Create flight plans + cls.plan_success = cls.Plan.create( + { + "name": "Waypoint Success Plan", + } + ) + cls.plan_line.create( + { + "sequence": 10, + "plan_id": cls.plan_success.id, + "command_id": cls.command_success.id, + } + ) + + cls.plan_error = cls.Plan.create( + { + "name": "Waypoint Error Plan", + } + ) + cls.plan_line.create( + { + "sequence": 10, + "plan_id": cls.plan_error.id, + "command_id": cls.command_error.id, + } + ) + + cls.plan_waypoint_check = cls.Plan.create( + { + "name": "Waypoint Check Plan", + } + ) + cls.plan_line.create( + { + "sequence": 10, + "plan_id": cls.plan_waypoint_check.id, + "command_id": cls.command_waypoint_check.id, + } + ) + + def test_save_variable_values_empty(self): + """ + Test _save_variable_values when jet has no variable values + """ + # Ensure jet has no variable values + self.jet_test.variable_value_ids.unlink() + + # Save variable values + result = self.waypoint._save_variable_values() + + # Should return True + self.assertTrue(result, "Should return True when saving values") + + # Waypoint should have empty variable_values (or False, which is equivalent) + variable_values = self.waypoint.variable_values or {} + self.assertEqual( + variable_values, + {}, + "Variable values should be empty dict when jet has no values", + ) + + def test_save_variable_values_with_values(self): + """ + Test _save_variable_values when jet has variable values + """ + # Create variable values for the jet + self.VariableValue.create( + { + "variable_id": self.variable_test_1.id, + "value_char": "value_1", + "jet_id": self.jet_test.id, + } + ) + self.VariableValue.create( + { + "variable_id": self.variable_test_2.id, + "value_char": "value_2", + "jet_id": self.jet_test.id, + } + ) + + # Save variable values + result = self.waypoint._save_variable_values() + + # Should return True + self.assertTrue(result, "Should return True when saving values") + + # Waypoint should have saved variable values + self.assertEqual( + self.waypoint.variable_values, + {"test_var_1": "value_1", "test_var_2": "value_2"}, + "Variable values should be saved correctly", + ) + + def test_save_variable_values_with_empty_string(self): + """ + Test _save_variable_values when variable value is empty string + """ + # Create variable value with empty string + self.VariableValue.create( + { + "variable_id": self.variable_test_1.id, + "value_char": "", + "jet_id": self.jet_test.id, + } + ) + + # Save variable values + self.waypoint._save_variable_values() + + # Waypoint should have saved empty string value + self.assertEqual( + self.waypoint.variable_values, + {"test_var_1": ""}, + "Empty string values should be saved", + ) + + def test_save_variable_values_only_jet_values(self): + """ + Test _save_variable_values only saves jet-specific values, + not template/server/global values + """ + # Create jet-specific variable value + self.VariableValue.create( + { + "variable_id": self.variable_test_1.id, + "value_char": "jet_value", + "jet_id": self.jet_test.id, + } + ) + + # Create template variable value (should not be saved) + self.VariableValue.create( + { + "variable_id": self.variable_test_2.id, + "value_char": "template_value", + "jet_template_id": self.jet_template_test.id, + } + ) + + # Save variable values + self.waypoint._save_variable_values() + + # Waypoint should only have jet-specific value + self.assertEqual( + self.waypoint.variable_values, + {"test_var_1": "jet_value"}, + "Should only save jet-specific values", + ) + self.assertNotIn( + "test_var_2", + self.waypoint.variable_values, + "Should not save template values", + ) + + def test_restore_variable_values_empty(self): + """ + Test _restore_variable_values when waypoint has no saved values + """ + # Create some variable values in jet + self.VariableValue.create( + { + "variable_id": self.variable_test_1.id, + "value_char": "existing_value", + "jet_id": self.jet_test.id, + } + ) + + # Set waypoint variable_values to empty + self.waypoint.variable_values = {} + + # Restore variable values + result = self.waypoint._restore_variable_values() + + # Should return True + self.assertTrue(result, "Should return True when restoring values") + + # Jet should have no variable values + self.assertEqual( + len(self.jet_test.variable_value_ids), + 0, + "All jet variable values should be removed when waypoint is empty", + ) + + def test_restore_variable_values_basic(self): + """ + Test _restore_variable_values restores values correctly + """ + # Set waypoint variable values + self.waypoint.variable_values = { + "test_var_1": "restored_value_1", + "test_var_2": "restored_value_2", + } + + # Restore variable values + result = self.waypoint._restore_variable_values() + + # Should return True + self.assertTrue(result, "Should return True when restoring values") + + # Check values were restored + self.assertEqual( + self.jet_test.get_variable_value("test_var_1", no_fallback=True), + "restored_value_1", + "Variable 1 should be restored", + ) + self.assertEqual( + self.jet_test.get_variable_value("test_var_2", no_fallback=True), + "restored_value_2", + "Variable 2 should be restored", + ) + + def test_restore_variable_values_removes_unsaved(self): + """ + Test _restore_variable_values removes variable values not in waypoint + """ + # Create variable values in jet + self.VariableValue.create( + { + "variable_id": self.variable_test_1.id, + "value_char": "value_1", + "jet_id": self.jet_test.id, + } + ) + self.VariableValue.create( + { + "variable_id": self.variable_test_2.id, + "value_char": "value_2", + "jet_id": self.jet_test.id, + } + ) + self.VariableValue.create( + { + "variable_id": self.variable_test_3.id, + "value_char": "value_3", + "jet_id": self.jet_test.id, + } + ) + + # Set waypoint to only have variable 1 and 2 + self.waypoint.variable_values = { + "test_var_1": "value_1", + "test_var_2": "value_2", + } + + # Restore variable values + self.waypoint._restore_variable_values() + + # Variable 3 should be removed + self.assertIsNone( + self.jet_test.get_variable_value("test_var_3", no_fallback=True), + "Variable 3 should be removed", + ) + + # Variables 1 and 2 should still exist + self.assertEqual( + self.jet_test.get_variable_value("test_var_1", no_fallback=True), + "value_1", + "Variable 1 should still exist", + ) + self.assertEqual( + self.jet_test.get_variable_value("test_var_2", no_fallback=True), + "value_2", + "Variable 2 should still exist", + ) + + def test_restore_variable_values_updates_existing(self): + """ + Test _restore_variable_values updates existing variable values + """ + # Create variable value in jet + self.VariableValue.create( + { + "variable_id": self.variable_test_1.id, + "value_char": "old_value", + "jet_id": self.jet_test.id, + } + ) + + # Set waypoint with new value + self.waypoint.variable_values = {"test_var_1": "new_value"} + + # Restore variable values + self.waypoint._restore_variable_values() + + # Value should be updated + self.assertEqual( + self.jet_test.get_variable_value("test_var_1", no_fallback=True), + "new_value", + "Variable value should be updated", + ) + + def test_save_and_restore_roundtrip(self): + """ + Test saving and restoring variable values in a roundtrip + """ + # Create initial variable values + self.VariableValue.create( + { + "variable_id": self.variable_test_1.id, + "value_char": "initial_value_1", + "jet_id": self.jet_test.id, + } + ) + self.VariableValue.create( + { + "variable_id": self.variable_test_2.id, + "value_char": "initial_value_2", + "jet_id": self.jet_test.id, + } + ) + + # Save variable values + self.waypoint._save_variable_values() + + # Modify jet values + self.jet_test.set_variable_value("test_var_1", "modified_value_1") + self.jet_test.set_variable_value("test_var_2", "modified_value_2") + self.VariableValue.create( + { + "variable_id": self.variable_test_3.id, + "value_char": "new_value", + "jet_id": self.jet_test.id, + } + ) + + # Restore variable values + self.waypoint._restore_variable_values() + + # Values should be restored to original + self.assertEqual( + self.jet_test.get_variable_value("test_var_1", no_fallback=True), + "initial_value_1", + "Variable 1 should be restored to original value", + ) + self.assertEqual( + self.jet_test.get_variable_value("test_var_2", no_fallback=True), + "initial_value_2", + "Variable 2 should be restored to original value", + ) + # Variable 3 should be removed (not in saved waypoint) + self.assertIsNone( + self.jet_test.get_variable_value("test_var_3", no_fallback=True), + "Variable 3 should be removed", + ) + + def test_write_waypoint_template_draft_allowed(self): + """ + Test that modifying waypoint_template_id is allowed when state is draft + """ + # Create waypoint in draft state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Draft", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "draft", + } + ) + + # Should be able to change template in draft state + waypoint.write({"waypoint_template_id": self.waypoint_template_2.id}) + self.assertEqual( + waypoint.waypoint_template_id.id, + self.waypoint_template_2.id, + "Should be able to change template in draft state", + ) + + def test_write_waypoint_template_not_draft_raises_error(self): + """ + Test that modifying waypoint_template_id raises ValidationError + when state is not draft + """ + # Create waypoint in ready state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Ready", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "ready", + } + ) + + # Should raise ValidationError when trying to change template + with self.assertRaises(ValidationError) as context: + waypoint.write({"waypoint_template_id": self.waypoint_template_2.id}) + + self.assertIn( + "draft state", + str(context.exception), + "Should raise ValidationError about draft state", + ) + + def test_write_waypoint_template_same_value_allowed(self): + """ + Test that setting waypoint_template_id to the same value is allowed + even when not in draft state + """ + # Create waypoint in ready state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Ready", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "ready", + } + ) + original_template_id = waypoint.waypoint_template_id.id + + # Should be able to set to the same template + waypoint.write({"waypoint_template_id": original_template_id}) + self.assertEqual( + waypoint.waypoint_template_id.id, + original_template_id, + "Should be able to set same template value", + ) + + def test_write_other_fields_not_draft_allowed(self): + """ + Test that modifying other fields is allowed when state is not draft + """ + # Create waypoint in ready state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Ready", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "ready", + } + ) + + # Should be able to modify other fields + waypoint.write({"name": "Updated Name"}) + self.assertEqual( + waypoint.name, + "Updated Name", + "Should be able to modify other fields when not in draft", + ) + + def test_prepare_without_flight_plan(self): + """ + Test prepare() when waypoint template has no plan_create_id + """ + # Create waypoint template without plan_create_id + waypoint_template_no_plan = self.JetWaypointTemplate.create( + { + "name": "Test Waypoint Template No Plan", + "jet_template_id": self.jet_template_test.id, + } + ) + + # Create waypoint in draft state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint No Plan", + "jet_id": self.jet_test.id, + "waypoint_template_id": waypoint_template_no_plan.id, + "state": "draft", + } + ) + + # Call prepare + result = waypoint.prepare() + + # Should return True and set state to ready + self.assertTrue(result, "Should return True") + self.assertEqual( + waypoint.state, + "ready", + "State should be set to ready when no flight plan", + ) + + def test_prepare_without_flight_plan_with_is_destination(self): + """ + Test prepare() when waypoint template has no plan_create_id + and is_destination=True + Should automatically call fly_to() when prepare completes + """ + # Create waypoint template without plan_create_id + waypoint_template_no_plan = self.JetWaypointTemplate.create( + { + "name": "Test Waypoint Template No Plan Destination", + "jet_template_id": self.jet_template_test.id, + } + ) + + # Create waypoint in draft state with is_destination=True + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint No Plan Destination", + "jet_id": self.jet_test.id, + "waypoint_template_id": waypoint_template_no_plan.id, + "state": "draft", + } + ) + + # Call prepare + result = waypoint.prepare(is_destination=True) + + # Should return True + self.assertTrue(result, "Should return True") + # State should be set to current (because fly_to() was called) + # Since there's no previous waypoint and no plan_arrive_id, + # fly_to() sets state to arriving and calls _arrive() which sets it to current + self.assertEqual( + waypoint.state, + "current", + "State should be set to current after fly_to() and _arrive()", + ) + # Waypoint should be set as current waypoint + self.assertEqual( + self.jet_test.waypoint_id.id, + waypoint.id, + "Waypoint should be set as current waypoint after fly_to()", + ) + # is_destination should be cleared after arriving + self.assertFalse( + waypoint.is_destination, + "is_destination should be cleared after arriving", + ) + + def test_prepare_with_flight_plan_success(self): + """ + Test prepare() when waypoint template has plan_create_id and plan succeeds + """ + # Set template to use success plan + self.waypoint_template.plan_create_id = self.plan_success.id + + # Create waypoint in draft state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint With Plan", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "draft", + } + ) + + # Call prepare - plan executes synchronously in tests + result = waypoint.prepare() + + # Should return True + self.assertTrue(result, "Should return True") + + # State should be set to ready after successful plan completion + # (plan executes synchronously in tests, preparing -> ready) + self.assertEqual( + waypoint.state, + "ready", + "State should be set to ready after successful plan completion", + ) + # Waypoint should NOT be set as current waypoint after preparing + # (only arriving sets waypoint as current) + self.assertNotEqual( + self.jet_test.waypoint_id.id if self.jet_test.waypoint_id else False, + waypoint.id, + "Waypoint should not be set as current waypoint after preparing", + ) + + def test_waypoint_variable_in_python_command_prepare(self): + """ + Test that 'waypoint' variable is available in Python commands + run for a waypoint plan (plan_create) and its id is used as exit code + """ + # Set template to use waypoint check plan + self.waypoint_template.plan_create_id = self.plan_waypoint_check.id + + # Create waypoint in draft state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint For Variable Check", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "draft", + } + ) + + # Call prepare - plan executes synchronously in tests + waypoint.prepare() + + # Find the plan log created by prepare + plan_log = self.PlanLog.search( + [("waypoint_id", "=", waypoint.id)], + order="create_date desc", + limit=1, + ) + self.assertTrue(plan_log, "Plan log should be created") + + # Plan exit code (plan_status) must equal waypoint id + self.assertEqual( + plan_log.plan_status, + waypoint.id, + "Plan status must equal waypoint id (from waypoint variable)", + ) + + def test_waypoint_variable_in_python_command_arrive(self): + """ + Test that 'waypoint' variable is available in Python commands + run for a waypoint arrive plan and its id is used as exit code + """ + # Create waypoint template with plan_arrive_id + waypoint_template = self.JetWaypointTemplate.create( + { + "name": "Waypoint Template For Arrive Check", + "jet_template_id": self.jet_template_test.id, + "plan_arrive_id": self.plan_waypoint_check.id, + } + ) + + # Create waypoint in arriving state (no previous waypoint) + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint For Arrive Variable Check", + "jet_id": self.jet_test.id, + "waypoint_template_id": waypoint_template.id, + "state": "arriving", + } + ) + + # Call arrive - plan executes synchronously in tests + waypoint._arrive() + + # Find the plan log created by arrive + plan_log = self.PlanLog.search( + [("waypoint_id", "=", waypoint.id)], + order="create_date desc", + limit=1, + ) + self.assertTrue(plan_log, "Plan log should be created") + + # Plan exit code (plan_status) must equal waypoint id + self.assertEqual( + plan_log.plan_status, + waypoint.id, + "Plan status must equal waypoint id (from waypoint variable)", + ) + + def test_prepare_with_flight_plan_error(self): + """ + Test prepare() when waypoint template has plan_create_id and plan fails + """ + # Set template to use error plan + self.waypoint_template.plan_create_id = self.plan_error.id + + # Create waypoint in draft state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint With Plan Error", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "draft", + } + ) + + # Call prepare - plan executes synchronously in tests + with mute_logger( + "odoo.addons.cetmix_tower_server.models.cx_tower_jet_waypoint" + ): + result = waypoint.prepare() + + # Should return True + self.assertTrue(result, "Should return True") + + # State should be set to error after failed plan completion + # (plan executes synchronously in tests) + self.assertEqual( + waypoint.state, + "error", + "State should be set to error after failed plan completion", + ) + # Waypoint should not be set as current waypoint on error + self.assertNotEqual( + self.jet_test.waypoint_id.id, + waypoint.id, + "Waypoint should not be set as current waypoint after failed prepare", + ) + + def test_prepare_not_draft_state(self): + """ + Test prepare() when waypoint is not in draft state + """ + # Create waypoint in ready state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Ready", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "ready", + } + ) + + # Call prepare. This will log and error because waypoint is not in draft state + with mute_logger( + "odoo.addons.cetmix_tower_server.models.cx_tower_jet_waypoint" + ): + with self.assertRaises(ValidationError): + waypoint.prepare() + + def test_plan_finished_preparing_success(self): + """ + Test _plan_finished when waypoint is in preparing state and plan succeeds + """ + # Create waypoint in preparing state (simulating async plan execution) + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Preparing", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "preparing", + } + ) + + # Create plan log with success status + plan_log = self.PlanLog.create( + { + "server_id": self.jet_test.server_id.id, + "plan_id": self.plan_success.id, + "plan_status": 0, # Success + } + ) + + # Call _plan_finished + result = waypoint._plan_finished(plan_log) + + # Should return True + self.assertTrue(result, "Should return True") + # State should be set to ready + # (preparing -> ready, not current) + self.assertEqual( + waypoint.state, + "ready", + "State should be set to ready after successful plan completion", + ) + # Waypoint should NOT be set as current waypoint after preparing + # (only arriving sets waypoint as current) + self.assertNotEqual( + self.jet_test.waypoint_id.id if self.jet_test.waypoint_id else False, + waypoint.id, + "Waypoint should not be set as current waypoint after preparing", + ) + + def test_plan_finished_preparing_success_with_is_destination(self): + """ + Test _plan_finished when waypoint is in preparing state with is_destination=True + Should automatically call fly_to() when preparing finishes + """ + # Create waypoint in preparing state with is_destination=True + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Preparing Destination", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "preparing", + "is_destination": True, + } + ) + + # Create plan log with success status + plan_log = self.PlanLog.create( + { + "server_id": self.jet_test.server_id.id, + "plan_id": self.plan_success.id, + "plan_status": 0, # Success + } + ) + + # Call _plan_finished + result = waypoint._plan_finished(plan_log) + + # Should return True + self.assertTrue(result, "Should return True") + # State should be set to arriving (because fly_to() was called) + # Since there's no previous waypoint and no plan_arrive_id, + # fly_to() sets state to arriving and calls _arrive() which sets it to current + self.assertEqual( + waypoint.state, + "current", + "State should be set to current after fly_to() and _arrive()", + ) + # Waypoint should be set as current waypoint + self.assertEqual( + self.jet_test.waypoint_id.id, + waypoint.id, + "Waypoint should be set as current waypoint after fly_to()", + ) + # is_destination should be cleared after arriving + self.assertFalse( + waypoint.is_destination, + "is_destination should be cleared after arriving", + ) + + def test_plan_finished_arriving_success(self): + """ + Test _plan_finished when waypoint is in arriving state and plan succeeds + """ + # Create waypoint in arriving state (simulating async plan execution) + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Arriving", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "arriving", + } + ) + + # Create plan log with success status + plan_log = self.PlanLog.create( + { + "server_id": self.jet_test.server_id.id, + "plan_id": self.plan_success.id, + "plan_status": 0, # Success + } + ) + + # Call _plan_finished + result = waypoint._plan_finished(plan_log) + + # Should return True + self.assertTrue(result, "Should return True") + # State should be set to current + # (waypoint becomes current after successful arrive) + self.assertEqual( + waypoint.state, + "current", + "State should be set to current after successful plan completion", + ) + # Waypoint should be set as current waypoint + self.assertEqual( + self.jet_test.waypoint_id.id, + waypoint.id, + "Waypoint should be set as current waypoint after successful arrive", + ) + + def test_plan_finished_leaving_success(self): + """ + Test _plan_finished when waypoint is in leaving state and plan succeeds + """ + # Create current waypoint in current state + current_waypoint = self.JetWaypoint.create( + { + "name": "Current Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "current", + } + ) + self.jet_test.waypoint_id = current_waypoint.id + + # Create destination waypoint in arriving state + destination_waypoint = self.JetWaypoint.create( + { + "name": "Destination Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "is_destination": True, + "state": "arriving", + } + ) + + # Set current waypoint to leaving state + # readonly=True only affects UI, can be written programmatically + current_waypoint.write({"state": "leaving"}) + + # Create plan log with success status + plan_log = self.PlanLog.create( + { + "server_id": self.jet_test.server_id.id, + "plan_id": self.plan_success.id, + "plan_status": 0, # Success + } + ) + + # Call _plan_finished on leaving waypoint + result = current_waypoint._plan_finished(plan_log) + + # Should return True + self.assertTrue(result, "Should return True") + # Leaving waypoint state should be set to ready + self.assertEqual( + current_waypoint.state, + "ready", + "Leaving waypoint state should be set to ready", + ) + # Destination waypoint should have _arrive() called + # (state should be current if no plan_arrive_id) + # Since waypoint_template has no plan_arrive_id by default, + # _arrive() sets state to current + self.assertEqual( + destination_waypoint.state, + "current", + "Destination waypoint should have _arrive() called", + ) + # Destination waypoint should be set as current waypoint + self.assertEqual( + self.jet_test.waypoint_id.id, + destination_waypoint.id, + "Destination waypoint should be set as current waypoint" + " after leaving completes", + ) + + def test_plan_finished_deleting_success(self): + """ + Test _plan_finished when waypoint is in deleting state and plan succeeds + """ + # Create waypoint template with plan_delete_id + waypoint_template = self.JetWaypointTemplate.create( + { + "name": "Test Template With Delete Plan", + "jet_template_id": self.jet_template_test.id, + "plan_delete_id": self.plan_success.id, + } + ) + + # Create waypoint and set it as current + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Deleting", + "jet_id": self.jet_test.id, + "waypoint_template_id": waypoint_template.id, + "state": "ready", + } + ) + self.jet_test.waypoint_id = waypoint.id + + # Set waypoint to deleting state + # readonly=True only affects UI, can be written programmatically + waypoint.write({"state": "deleting"}) + + # Create plan log with success status + plan_log = self.PlanLog.create( + { + "server_id": self.jet_test.server_id.id, + "plan_id": self.plan_success.id, + "plan_status": 0, # Success + } + ) + + # Call _plan_finished + result = waypoint._plan_finished(plan_log) + + # Should return True + self.assertTrue(result, "Should return True") + # Waypoint should be unlinked (deleted) + # State is set to "deleted" before unlink + self.assertFalse( + waypoint.exists(), + "Waypoint should be unlinked after successful delete plan", + ) + # Jet waypoint_id should be set to False + self.assertFalse( + self.jet_test.waypoint_id, + "Jet waypoint_id should be set to False after successful delete", + ) + + def test_plan_finished_error(self): + """ + Test _plan_finished when plan fails (plan_status != 0) + """ + # Create waypoint in preparing state (simulating async plan execution) + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Preparing", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "preparing", + } + ) + original_waypoint_id = ( + self.jet_test.waypoint_id.id if self.jet_test.waypoint_id else False + ) + + # Create plan log with error status + plan_log = self.PlanLog.create( + { + "server_id": self.jet_test.server_id.id, + "plan_id": self.plan_error.id, + "plan_status": 1, # Error + } + ) + + # Call _plan_finished + with mute_logger( + "odoo.addons.cetmix_tower_server.models.cx_tower_jet_waypoint" + ): + result = waypoint._plan_finished(plan_log) + + # Should return True + self.assertTrue(result, "Should return True") + # State should be set to error + self.assertEqual( + waypoint.state, + "error", + "State should be set to error after failed plan completion", + ) + # Waypoint should not be set as current waypoint + if original_waypoint_id: + self.assertEqual( + self.jet_test.waypoint_id.id, + original_waypoint_id, + "Current waypoint should not change on error", + ) + else: + self.assertFalse( + self.jet_test.waypoint_id, + "Current waypoint should remain False on error", + ) + + def test_plan_finished_error_arriving(self): + """ + Test _plan_finished when waypoint is in arriving state and plan fails + """ + # Create waypoint in arriving state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Arriving", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "arriving", + } + ) + + # Create plan log with error status + plan_log = self.PlanLog.create( + { + "server_id": self.jet_test.server_id.id, + "plan_id": self.plan_error.id, + "plan_status": 1, # Error + } + ) + + # Call _plan_finished + with mute_logger( + "odoo.addons.cetmix_tower_server.models.cx_tower_jet_waypoint" + ): + result = waypoint._plan_finished(plan_log) + + # Should return True + self.assertTrue(result, "Should return True") + # State should be set to error + self.assertEqual( + waypoint.state, + "error", + "State should be set to error after failed plan completion", + ) + # Waypoint should not be set as current waypoint on error + self.assertNotEqual( + self.jet_test.waypoint_id.id if self.jet_test.waypoint_id else False, + waypoint.id, + "Waypoint should not be set as current waypoint after failed arrive", + ) + + def test_get_custom_variable_values_with_metadata(self): + """ + Test _get_custom_variable_values with metadata + """ + # Set template to use success plan + self.waypoint_template.plan_create_id = self.plan_success.id + + # Create waypoint with metadata + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint With Metadata", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "draft", + "metadata": {"key1": "value1", "key2": "value2", "env": "production"}, + } + ) + + # Call prepare to trigger flight plan + waypoint.prepare() + + # Find the plan log created by prepare + plan_log = self.PlanLog.search( + [ + ("waypoint_id", "=", waypoint.id), + ], + order="create_date desc", + limit=1, + ) + self.assertTrue(plan_log, "Plan log should be created") + + # Check custom variable values in plan log + self.assertEqual( + plan_log.variable_values.get("__waypoint"), + waypoint.reference, + "__waypoint should match waypoint reference", + ) + self.assertEqual( + plan_log.variable_values.get("__waypoint_type"), + self.waypoint_template.reference, + "__waypoint_type should match waypoint template reference", + ) + self.assertEqual( + plan_log.variable_values.get("__waypoint_state"), + "preparing", + "__waypoint_state should be preparing", + ) + # Check metadata keys + self.assertEqual( + plan_log.variable_values.get("__waypoint_key1"), + "value1", + "__waypoint_key1 should match metadata value", + ) + self.assertEqual( + plan_log.variable_values.get("__waypoint_key2"), + "value2", + "__waypoint_key2 should match metadata value", + ) + self.assertEqual( + plan_log.variable_values.get("__waypoint_env"), + "production", + "__waypoint_env should match metadata value", + ) + + def test_get_custom_variable_values_without_metadata(self): + """ + Test _get_custom_variable_values without metadata + """ + # Set template to use success plan + self.waypoint_template.plan_create_id = self.plan_success.id + + # Create waypoint without metadata + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Without Metadata", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "draft", + } + ) + + # Call prepare to trigger flight plan + waypoint.prepare() + + # Find the plan log created by prepare + plan_log = self.PlanLog.search( + [("waypoint_id", "=", waypoint.id)], + order="create_date desc", + limit=1, + ) + self.assertTrue(plan_log, "Plan log should be created") + + # Check basic custom variable values + self.assertEqual( + plan_log.variable_values.get("__waypoint"), + waypoint.reference, + "__waypoint should match waypoint reference", + ) + self.assertEqual( + plan_log.variable_values.get("__waypoint_type"), + self.waypoint_template.reference, + "__waypoint_type should match waypoint template reference", + ) + self.assertEqual( + plan_log.variable_values.get("__waypoint_state"), + "preparing", + "__waypoint_state should be preparing", + ) + # Check that metadata keys are not present + self.assertNotIn( + "__waypoint_key1", + plan_log.variable_values, + "Metadata keys should not be present when metadata is empty", + ) + + def test_leave_from_current_state(self): + """ + Test _leave() when waypoint is in current state + """ + # Create waypoint in current state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Current", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "current", + } + ) + self.jet_test.waypoint_id = waypoint.id + + # Call _leave + result = waypoint._leave() + + # Should return True + self.assertTrue(result, "Should return True") + # State should be set to ready + # (_leave() completes immediately when no plan_leave_id in tests) + self.assertEqual( + waypoint.state, + "ready", + "State should be set to ready after leaving completes", + ) + + def test_fly_to_from_current_waypoint(self): + """ + Test fly_to() when previous waypoint is in current state + """ + # Create current waypoint + current_waypoint = self.JetWaypoint.create( + { + "name": "Current Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "current", + } + ) + self.jet_test.waypoint_id = current_waypoint.id + + # Create destination waypoint + destination_waypoint = self.JetWaypoint.create( + { + "name": "Destination Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "ready", + } + ) + + # Call fly_to on destination waypoint + result = destination_waypoint.fly_to() + + # Should return True + self.assertTrue(result, "Should return True") + # Current waypoint should be in ready state + # (_leave() completes immediately when no plan_leave_id in tests) + self.assertEqual( + current_waypoint.state, + "ready", + "Current waypoint should be in ready state after leaving completes", + ) + # Destination waypoint should be in current state + # (_arrive() completes immediately when no plan_arrive_id in tests) + self.assertEqual( + destination_waypoint.state, + "current", + "Destination waypoint should be in current state after arriving", + ) + # Destination waypoint should be set as current waypoint + self.assertEqual( + self.jet_test.waypoint_id.id, + destination_waypoint.id, + "Destination waypoint should be set as current waypoint", + ) + + def test_fly_to_leave_failure_does_not_keep_destination_arriving(self): + """ + Regression: if source leave plan fails during fly_to(), + destination must not stay in arriving. + """ + # Create template with failing leave plan. + waypoint_template_with_leave_error = self.JetWaypointTemplate.create( + { + "name": "Template Leave Error", + "jet_template_id": self.jet_template_test.id, + "plan_leave_id": self.plan_error.id, + } + ) + + # Create current waypoint that will fail while leaving. + current_waypoint = self.JetWaypoint.create( + { + "name": "Current Waypoint Failing Leave", + "jet_id": self.jet_test.id, + "waypoint_template_id": waypoint_template_with_leave_error.id, + "state": "current", + } + ) + self.jet_test.waypoint_id = current_waypoint.id + + # Create destination waypoint (target of fly_to). + destination_waypoint = self.JetWaypoint.create( + { + "name": "Destination Waypoint Stuck Arriving", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "ready", + } + ) + + # Execute fly_to; leaving fails synchronously in tests. + with mute_logger( + "odoo.addons.cetmix_tower_server.models.cx_tower_jet_waypoint" + ): + result = destination_waypoint.fly_to() + + self.assertFalse(result, "fly_to() should return False when leave fails") + self.assertEqual( + current_waypoint.state, + "error", + "Source waypoint should become error after failed leave plan", + ) + self.assertNotEqual( + destination_waypoint.state, + "arriving", + "Destination waypoint must be reverted from arriving when leave fails", + ) + self.assertFalse( + destination_waypoint.is_destination, + "Destination flag must be cleared when leave fails", + ) + + def test_unlink_current_state_raises_error(self): + """ + Test unlink() when waypoint is in current state raises ValidationError + """ + # Create waypoint in current state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Current", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "current", + } + ) + self.jet_test.waypoint_id = waypoint.id + + # Should raise ValidationError when trying to delete + + with self.assertRaises(ValidationError) as context: + waypoint.unlink() + + self.assertIn( + "current waypoint", + str(context.exception), + "Should raise ValidationError about current waypoint", + ) + + def test_unlink_current_state_with_no_raise_context(self): + """ + Test unlink() when waypoint is in current state + with 'waypoint_no_raise_on_delete' context. + The context prevents exception but waypoint is not deleted. + """ + # Create waypoint in current state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint Current", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "current", + } + ) + self.jet_test.waypoint_id = waypoint.id + waypoint_id = waypoint.id + + # Mute logger error for this test + with mute_logger( + "odoo.addons.cetmix_tower_server.models.cx_tower_jet_waypoint" + ): + # Should not raise error with waypoint_no_raise_on_delete context + waypoint.with_context(waypoint_no_raise_on_delete=True).unlink() + + # Waypoint should still exist (not deleted) + # The context only prevents exception, but doesn't allow deletion + self.assertTrue( + waypoint.exists(), + "Waypoint should still exist - context only prevents exception", + ) + self.assertEqual( + waypoint.id, + waypoint_id, + "Waypoint ID should remain the same", + ) + self.assertEqual( + waypoint.state, + "current", + "Waypoint state should remain current", + ) + + def test_prepare_saves_variable_values(self): + """ + Test that prepare() saves variable values when state changes to ready + """ + # Set some variable values on the jet + self.jet_test.set_variable_value("test_var_1", "value1") + self.jet_test.set_variable_value("test_var_2", "value2") + + # Create waypoint in draft state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "draft", + } + ) + + # Ensure waypoint has no plan_create_id (so it goes directly to ready) + waypoint.waypoint_template_id.plan_create_id = False + + # Call prepare + waypoint.prepare() + + # Variable values should be saved in waypoint + variable_values = waypoint.variable_values or {} + self.assertEqual( + variable_values.get("test_var_1"), + "value1", + "Variable value should be saved when preparing", + ) + self.assertEqual( + variable_values.get("test_var_2"), + "value2", + "Variable value should be saved when preparing", + ) + + def test_prepare_with_plan_saves_variable_values(self): + """ + Test that prepare() saves variable values when plan completes + """ + # Set some variable values on the jet + self.jet_test.set_variable_value("test_var_1", "value1") + self.jet_test.set_variable_value("test_var_2", "value2") + + # Create waypoint template with plan_create_id + waypoint_template = self.JetWaypointTemplate.create( + { + "name": "Test Template", + "jet_template_id": self.jet_template_test.id, + "plan_create_id": self.plan_success.id, + } + ) + + # Create waypoint in draft state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": waypoint_template.id, + "state": "draft", + } + ) + + # Call prepare (plan executes synchronously in tests) + waypoint.prepare() + + # Variable values should be saved in waypoint after plan completes + variable_values = waypoint.variable_values or {} + self.assertEqual( + variable_values.get("test_var_1"), + "value1", + "Variable value should be saved when preparing completes", + ) + self.assertEqual( + variable_values.get("test_var_2"), + "value2", + "Variable value should be saved when preparing completes", + ) + + def test_leave_saves_variable_values(self): + """ + Test that _leave() saves variable values when state changes to ready + """ + # Set some variable values on the jet + self.jet_test.set_variable_value("test_var_1", "value1") + self.jet_test.set_variable_value("test_var_2", "value2") + + # Create waypoint in current state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "current", + } + ) + self.jet_test.waypoint_id = waypoint.id + + # Ensure waypoint has no plan_leave_id (so it goes directly to ready) + waypoint.waypoint_template_id.plan_leave_id = False + + # Call _leave + waypoint._leave() + + # Variable values should be saved in waypoint + variable_values = waypoint.variable_values or {} + self.assertEqual( + variable_values.get("test_var_1"), + "value1", + "Variable value should be saved when leaving", + ) + self.assertEqual( + variable_values.get("test_var_2"), + "value2", + "Variable value should be saved when leaving", + ) + + def test_leave_with_plan_saves_variable_values(self): + """ + Test that _leave() saves variable values when plan completes + """ + # Set some variable values on the jet + self.jet_test.set_variable_value("test_var_1", "value1") + self.jet_test.set_variable_value("test_var_2", "value2") + + # Create waypoint template with plan_leave_id + waypoint_template = self.JetWaypointTemplate.create( + { + "name": "Test Template", + "jet_template_id": self.jet_template_test.id, + "plan_leave_id": self.plan_success.id, + } + ) + + # Create waypoint in current state + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": waypoint_template.id, + "state": "current", + } + ) + self.jet_test.waypoint_id = waypoint.id + + # Call _leave (plan executes synchronously in tests) + waypoint._leave() + + # Variable values should be saved in waypoint after plan completes + variable_values = waypoint.variable_values or {} + self.assertEqual( + variable_values.get("test_var_1"), + "value1", + "Variable value should be saved when leaving completes", + ) + self.assertEqual( + variable_values.get("test_var_2"), + "value2", + "Variable value should be saved when leaving completes", + ) + + def test_fly_to_restores_variable_values(self): + """ + Test that fly_to() restores variable values when state changes to arriving + """ + # Create waypoint with saved variable values + waypoint = self.JetWaypoint.create( + { + "name": "Test Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "ready", + "variable_values": { + "test_var_1": "saved_value1", + "test_var_2": "saved_value2", + }, + } + ) + + # Set different values on the jet + self.jet_test.set_variable_value("test_var_1", "current_value1") + self.jet_test.set_variable_value("test_var_2", "current_value2") + + # Call fly_to (no previous waypoint) + waypoint.fly_to() + + # Variable values should be restored from waypoint + self.assertEqual( + self.jet_test.get_variable_value("test_var_1"), + "saved_value1", + "Variable value should be restored when flying to waypoint", + ) + self.assertEqual( + self.jet_test.get_variable_value("test_var_2"), + "saved_value2", + "Variable value should be restored when flying to waypoint", + ) + + def test_fly_to_restores_variable_values_with_previous_waypoint(self): + """ + Test that fly_to() restores variable values + after previous waypoint saves its values + """ + # Create previous waypoint in current state + previous_waypoint = self.JetWaypoint.create( + { + "name": "Previous Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "current", + } + ) + self.jet_test.waypoint_id = previous_waypoint.id + + # Set variable values on the jet + self.jet_test.set_variable_value("test_var_1", "previous_value1") + self.jet_test.set_variable_value("test_var_2", "previous_value2") + + # Create destination waypoint with saved variable values + destination_waypoint = self.JetWaypoint.create( + { + "name": "Destination Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "ready", + "variable_values": { + "test_var_1": "destination_value1", + "test_var_2": "destination_value2", + }, + } + ) + + # Ensure previous waypoint has no plan_leave_id (so it saves values immediately) + previous_waypoint.waypoint_template_id.plan_leave_id = False + + # Call fly_to + destination_waypoint.fly_to() + + # Previous waypoint should have saved its values + previous_values = previous_waypoint.variable_values or {} + self.assertEqual( + previous_values.get("test_var_1"), + "previous_value1", + "Previous waypoint should save its variable values", + ) + + # Variable values should be restored from destination waypoint + self.assertEqual( + self.jet_test.get_variable_value("test_var_1"), + "destination_value1", + "Variable value should be restored from destination waypoint", + ) + self.assertEqual( + self.jet_test.get_variable_value("test_var_2"), + "destination_value2", + "Variable value should be restored from destination waypoint", + ) + + def test_arriving_error_restores_variable_values(self): + """ + Test that when arriving fails, + variable values are restored from current waypoint + """ + # Create current waypoint with saved variable values + current_waypoint = self.JetWaypoint.create( + { + "name": "Current Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "current", + "variable_values": { + "test_var_1": "current_value1", + "test_var_2": "current_value2", + }, + } + ) + self.jet_test.waypoint_id = current_waypoint.id + + # Create arriving waypoint + arriving_waypoint = self.JetWaypoint.create( + { + "name": "Arriving Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "state": "arriving", + } + ) + + # Set different values on the jet + self.jet_test.set_variable_value("test_var_1", "arriving_value1") + self.jet_test.set_variable_value("test_var_2", "arriving_value2") + + # Create plan log with error status + plan_log = self.PlanLog.create( + { + "server_id": self.jet_test.server_id.id, + "plan_id": self.plan_error.id, + "plan_status": -100, # Error + } + ) + + # Call _plan_finished with error + with mute_logger( + "odoo.addons.cetmix_tower_server.models.cx_tower_jet_waypoint" + ): + arriving_waypoint._plan_finished(plan_log) + + # Variable values should be restored from current waypoint + self.assertEqual( + self.jet_test.get_variable_value("test_var_1"), + "current_value1", + "Variable value should be restored from current waypoint on error", + ) + self.assertEqual( + self.jet_test.get_variable_value("test_var_2"), + "current_value2", + "Variable value should be restored from current waypoint on error", + ) + + # Current waypoint state should be "current" + self.assertEqual( + current_waypoint.state, + "current", + "Current waypoint state should remain current", + ) + + # Arriving waypoint state should be "error" + self.assertEqual( + arriving_waypoint.state, + "error", + "Arriving waypoint state should be error", + ) + + # ------------------------------------ + # --- _check_is_destination tests ---- + # ------------------------------------ + + def _make_destination_waypoint(self, name, jet=None): + """ + Helper: create a waypoint and atomically transition it to the + ``preparing`` state with ``is_destination=True``. + + This mirrors what ``prepare(is_destination=True)`` does internally + when the waypoint template has a ``plan_create_id`` (it writes + ``state=preparing`` + ``is_destination`` in one call and does not + proceed to ``fly_to()``). Using that path keeps ``is_destination`` + stable for subsequent constraint assertions, whereas calling + ``prepare()`` without a plan triggers ``fly_to()`` ā†’ ``_arrive()``, + which clears ``is_destination`` immediately. + + Args: + name (str): Name of the waypoint. + jet (cx.tower.jet, optional): Target jet. Defaults to jet_test. + + Returns: + cx.tower.jet.waypoint: Waypoint in ``preparing`` state with + ``is_destination=True``. + """ + if jet is None: + jet = self.jet_test + waypoint = self.JetWaypoint.create( + { + "name": name, + "jet_id": jet.id, + "waypoint_template_id": self.waypoint_template.id, + } + ) + waypoint.write({"state": "preparing", "is_destination": True}) + return waypoint + + def test_check_is_destination_single_allowed(self): + """ + Preparing one destination waypoint for a jet via prepare() is valid. + """ + waypoint = self._make_destination_waypoint("Destination Waypoint") + self.assertTrue(waypoint.is_destination) + + def test_check_is_destination_different_jets_allowed(self): + """ + Each jet may independently have its own destination waypoint. + """ + self._make_destination_waypoint("Destination Jet Test", jet=self.jet_test) + waypoint_other = self._make_destination_waypoint( + "Destination Jet Odoo", jet=self.jet_odoo + ) + self.assertTrue(waypoint_other.is_destination) + + def test_check_is_destination_false_ignored(self): + """ + Waypoints with is_destination=False are never checked, even when + another destination already exists for the same jet. + """ + self._make_destination_waypoint("Existing Destination") + # Creating a non-destination waypoint must not raise. + non_dest = self.JetWaypoint.create( + { + "name": "Non Destination", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + "is_destination": False, + } + ) + self.assertFalse(non_dest.is_destination) + + def _assert_state_blocks_destination(self, state): + """ + Helper: create a waypoint, force it into ``state``, then assert that + writing ``is_destination=True`` raises a ValidationError. + + Args: + state (str): Waypoint state to test. + """ + waypoint = self.JetWaypoint.create( + { + "name": f"Waypoint in {state}", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + } + ) + waypoint.write({"state": state}) + with self.assertRaises(ValidationError): + waypoint.write({"is_destination": True}) + + def test_check_is_destination_draft_state_raises(self): + """ + Setting is_destination=True directly on a waypoint in the 'draft' state + must raise a ValidationError. + Use prepare(is_destination=True) to designate a destination waypoint. + """ + self._assert_state_blocks_destination("draft") + + def test_check_is_destination_error_state_raises(self): + """ + Setting is_destination=True on a waypoint in the 'error' state + must raise a ValidationError. + """ + self._assert_state_blocks_destination("error") + + def test_check_is_destination_leaving_state_raises(self): + """ + Setting is_destination=True on a waypoint in the 'leaving' state + must raise a ValidationError. + """ + self._assert_state_blocks_destination("leaving") + + def test_check_is_destination_deleting_state_raises(self): + """ + Setting is_destination=True on a waypoint in the 'deleting' state + must raise a ValidationError. + """ + self._assert_state_blocks_destination("deleting") + + def test_check_is_destination_deleted_state_raises(self): + """ + Setting is_destination=True on a waypoint in the 'deleted' state + must raise a ValidationError. + """ + self._assert_state_blocks_destination("deleted") + + def test_check_is_destination_duplicate_on_create_raises(self): + """ + Setting is_destination via prepare() then trying to prepare a second + destination for the same jet must raise a ValidationError. + """ + self._make_destination_waypoint("First Destination") + second = self.JetWaypoint.create( + { + "name": "Second Destination", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + } + ) + with self.assertRaises(ValidationError): + second.write({"state": "ready", "is_destination": True}) + + def test_check_is_destination_duplicate_on_write_raises(self): + """ + Writing is_destination=True on a second ready waypoint for the same jet + must raise a ValidationError. + """ + self._make_destination_waypoint("Existing Destination") + second = self.JetWaypoint.create( + { + "name": "Second Waypoint", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + } + ) + second.write({"state": "ready"}) + with self.assertRaises(ValidationError): + second.write({"is_destination": True}) + + def test_check_is_destination_duplicate_within_same_batch_raises(self): + """ + Writing is_destination=True on two ready waypoints for the same jet + in a single write() call must raise a ValidationError. + + Both records are excluded from the DB search (neither is a destination + yet), so the constraint must also detect duplicates within the batch. + """ + wp1 = self.JetWaypoint.create( + { + "name": "Batch Destination 1", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + } + ) + wp2 = self.JetWaypoint.create( + { + "name": "Batch Destination 2", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + } + ) + (wp1 | wp2).write({"state": "ready"}) + with self.assertRaises(ValidationError): + (wp1 | wp2).write({"is_destination": True}) + + # ------------------------------------ + # --- unlink destination guard tests - + # ------------------------------------ + + @mute_logger("odoo.addons.cetmix_tower_server.models.cx_tower_jet_waypoint") + def test_unlink_destination_waypoint_raises(self): + """ + Deleting a waypoint with is_destination=True must raise a + ValidationError regardless of state, to prevent the jet from being + stranded mid-flight while a leave plan is still running. + """ + waypoint = self._make_destination_waypoint("Active Destination") + with self.assertRaises(ValidationError): + waypoint.unlink() + + @mute_logger("odoo.addons.cetmix_tower_server.models.cx_tower_jet_waypoint") + def test_unlink_destination_waypoint_no_raise_context_logs(self): + """ + When waypoint_no_raise_on_delete=True is set in context, deleting a + destination waypoint must not raise but must log the error and skip + the record. + """ + waypoint = self._make_destination_waypoint("Active Destination No Raise") + waypoint.with_context(waypoint_no_raise_on_delete=True).unlink() + # Record must still exist ā€” it was skipped, not deleted. + self.assertTrue(waypoint.exists()) + + def test_unlink_non_destination_ready_waypoint_allowed(self): + """ + Deleting a ready waypoint that is NOT a destination must still work. + """ + waypoint = self.JetWaypoint.create( + { + "name": "Ready Non-Destination", + "jet_id": self.jet_test.id, + "waypoint_template_id": self.waypoint_template.id, + } + ) + waypoint.write({"state": "ready"}) + waypoint.unlink() + self.assertFalse(waypoint.exists())