update tests and add new for post expedition report

Author: j-atkins

tests/make_realistic/problems/test_simulator.py

@@ -1,7 +1,12 @@
 import json
+import random
 from datetime import datetime, timedelta
 
-from virtualship.make_realistic.problems.scenarios import GeneralProblem
+from virtualship.instruments.types import InstrumentType
+from virtualship.make_realistic.problems.scenarios import (
+    GeneralProblem,
+    InstrumentProblem,
+)
 from virtualship.make_realistic.problems.simulator import ProblemSimulator
 from virtualship.models.expedition import (
     Expedition,
@@ -11,21 +16,26 @@
     Waypoint,
 )
 from virtualship.models.location import Location
-from virtualship.utils import GENERAL_PROBLEM_REG
+from virtualship.utils import GENERAL_PROBLEM_REG, REPORT
 
 
 def _make_simple_expedition(
-    num_waypoints: int = 2, distance_scale: float = 1.0
+    num_waypoints: int = 2, distance_scale: float = 1.0, no_instruments: bool = False
 ) -> Expedition:
+    """Func. rather than fixture to allow for configurability in different tests."""
     sample_datetime = datetime(2024, 1, 1, 0, 0, 0)
+    instruments_non_underway = [inst for inst in InstrumentType if not inst.is_underway]
+
     waypoints = []
     for i in range(num_waypoints):
         wp = Waypoint(
             location=Location(
                 latitude=0.0 + i * distance_scale, longitude=0.0 + i * distance_scale
             ),
             time=sample_datetime + timedelta(days=i),
-            instrument=[],  # ensure is list, not None
+            instrument=[]
+            if no_instruments
+            else random.sample(instruments_non_underway, 3),
         )
         waypoints.append(wp)
 
@@ -39,8 +49,9 @@ def _make_simple_expedition(
 
 def test_select_problems_single_waypoint_returns_pre_departure(tmp_path):
     expedition = _make_simple_expedition(num_waypoints=1)
+    instruments_in_expedition = expedition.get_instruments()
     simulator = ProblemSimulator(expedition, str(tmp_path))
-    problems = simulator.select_problems(set(), prob_level=2)
+    problems = simulator.select_problems(instruments_in_expedition, prob_level=2)
 
     assert isinstance(problems, dict)
     assert len(problems["problem_class"]) == 1
@@ -51,11 +62,28 @@ def test_select_problems_single_waypoint_returns_pre_departure(tmp_path):
     assert getattr(problem_cls, "pre_departure", False) is True
 
 
+def test_no_instruments_no_instruments_problems(tmp_path):
+    expedition = _make_simple_expedition(num_waypoints=2, no_instruments=True)
+    instruments_in_expedition = expedition.get_instruments()
+    assert len(instruments_in_expedition) == 0, "Expedition should have no instruments"
+
+    simulator = ProblemSimulator(expedition, str(tmp_path))
+    problems = simulator.select_problems(instruments_in_expedition, prob_level=2)
+
+    has_instrument_problems = any(
+        issubclass(cls, InstrumentProblem) for cls in problems["problem_class"]
+    )
+    assert not has_instrument_problems, (
+        "Should not select instrument problems when no instruments are present"
+    )
+
+
 def test_select_problems_prob_level_zero():
     expedition = _make_simple_expedition(num_waypoints=2)
+    instruments_in_expedition = expedition.get_instruments()
     simulator = ProblemSimulator(expedition, ".")
 
-    problems = simulator.select_problems(set(), prob_level=0)
+    problems = simulator.select_problems(instruments_in_expedition, prob_level=0)
     assert problems is None
 
 
@@ -64,10 +92,10 @@ def test_cache_and_load_selected_problems_roundtrip(tmp_path):
     simulator = ProblemSimulator(expedition, str(tmp_path))
 
     # pick two general problems (registry should contain entries)
-    cls1 = GENERAL_PROBLEM_REG[0]
-    cls2 = GENERAL_PROBLEM_REG[1] if len(GENERAL_PROBLEM_REG) > 1 else cls1
+    problem1 = GENERAL_PROBLEM_REG[0]
+    problem2 = GENERAL_PROBLEM_REG[1] if len(GENERAL_PROBLEM_REG) > 1 else problem1
 
-    problems = {"problem_class": [cls1, cls2], "waypoint_i": [None, 0]}
+    problems = {"problem_class": [problem1, problem2], "waypoint_i": [None, 0]}
 
     sel_fpath = tmp_path / "subdir" / "selected_problems.json"
     simulator.cache_selected_problems(problems, str(sel_fpath))
@@ -89,11 +117,11 @@ def test_hash_to_json(tmp_path):
     expedition = _make_simple_expedition(num_waypoints=2)
     simulator = ProblemSimulator(expedition, str(tmp_path))
 
-    cls = GENERAL_PROBLEM_REG[0]
+    any_problem = GENERAL_PROBLEM_REG[0]
 
     hash_path = tmp_path / "problem_hash.json"
     simulator._hash_to_json(
-        cls, "deadbeef", None, hash_path
+        any_problem, "deadbeef", None, hash_path
     )  # "deadbeef" as sub for hex in test
 
     assert hash_path.exists()
@@ -107,18 +135,27 @@ def test_hash_to_json(tmp_path):
 def test_has_contingency_pre_departure(tmp_path):
     expedition = _make_simple_expedition(num_waypoints=2)
     simulator = ProblemSimulator(expedition, str(tmp_path))
-    cls = GENERAL_PROBLEM_REG[0]
 
-    # _has_contingency should return False for pre-departure (None)
-    assert simulator._has_contingency(cls, None) is False
+    pre_departure_problem = next(
+        gp for gp in GENERAL_PROBLEM_REG if getattr(gp, "pre_departure", False)
+    )
+    assert pre_departure_problem is not None, (
+        "Need at least one pre-departure problem class in the general problem registry"
+    )
+
+    # _has_contingency should return False for pre-departure (waypoint = None)
+    assert simulator._has_contingency(pre_departure_problem, None) is False
 
 
 def test_select_problems_prob_levels(tmp_path):
     expedition = _make_simple_expedition(num_waypoints=3)
+    instruments_in_expedition = expedition.get_instruments()
     simulator = ProblemSimulator(expedition, str(tmp_path))
 
     for level in range(3):  # prob levels 0, 1, 2
-        problems = simulator.select_problems(set(), prob_level=level)
+        problems = simulator.select_problems(
+            instruments_in_expedition, prob_level=level
+        )
         if level == 0:
             assert problems is None
         else:
@@ -132,13 +169,22 @@ def test_select_problems_prob_levels(tmp_path):
 def test_prob_level_two_more_problems(tmp_path):
     prob_level = 2
 
-    short_expedition = _make_simple_expedition(num_waypoints=2)
+    short_expedition = _make_simple_expedition(
+        num_waypoints=2
+    )  # short in terms of number of waypoints
+    instruments_in_short_expedition = short_expedition.get_instruments()
     simulator_short = ProblemSimulator(short_expedition, str(tmp_path))
+
     long_expedition = _make_simple_expedition(num_waypoints=12)
+    instruments_in_long_expedition = long_expedition.get_instruments()
     simulator_long = ProblemSimulator(long_expedition, str(tmp_path))
 
-    problems_short = simulator_short.select_problems(set(), prob_level=prob_level)
-    problems_long = simulator_long.select_problems(set(), prob_level=prob_level)
+    problems_short = simulator_short.select_problems(
+        instruments_in_short_expedition, prob_level=prob_level
+    )
+    problems_long = simulator_long.select_problems(
+        instruments_in_long_expedition, prob_level=prob_level
+    )
 
     assert len(problems_long["problem_class"]) >= len(
         problems_short["problem_class"]
@@ -165,3 +211,31 @@ def test_has_contingency_during_expedition(tmp_path):
     # short distance expedition should have contingency, long distance should not (given time between waypoints and ship speed is constant)
     assert short_simulator._has_contingency(problem_cls, problem_waypoint_i=0) is True
     assert long_simulator._has_contingency(problem_cls, problem_waypoint_i=0) is False
+
+
+def test_post_expedition_report(tmp_path):
+    expedition = _make_simple_expedition(
+        num_waypoints=12
+    )  # longer expedition to increase likelihood of multiple problems at prob_level=2
+    instruments_in_expedition = expedition.get_instruments()
+
+    simulator = ProblemSimulator(expedition, str(tmp_path))
+    problems = simulator.select_problems(instruments_in_expedition, prob_level=2)
+
+    report_path = tmp_path / REPORT
+    simulator.post_expedition_report(problems, report_path)
+
+    assert report_path.exists()
+    with open(report_path, encoding="utf-8") as f:
+        content = f.read()
+
+    assert content.count("Problem:") == len(problems["problem_class"]), (
+        "Number of reported problems should match number of selected problems."
+    )
+    assert content.count("Delay caused:") == len(problems["problem_class"]), (
+        "Number of reported delay durations should match number of selected problems."
+    )
+    for problem in problems["problem_class"]:
+        assert problem.message in content, (
+            "Problem messages in report should match those of selected problems."
+        )