Move eventhandler none assert to function

Author: andchiind

src/isar/state_machine/state.py

@@ -42,27 +42,25 @@ def __init__(
         self.event_handler_mappings = event_handler_mappings
         self.timers = timers if timers is not None else []
 
-    def get_event_handler_by_name(
-        self, event_handler_name: str
-    ) -> EventHandlerMapping | None:
+    def get_event_handler_by_name(self, event_handler_name: str) -> EventHandlerMapping:
         filtered_handlers = list(
             filter(
                 lambda mapping: mapping.name == event_handler_name,
                 self.event_handler_mappings,
             )
         )
-        return filtered_handlers[0] if len(filtered_handlers) > 0 else None
+        assert len(filtered_handlers) > 0
+        return filtered_handlers[0]
 
-    def get_event_timer_by_name(
-        self, event_timer_name: str
-    ) -> TimeoutHandlerMapping | None:
+    def get_event_timer_by_name(self, event_timer_name: str) -> TimeoutHandlerMapping:
         filtered_timers = list(
             filter(
                 lambda mapping: mapping.name == event_timer_name,
                 self.timers,
             )
         )
-        return filtered_timers[0] if len(filtered_timers) > 0 else None
+        assert len(filtered_timers) > 0
+        return filtered_timers[0]
 
     def handles_event(self, event: Event) -> bool:
         allowed_events: list[Event] = [m.event for m in self.event_handler_mappings]

tests/isar/state_machine/states/test_await_next_mission_state.py

@@ -82,12 +82,10 @@ def test_state_machine_with_successful_mission_stop(
 def test_transition_from_resuming_to_paused(events: Events) -> None:
     current_state = Resuming(events, "mission_id")
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "failed_resume_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(
         ErrorMessage(
             error_reason=ErrorReason.RobotUnknownErrorException, error_description=""
@@ -103,10 +101,9 @@ def test_unknown_status_transitions_to_await_next_mission_if_it_was_already_avai
 ) -> None:
     current_state = UnknownStatus(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "robot_status_event"
     )
-    assert event_handler is not None
 
     transition = event_handler.handler(RobotStatus.Available)
 
@@ -119,12 +116,10 @@ def test_transition_from_resuming_return_home_to_await_next_mission(
 ) -> None:
     current_state = ResumingReturnHome(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "failed_resume_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(
         ErrorMessage(
             error_reason=ErrorReason.RobotUnknownErrorException, error_description=""

tests/isar/state_machine/states/test_going_to_lockdown_state.py

@@ -16,7 +16,7 @@ def test_transition_from_return_home_paused_to_going_to_lockdown(
 ) -> None:
     current_state: State = ReturnHomePaused(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "send_to_lockdown_event"
     )
 
@@ -27,10 +27,9 @@ def test_transition_from_return_home_paused_to_going_to_lockdown(
     assert events.api_requests.send_to_lockdown.response.has_event()
     assert type(current_state) is GoingToLockdown
 
-    lockdown_event_handler: EventHandlerMapping | None = (
+    lockdown_event_handler: EventHandlerMapping = (
         current_state.get_event_handler_by_name("mission_failed_to_resume")
     )
-    assert lockdown_event_handler is not None
 
     transition = lockdown_event_handler.handler(
         ErrorMessage(
@@ -46,12 +45,10 @@ def test_transition_from_return_home_paused_to_going_to_lockdown(
 def test_stopping_lockdown_transitions_to_going_to_lockdown(events: Events) -> None:
     current_state = StoppingGoToLockdown(events, "mission_id")
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "successful_stop_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(EmptyMessage())
 
     current_state = transition(events)
@@ -69,12 +66,10 @@ def test_stopping_lockdown_transitions_to_going_to_lockdown(events: Events) -> N
 def test_return_home_transitions_to_going_to_lockdown(events: Events) -> None:
     current_state = ReturningHome(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "send_to_lockdown_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(EmptyMessage())
 
     current_state = transition(events)
@@ -84,12 +79,10 @@ def test_return_home_transitions_to_going_to_lockdown(events: Events) -> None:
 def test_recharging_transitions_to_going_to_lockdown(events: Events) -> None:
     current_state = GoingToRecharging(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "send_to_lockdown_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(EmptyMessage())
 
     current_state = transition(events)
@@ -99,12 +92,10 @@ def test_recharging_transitions_to_going_to_lockdown(events: Events) -> None:
 def test_await_next_mission_transitions_to_going_to_lockdown(events: Events) -> None:
     current_state = AwaitNextMission(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "send_to_lockdown_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(EmptyMessage())
 
     current_state = transition(events)

tests/isar/state_machine/states/test_going_to_recharging_state.py

@@ -12,12 +12,10 @@ def test_stopping_to_recharge_goes_to_going_to_recharging_when_no_remaining_task
     events: Events,
 ) -> None:
     current_state = StoppingGoToRecharge(events)
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "mission_already_done_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(EmptyMessage())
 
     current_state = transition(events)
@@ -28,12 +26,10 @@ def test_stopping_to_recharge_goes_to_going_to_recharging_with_aborted_mission(
     events: Events,
 ) -> None:
     current_state = StoppingGoToRecharge(events)
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "successful_stop_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(AbortedMission(id="id", name="test"))
 
     assert events.mqtt_queue.empty()
@@ -44,12 +40,10 @@ def test_stopping_to_recharge_goes_to_going_to_recharging_with_aborted_mission(
 
 def test_return_home_goes_to_recharging_when_battery_low(events: Events) -> None:
     current_state = ReturningHome(events)
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "robot_battery_below_threshold_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(EmptyMessage())
 
     current_state = transition(events)
@@ -60,12 +54,10 @@ def test_cancelling_mission_when_going_home_to_recharge(events: Events) -> None:
     current_state = GoingToRechargingWithMission(
         events, mission=AbortedMission(name="test", id="test_id")
     )
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "stop_mission_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler("test_id")
 
     current_state = transition(events)

tests/isar/state_machine/states/test_home_state.py

@@ -12,12 +12,10 @@
 def test_lockdown_transitions_to_home(events: Events) -> None:
     current_state = Lockdown(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "release_from_lockdown"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(EmptyMessage())
 
     assert events.api_requests.release_from_lockdown.response.check()
@@ -30,16 +28,13 @@ def test_state_machine_with_return_home_failure_successful_retries(
 ) -> None:
     current_state = ReturningHome(events)
 
-    event_handler_success: EventHandlerMapping | None = (
+    event_handler_success: EventHandlerMapping = (
         current_state.get_event_handler_by_name("mission_succeeded_event")
     )
-    event_handler_failure: EventHandlerMapping | None = (
+    event_handler_failure: EventHandlerMapping = (
         current_state.get_event_handler_by_name("mission_failed_event")
     )
 
-    assert event_handler_success is not None
-    assert event_handler_failure is not None
-
     transition = event_handler_failure.handler(
         ErrorMessage(
             error_reason=ErrorReason.RobotUnknownErrorException,
@@ -61,10 +56,9 @@ def test_intervention_needed_transitions_to_home_if_robot_is_home(
 ) -> None:
     current_state = InterventionNeeded(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "robot_status_event"
     )
-    assert event_handler is not None
 
     transition = event_handler.handler(RobotStatus.Home)
 
@@ -77,12 +71,10 @@ def test_intervention_needed_transitions_to_home_if_robot_is_home(
 def test_recharging_goes_to_home_when_battery_high(events: Events) -> None:
     current_state = Recharging(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "robot_battery_above_recharge_threshold_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(EmptyMessage())
 
     current_state = transition(events)

tests/isar/state_machine/states/test_intervention_needed_state.py

@@ -11,12 +11,10 @@
 
 def test_going_to_recharging_goes_to_intervention_needed(events: Events) -> None:
     current_state = GoingToRecharging(events)
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "mission_failed_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(
         ErrorMessage(
             error_reason=ErrorReason.RobotUnknownErrorException,
@@ -33,12 +31,10 @@ def test_going_to_lockdown_task_failed_transitions_to_intervention_needed(
 ) -> None:
     current_state = GoingToLockdown(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "mission_failed_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(
         ErrorMessage(
             error_reason=ErrorReason.RobotUnknownErrorException,
@@ -55,12 +51,10 @@ def test_going_to_lockdown_mission_failed_transitions_to_intervention_needed(
 ) -> None:
     current_state = GoingToLockdown(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "mission_failed_event"
     )
 
-    assert event_handler is not None
-
     # The type of error reason is not important for this test
     transition = event_handler.handler(
         ErrorMessage(error_description="", error_reason=ErrorReason.RobotAPIException)
@@ -73,7 +67,7 @@ def test_going_to_lockdown_mission_failed_transitions_to_intervention_needed(
 def test_state_machine_with_return_home_failure(events: Events) -> None:
     current_state = ReturningHome(events)
 
-    failure_event_handler: EventHandlerMapping | None
+    failure_event_handler: EventHandlerMapping
 
     for i in range(settings.RETURN_HOME_RETRY_LIMIT - 1):
 
@@ -112,10 +106,9 @@ def test_intervention_needed_transitions_does_not_transition_if_status_is_not_ho
 ) -> None:
     current_state = InterventionNeeded(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "robot_status_event"
     )
-    assert event_handler is not None
 
     statuses = [
         RobotStatus.Available,

tests/isar/state_machine/states/test_lockdown_state.py

@@ -9,12 +9,10 @@
 def test_mission_stopped_when_going_to_lockdown(events: Events) -> None:
     current_state = Monitor(events, "mission_id")
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "send_to_lockdown_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(EmptyMessage())
 
     current_state = transition(events)
@@ -24,12 +22,10 @@ def test_mission_stopped_when_going_to_lockdown(events: Events) -> None:
 def test_going_to_lockdown_transitions_to_lockdown(events: Events) -> None:
     current_state = GoingToLockdown(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "mission_succeeded_event"
     )
 
-    assert event_handler is not None
-
     transition = event_handler.handler(EmptyMessage())
 
     current_state = transition(events)

tests/isar/state_machine/states/test_maintenance_mode_state.py

@@ -12,10 +12,9 @@ def test_home_transitions_to_maintenance_mode_when_teleoperating(
 ) -> None:
     current_state = Home(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "robot_status_event"
     )
-    assert event_handler is not None
 
     transition = event_handler.handler(RobotStatus.TeleOperation)
 
@@ -30,10 +29,9 @@ def test_unknown_status_transitions_to_maintenance_mode_when_teleoperating(
 ) -> None:
     current_state = UnknownStatus(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "robot_status_event"
     )
-    assert event_handler is not None
 
     transition = event_handler.handler(RobotStatus.TeleOperation)
 
@@ -48,10 +46,9 @@ def test_offline_transitions_to_maintenance_mode_when_teleoperating(
 ) -> None:
     current_state = Offline(events)
 
-    event_handler: EventHandlerMapping | None = current_state.get_event_handler_by_name(
+    event_handler: EventHandlerMapping = current_state.get_event_handler_by_name(
         "robot_status_event"
     )
-    assert event_handler is not None
 
     transition = event_handler.handler(RobotStatus.TeleOperation)