diff --git a/monitoring/monitorlib/geo.py b/monitoring/monitorlib/geo.py index a6cbc4faab..539e6668e8 100644 --- a/monitoring/monitorlib/geo.py +++ b/monitoring/monitorlib/geo.py @@ -158,6 +158,28 @@ def from_f3548v21(vol: f3548v21.Polygon | dict) -> Polygon: vertices=[ImplicitDict.parse(p, LatLngPoint) for p in vol.vertices] ) + def is_equivalent(self, other: Polygon) -> bool: + if "vertices" not in self and "vertices" not in other: + return True + elif "vertices" not in self or "vertices" not in other: + return False + + if self.vertices == other.vertices: + # covers both None and exact equality + return True + elif not self.vertices or not other.vertices: + # covers one being None + return False + elif len(self.vertices) != len(other.vertices): + return False + + return all( + [ + vertices[0].match(vertices[1]) + for vertices in zip(self.vertices, other.vertices) + ] + ) + class Circle(ImplicitDict): center: LatLngPoint @@ -181,6 +203,15 @@ def from_f3548v21(vol: f3548v21.Circle | dict) -> Circle: radius=ImplicitDict.parse(vol.radius, Radius), ) + def is_equivalent(self, other: Circle) -> bool: + if not self.center.match(other.center): + return False + + return ( + abs(self.radius.in_meters() - other.radius.in_meters()) + <= DISTANCE_TOLERANCE_M + ) + class AltitudeDatum(str, Enum): W84 = "W84" @@ -224,6 +255,14 @@ def to_w84_m(self) -> float: def from_f3548v21(vol: f3548v21.Altitude | dict) -> Altitude: return ImplicitDict.parse(vol, Altitude) + def is_equivalent(self, other: Altitude) -> bool: + if self.reference != other.reference: + return False + return ( + abs(self.units.in_meters(self.value) - other.units.in_meters(other.value)) + <= DISTANCE_TOLERANCE_M + ) + class Volume3D(ImplicitDict): outline_circle: Optional[Circle] = None @@ -447,6 +486,38 @@ def s2_vertices(self) -> list[s2sphere.LatLng]: else: return get_latlngrect_vertices(make_latlng_rect(self)) + def is_equivalent( + self, + other: Volume3D, + ) -> bool: + if self.altitude_lower and other.altitude_lower: + if not self.altitude_lower.is_equivalent(other.altitude_lower): + return False + elif self.altitude_lower or other.altitude_lower: + return False + + if self.altitude_upper and other.altitude_upper: + if not self.altitude_upper.is_equivalent(other.altitude_upper): + return False + elif self.altitude_upper or other.altitude_upper: + return False + + if self.outline_polygon and other.outline_polygon: + if not self.outline_polygon.is_equivalent(other.outline_polygon): + return False + elif self.outline_circle and other.outline_circle: + if not self.outline_circle.is_equivalent(other.outline_circle): + return False + elif ( + self.outline_circle + or self.outline_polygon + or other.outline_circle + or other.outline_polygon + ): + return False + + return True + def make_latlng_rect(area) -> s2sphere.LatLngRect: """Make an S2 LatLngRect from the provided input. diff --git a/monitoring/monitorlib/geo_test.py b/monitoring/monitorlib/geo_test.py index a8396638dd..58938ed25c 100644 --- a/monitoring/monitorlib/geo_test.py +++ b/monitoring/monitorlib/geo_test.py @@ -1,6 +1,16 @@ +import unittest + from s2sphere import LatLng from monitoring.monitorlib.geo import ( + Altitude, + AltitudeDatum, + Circle, + DistanceUnits, + LatLngPoint, + Polygon, + Radius, + Volume3D, generate_area_in_vicinity, generate_slight_overlap_area, ) @@ -12,6 +22,196 @@ def _points(in_points: list[tuple[float, float]]) -> list[LatLng]: return [LatLng.from_degrees(*p) for p in in_points] +class AltitudeIsEquivalentTest(unittest.TestCase): + def setUp(self): + self.alt1 = Altitude( + value=100, reference=AltitudeDatum.W84, units=DistanceUnits.M + ) + + def test_equivalent_altitudes(self): + alt2 = Altitude(value=100, reference=AltitudeDatum.W84, units=DistanceUnits.M) + self.assertTrue(self.alt1.is_equivalent(alt2)) + + def test_equivalent_altitudes_within_tolerance(self): + alt2 = Altitude( + value=100.000001, reference=AltitudeDatum.W84, units=DistanceUnits.M + ) + self.assertTrue(self.alt1.is_equivalent(alt2)) + + def test_equivalent_altitudes_different_units(self): + alt2 = Altitude( + value=328.084, reference=AltitudeDatum.W84, units=DistanceUnits.FT + ) + self.assertTrue(self.alt1.is_equivalent(alt2)) + + def test_nonequivalent_altitudes_different_value(self): + alt2 = Altitude(value=101, reference=AltitudeDatum.W84, units=DistanceUnits.M) + self.assertFalse(self.alt1.is_equivalent(alt2)) + + def test_nonequivalent_altitudes_different_reference(self): + alt2 = Altitude(value=100, reference=AltitudeDatum.SFC, units=DistanceUnits.M) + self.assertFalse(self.alt1.is_equivalent(alt2)) + + +class PolygonIsEquivalentTest(unittest.TestCase): + def setUp(self): + self.poly1 = Polygon( + vertices=[ + LatLngPoint(lat=10, lng=10), + LatLngPoint(lat=11, lng=10), + LatLngPoint(lat=11, lng=11), + LatLngPoint(lat=10, lng=11), + ] + ) + + def test_equivalent_polygons(self): + poly2 = Polygon( + vertices=[ + LatLngPoint(lat=10, lng=10), + LatLngPoint(lat=11, lng=10), + LatLngPoint(lat=11, lng=11), + LatLngPoint(lat=10, lng=11), + ] + ) + self.assertTrue(self.poly1.is_equivalent(poly2)) + + def test_equivalent_polygons_within_tolerance(self): + poly2 = Polygon( + vertices=[ + LatLngPoint(lat=10.00000001, lng=10.00000001), + LatLngPoint(lat=11.00000001, lng=10.00000001), + LatLngPoint(lat=11.00000001, lng=11.00000001), + LatLngPoint(lat=10.00000001, lng=11.00000001), + ] + ) + self.assertTrue(self.poly1.is_equivalent(poly2)) + + def test_nonequivalent_polygons(self): + poly2 = Polygon( + vertices=[ + LatLngPoint(lat=10, lng=10), + LatLngPoint(lat=12, lng=10), + LatLngPoint(lat=12, lng=11), + LatLngPoint(lat=10, lng=11), + ] + ) + self.assertFalse(self.poly1.is_equivalent(poly2)) + + def test_equivalent_polygons_none(self): + poly1 = Polygon(vertices=None) + poly2 = Polygon(vertices=None) + self.assertTrue(poly1.is_equivalent(poly2)) + + def test_nonequivalent_polygons_one_none(self): + poly1 = Polygon(vertices=[]) + poly2 = Polygon(vertices=None) + self.assertFalse(poly1.is_equivalent(poly2)) + + +class Volume3DIsEquivalentTest(unittest.TestCase): + def setUp(self): + self.vol_poly = Volume3D( + outline_polygon=Polygon( + vertices=[ + LatLngPoint(lat=10, lng=10), + LatLngPoint(lat=11, lng=10), + LatLngPoint(lat=11, lng=11), + LatLngPoint(lat=10, lng=11), + ] + ), + altitude_lower=Altitude( + value=100, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + altitude_upper=Altitude( + value=200, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + ) + self.vol_circle = Volume3D( + outline_circle=Circle( + center=LatLngPoint(lat=10, lng=10), + radius=Radius(value=100, units=DistanceUnits.M), + ), + altitude_lower=Altitude( + value=100, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + altitude_upper=Altitude( + value=200, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + ) + + def test_equivalent_volumes_polygon(self): + vol2 = Volume3D( + outline_polygon=Polygon( + vertices=[ + LatLngPoint(lat=10, lng=10), + LatLngPoint(lat=11, lng=10), + LatLngPoint(lat=11, lng=11), + LatLngPoint(lat=10, lng=11), + ] + ), + altitude_lower=Altitude( + value=100, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + altitude_upper=Altitude( + value=200, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + ) + self.assertTrue(self.vol_poly.is_equivalent(vol2)) + + def test_equivalent_volumes_circle(self): + vol2 = Volume3D( + outline_circle=Circle( + center=LatLngPoint(lat=10, lng=10), + radius=Radius(value=100, units=DistanceUnits.M), + ), + altitude_lower=Altitude( + value=100, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + altitude_upper=Altitude( + value=200, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + ) + self.assertTrue(self.vol_circle.is_equivalent(vol2)) + + def test_nonequivalent_volumes_circle(self): + vol2 = Volume3D( + outline_circle=Circle( + center=LatLngPoint(lat=10, lng=10), + radius=Radius(value=200, units=DistanceUnits.M), + ), + altitude_lower=Altitude( + value=100, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + altitude_upper=Altitude( + value=200, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + ) + self.assertFalse(self.vol_circle.is_equivalent(vol2)) + + def test_nonequivalent_volumes_different_shape(self): + vol2 = Volume3D( + outline_circle=Circle( + center=LatLngPoint(lat=10.5, lng=10.5), + radius=Radius(value=50000, units=DistanceUnits.M), + ) + ) + self.assertFalse(self.vol_poly.is_equivalent(vol2)) + + def test_equivalent_volumes_none_fields(self): + vol1 = Volume3D() + vol2 = Volume3D() + self.assertTrue(vol1.is_equivalent(vol2)) + + def test_nonequivalent_volumes_one_none_field(self): + vol1 = Volume3D( + altitude_lower=Altitude( + value=100, reference=AltitudeDatum.W84, units=DistanceUnits.M + ) + ) + vol2 = Volume3D() + self.assertFalse(vol1.is_equivalent(vol2)) + + def test_generate_slight_overlap_area(): # Square around 0,0 of edge length 2 -> first corner at 1,1 -> expect a square with overlapping corner at 1,1 assert generate_slight_overlap_area( diff --git a/monitoring/monitorlib/geotemporal.py b/monitoring/monitorlib/geotemporal.py index e4bf3f9843..66e24138ae 100644 --- a/monitoring/monitorlib/geotemporal.py +++ b/monitoring/monitorlib/geotemporal.py @@ -29,6 +29,8 @@ ) from monitoring.monitorlib.transformations import Transformation +TIME_TOLERANCE = timedelta(milliseconds=10) + class Volume4DTemplate(ImplicitDict): outline_polygon: Optional[Polygon] = None @@ -129,6 +131,30 @@ class Volume4D(ImplicitDict): time_start: Optional[Time] = None time_end: Optional[Time] = None + def is_equivalent( + self, + other: Volume4D, + ) -> bool: + if not self.volume.is_equivalent(other.volume): + return False + + if (self.time_start is None) != (other.time_start is None): + return False + if self.time_start and other.time_start: + if ( + abs(self.time_start.datetime - other.time_start.datetime) + > TIME_TOLERANCE + ): + return False + + if (self.time_end is None) != (other.time_end is None): + return False + if self.time_end and other.time_end: + if abs(self.time_end.datetime - other.time_end.datetime) > TIME_TOLERANCE: + return False + + return True + def offset_time(self, dt: timedelta) -> Volume4D: kwargs = {"volume": self.volume} if self.time_start: @@ -296,6 +322,26 @@ def __iadd__(self, other): f"Cannot iadd {type(other).__name__} to {type(self).__name__}" ) + def is_equivalent( + self, + other: Volume4DCollection, + ) -> bool: + if len(self) != len(other): + return False + + # different order is acceptable + other_copy = list(other) + for vol in self: + found = False + for i, other_vol in enumerate(other_copy): + if vol.is_equivalent(other_vol): + other_copy.pop(i) + found = True + break + if not found: + return False + return True + @property def time_start(self) -> Time | None: return ( diff --git a/monitoring/monitorlib/geotemporal_test.py b/monitoring/monitorlib/geotemporal_test.py new file mode 100644 index 0000000000..52de25c148 --- /dev/null +++ b/monitoring/monitorlib/geotemporal_test.py @@ -0,0 +1,138 @@ +import unittest +from datetime import datetime, timedelta + +from monitoring.monitorlib.geo import ( + Altitude, + AltitudeDatum, + Circle, + DistanceUnits, + LatLngPoint, + Radius, + Volume3D, +) +from monitoring.monitorlib.geotemporal import Volume4D, Volume4DCollection +from monitoring.monitorlib.temporal import Time + + +class Volume4DIsEquivalentTest(unittest.TestCase): + def setUp(self): + self.t0 = datetime.now() + self.t1 = self.t0 + timedelta(minutes=10) + self.vol_3d = Volume3D( + outline_circle=Circle( + center=LatLngPoint(lat=10, lng=10), + radius=Radius(value=100, units=DistanceUnits.M), + ), + altitude_lower=Altitude( + value=100, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + altitude_upper=Altitude( + value=200, reference=AltitudeDatum.W84, units=DistanceUnits.M + ), + ) + self.vol1 = Volume4D( + volume=self.vol_3d, + time_start=Time(self.t0), + time_end=Time(self.t1), + ) + + def test_equivalent_volume4d(self): + vol2 = Volume4D( + volume=self.vol_3d, + time_start=Time(self.t0), + time_end=Time(self.t1), + ) + self.assertTrue(self.vol1.is_equivalent(vol2)) + + def test_equivalent_volume4d_within_tolerance(self): + # Time within tolerance (default is 10ms) + vol2 = Volume4D( + volume=self.vol_3d, + time_start=Time(self.t0 + timedelta(milliseconds=3)), + time_end=Time(self.t1 - timedelta(milliseconds=3)), + ) + self.assertTrue(self.vol1.is_equivalent(vol2)) + + def test_nonequivalent_volume4d_time_outside_tolerance(self): + vol2 = Volume4D( + volume=self.vol_3d, + time_start=Time(self.t0 + timedelta(seconds=2)), + time_end=Time(self.t1), + ) + self.assertFalse(self.vol1.is_equivalent(vol2)) + + def test_nonequivalent_volume4d_different_volume3d(self): + vol2 = Volume4D( + volume=Volume3D( + outline_circle=Circle( + center=LatLngPoint(lat=10, lng=10), + radius=Radius(value=200, units=DistanceUnits.M), + ), + altitude_lower=self.vol_3d.altitude_lower, + altitude_upper=self.vol_3d.altitude_upper, + ), + time_start=Time(self.t0), + time_end=Time(self.t1), + ) + self.assertFalse(self.vol1.is_equivalent(vol2)) + + def test_equivalent_volume4d_none_times(self): + vol1_no_time = Volume4D(volume=self.vol_3d) + vol2_no_time = Volume4D(volume=self.vol_3d) + self.assertTrue(vol1_no_time.is_equivalent(vol2_no_time)) + + def test_nonequivalent_volume4d_one_none_time(self): + vol2 = Volume4D(volume=self.vol_3d, time_start=Time(self.t0)) + self.assertFalse(self.vol1.is_equivalent(vol2)) + + +class Volume4DCollectionIsEquivalentTest(unittest.TestCase): + def setUp(self): + self.t0 = datetime.now() + self.v1 = Volume4D( + volume=Volume3D( + outline_circle=Circle( + center=LatLngPoint(lat=10, lng=10), + radius=Radius(value=100, units=DistanceUnits.M), + ) + ), + time_start=Time(self.t0), + ) + self.v2 = Volume4D( + volume=Volume3D( + outline_circle=Circle( + center=LatLngPoint(lat=20, lng=20), + radius=Radius(value=200, units=DistanceUnits.M), + ) + ), + time_start=Time(self.t0), + ) + self.v3 = Volume4D( + volume=Volume3D( + outline_circle=Circle( + center=LatLngPoint(lat=30, lng=30), + radius=Radius(value=300, units=DistanceUnits.M), + ) + ), + time_start=Time(self.t0), + ) + + def test_equivalent_collection_same_order(self): + c1 = Volume4DCollection([self.v1, self.v2]) + c2 = Volume4DCollection([self.v1, self.v2]) + self.assertTrue(c1.is_equivalent(c2)) + + def test_equivalent_collection_different_order(self): + c1 = Volume4DCollection([self.v1, self.v2]) + c2 = Volume4DCollection([self.v2, self.v1]) + self.assertTrue(c1.is_equivalent(c2)) + + def test_nonequivalent_collection_different_lengths(self): + c1 = Volume4DCollection([self.v1]) + c2 = Volume4DCollection([]) + self.assertFalse(c1.is_equivalent(c2)) + + def test_nonequivalent_collection_different_content(self): + c1 = Volume4DCollection([self.v1, self.v2]) + c2 = Volume4DCollection([self.v1, self.v3]) + self.assertFalse(c1.is_equivalent(c2)) diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/evaluation.py b/monitoring/uss_qualifier/scenarios/astm/utm/evaluation.py index 27e9ab06d6..2aa249c235 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/evaluation.py +++ b/monitoring/uss_qualifier/scenarios/astm/utm/evaluation.py @@ -2,6 +2,7 @@ from urllib.parse import urlparse from uas_standards.astm.f3548.v21.api import ( + OperationalIntent, OperationalIntentDetails, OperationalIntentReference, UssAvailabilityState, @@ -121,3 +122,58 @@ def validate_op_intent_details( ) return "; ".join(errors_text) if len(errors_text) > 0 else None + + +def errors_for_nonequivalent_op_intent_details( + old_oi: OperationalIntent, + new_oi: OperationalIntent, +) -> str | None: + errors_text: list[str] = [] + old_ref = old_oi.reference + new_ref = new_oi.reference + old_details = old_oi.details + new_details = new_oi.details + + def append_ovn_err(): + errors_text.append( + f"The OVN {new_ref.ovn} reported by USS for operational intent {new_ref.id} at version {new_ref.version} does not match the value {old_ref.ovn} previously indicated by the USS for the same operational intent with the same version" + ) + + def append_err(field_name: str): + errors_text.append( + f"The value {getattr(new_details, field_name)} for `{field_name}` reported by USS for details of operational intent {new_ref.id} at version {new_ref.version} (OVN {new_ref.ovn if 'ovn' in new_ref else 'empty'}) does not match the value {getattr(old_details, field_name)} previously indicated by the USS for the same operational intent with the same version" + ) + + if "ovn" in old_ref and "ovn" in new_ref: + if old_ref.ovn != new_ref.ovn: + append_ovn_err() + elif "ovn" in old_ref or "ovn" in new_ref: + append_ovn_err() + + if "priority" in old_details and "priority" in new_details: + if old_details.priority != new_details.priority: + append_err("priority") + elif "priority" in old_details or "priority" in new_details: + append_err("priority") + + if (old_details.volumes is None) != (new_details.volumes is None): + append_err("volumes") + elif old_details.volumes and new_details.volumes: + if not Volume4DCollection.from_f3548v21(old_details.volumes).is_equivalent( + Volume4DCollection.from_f3548v21(new_details.volumes) + ): + append_err("volumes") + + if (old_details.off_nominal_volumes is None) != ( + new_details.off_nominal_volumes is None + ): + append_err("off_nominal_volumes") + elif old_details.off_nominal_volumes and new_details.off_nominal_volumes: + if not Volume4DCollection.from_f3548v21( + old_details.off_nominal_volumes + ).is_equivalent( + Volume4DCollection.from_f3548v21(new_details.off_nominal_volumes) + ): + append_err("off_nominal_volumes") + + return "; ".join(errors_text) if errors_text else None diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py b/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py index 942008864b..6708ca2673 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py +++ b/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py @@ -1,12 +1,14 @@ from __future__ import annotations import datetime +from dataclasses import dataclass from enum import Enum -from implicitdict import ImplicitDict, Optional +from implicitdict import ImplicitDict, Optional, StringBasedDateTime from uas_standards.astm.f3548.v21.api import ( EntityID, GetOperationalIntentDetailsResponse, + OperationalIntent, OperationalIntentReference, OperationalIntentState, UssAvailabilityState, @@ -29,6 +31,7 @@ set_uss_availability, ) from monitoring.uss_qualifier.scenarios.astm.utm.evaluation import ( + errors_for_nonequivalent_op_intent_details, validate_op_intent_details, validate_op_intent_reference, ) @@ -43,6 +46,12 @@ NUMERIC_PRECISION_DISTANCE = 0.001 # meters +@dataclass +class CachedOpIntent: + query_timestamp: StringBasedDateTime + op_intent: OperationalIntent + + class OpIntentValidator: """ This class enables the validation of the sharing (or not) of an operational @@ -465,6 +474,36 @@ def _check_op_intent_details( query_timestamps=[oi_full_query.request.timestamp], ) + with self._scenario.check( + "Operational intent details have not changed without publishing a new version to the DSS", + [self._flight_planner.participant_id], + ) as check: + cache_key = ( + f"full_op_intent:{oi_full.reference.id}:{oi_full.reference.version}" + ) + old_oi: CachedOpIntent | None = self._scenario.cache.get(cache_key) + if not old_oi: + self._scenario.cache[cache_key] = CachedOpIntent( + op_intent=oi_full, + query_timestamp=StringBasedDateTime( + oi_full_query.request.timestamp + ), + ) + else: + error_text = errors_for_nonequivalent_op_intent_details( + old_oi.op_intent, + oi_full, + ) + if error_text: + check.record_failed( + summary="Operational intent details have changed without the change being published to the DSS", + details=error_text, + query_timestamps=[ + old_oi.query_timestamp, + oi_full_query.request.timestamp, + ], + ) + with self._scenario.check( "Correct operational intent details", [self._flight_planner.participant_id] ) as check: diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/validate_shared_operational_intent.md b/monitoring/uss_qualifier/scenarios/astm/utm/validate_shared_operational_intent.md index 0cafa7e1f3..2fd099b463 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/validate_shared_operational_intent.md +++ b/monitoring/uss_qualifier/scenarios/astm/utm/validate_shared_operational_intent.md @@ -33,6 +33,12 @@ If the operational intent details response does not validate against [the GetOpe If any of the values in the operational intent reference reported by the USS do not match those values in the operational intent reference published to (and known by) the DSS, save for the OVN, this check will fail per **[astm.f3548.v21.USS0005](../../../requirements/astm/f3548/v21.md)** since the values reported by the USS were not made discoverable via the DSS. +## 🛑 Operational intent details have not changed without publishing a new version to the DSS check + +If the operational intent details reported by the USS have changed without the USS having updated the operational intent reference in the DSS, this check will fail because the USS did not make the changes to the operational intent discoverable using the DSS, per **[astm.f3548.v21.USS0005](../../../requirements/astm/f3548/v21.md)**. Per **[astm.f3548.v21.USS0105,1](../../../requirements/astm/f3548/v21.md)** (definition of `version` field), a USS must update the operational intent reference in the DSS, even when no data fields of the reference have changed, to make changes to the operational intent discoverable using the DSS. + +To be clear, this check specifically targets cases where the USS changes details of the operational intent without a version change. + ## 🛑 Correct operational intent details check If the operational intent details reported by the USS do not match the user's flight intent, this check will fail per **[interuss.automated_testing.flight_planning.ExpectedBehavior](../../../requirements/interuss/automated_testing/flight_planning.md)** and **[astm.f3548.v21.OPIN0025](../../../requirements/astm/f3548/v21.md)**. diff --git a/monitoring/uss_qualifier/scenarios/scenario.py b/monitoring/uss_qualifier/scenarios/scenario.py index 839e49e103..8ffdf0758d 100644 --- a/monitoring/uss_qualifier/scenarios/scenario.py +++ b/monitoring/uss_qualifier/scenarios/scenario.py @@ -5,7 +5,7 @@ from collections.abc import Callable, Iterable from datetime import UTC, datetime, timedelta from enum import Enum -from typing import TypeVar +from typing import Any, TypeVar import arrow from implicitdict import StringBasedDateTime, StringBasedTimeDelta @@ -257,6 +257,9 @@ class GenericTestScenario(ABC): on_failed_check: Callable[[FailedCheck], None] | None = None time_context: TestTimeContext + cache: dict[str, Any] + """Cached data scoped to the lifetime of the scenario.""" + resource_origins: dict[ResourceID, str] """Map between local resource name (as defined in test scenario) to where that resource originated.""" @@ -277,6 +280,7 @@ def __init__(self): self.documentation = get_documentation(self.__class__) self._phase = ScenarioPhase.NotStarted self.time_context = TestTimeContext() + self.cache = {} @staticmethod def make_test_scenario(