fixup! tests: adapt fuzzer for timetable v2

Author: eckter

tests/fuzzer/fuzzer.py

@@ -27,6 +27,8 @@
 
 """
 Generates random tests, running pathfinding + simulations on random paths.
+DEPRECATED: the requests follow the train schedule v1 format.
+To be deleted with the old version of the endpoints.
 """
 
 

tests/fuzzer/fuzzer_v2.py

@@ -8,7 +8,7 @@
 from enum import Enum, IntEnum
 from functools import cache
 from pathlib import Path
-from typing import Dict, Iterable, List, Optional, Tuple
+from typing import Dict, Iterable, List, Optional, Tuple, TypeVar
 
 import requests
 from osrd_schemas.switch_type import builtin_node_types
@@ -36,51 +36,6 @@
 """
 
 
-class _FailedTest(Exception):
-    pass
-
-
-class _ErrorType(str, Enum):
-    SCHEDULE = "SCHEDULE"
-    RESULT = "RESULT"
-    STDCM = "STDCM"
-
-
-class _Endpoint(IntEnum):
-    BEGIN = 0
-    END = 1
-
-    def opposite(self):
-        return _Endpoint.END if self == _Endpoint.BEGIN else _Endpoint.BEGIN
-
-
-@dataclass(eq=True, frozen=True)
-class _TrackEndpoint:
-    track: str
-    endpoint: _Endpoint
-
-    @staticmethod
-    def from_dict(obj: Dict) -> "_TrackEndpoint":
-        return _TrackEndpoint(obj["track"], _Endpoint._member_map_[obj["endpoint"]])
-
-
-@dataclass
-class _InfraGraph:
-    RJSInfra: Dict
-    tracks: Dict[str, Dict] = field(default_factory=dict)
-    links: Dict[_TrackEndpoint, List[_TrackEndpoint]] = field(default_factory=lambda: defaultdict(list))
-
-    def link(self, a: _TrackEndpoint, b: _TrackEndpoint):
-        self.links[a].append(b)
-        self.links[b].append(a)
-
-
-@dataclass
-class _RollingStock:
-    name: str
-    id: int
-
-
 def run(
     editoast_url: str,
     scenario: Scenario,
@@ -186,6 +141,54 @@ def create_scenario(editoast_url: str, infra_id: int) -> Scenario:
     return Scenario(project_id, study_id, id, infra_id, timetable_id)
 
 
+class _FailedTest(Exception):
+    pass
+
+
+class _ErrorType(str, Enum):
+    SCHEDULE = "SCHEDULE"
+    RESULT = "RESULT"
+    STDCM = "STDCM"
+
+
+class _Endpoint(IntEnum):
+    BEGIN = 0
+    END = 1
+
+    def opposite(self):
+        return _Endpoint.END if self == _Endpoint.BEGIN else _Endpoint.BEGIN
+
+
+@dataclass(eq=True, frozen=True)
+class _TrackEndpoint:
+    track: str
+    endpoint: _Endpoint
+
+    @staticmethod
+    def from_dict(obj: Dict) -> "_TrackEndpoint":
+        return _TrackEndpoint(obj["track"], _Endpoint._member_map_[obj["endpoint"]])
+
+
+@dataclass
+class _InfraGraph:
+    RJSInfra: Dict
+    tracks: Dict[str, Dict] = field(default_factory=dict)
+    links: Dict[_TrackEndpoint, List[_TrackEndpoint]] = field(default_factory=lambda: defaultdict(list))
+
+    def link(self, a: _TrackEndpoint, b: _TrackEndpoint):
+        self.links[a].append(b)
+        self.links[b].append(a)
+
+
+@dataclass
+class _RollingStock:
+    name: str
+    id: int
+
+
+U = TypeVar("U")
+
+
 def _make_error(
     error_type: _ErrorType,
     response: Response,
@@ -373,7 +376,7 @@ def _make_graph(editoast_url: str, infra: int) -> _InfraGraph:
     return graph
 
 
-def _random_set_element(s: Iterable):
+def _random_set_element(s: Iterable[U]) -> U:
     """
     Picks a random element in an iterable
     """

tests/tests/test_regressions.py

@@ -14,10 +14,18 @@
 REGRESSION_TESTS_JSON_FILES = [
     str(json_file.relative_to(REGRESSION_TESTS_DATA_FOLDER))
     for json_file in REGRESSION_TESTS_DATA_FOLDER.rglob("*.json")
-    if json.loads(json_file.read_bytes()).get("timetable_version") != 2
+    if json.loads(json_file.read_text()).get("timetable_version") != 2
 ]
 
 
+"""
+This file runs the regression tests generated by the fuzzer.
+DEPRECATED: the requests follow the train schedule v1 format.
+To be deleted with the old version of the endpoints,
+along with the old test files.
+"""
+
+
 def _load_infra(editoast_url: str, infra_id):
     """
     Send a request to preload the infra, raise a RuntimeError on failure

tests/tests/test_regressions_v2.py

@@ -1,6 +1,6 @@
 import json
 from pathlib import Path
-from typing import List
+from typing import Dict, List, Optional
 
 import pytest
 import requests
@@ -14,11 +14,16 @@
 REGRESSION_TESTS_JSON_FILES = [
     str(json_file.relative_to(REGRESSION_TESTS_DATA_FOLDER))
     for json_file in REGRESSION_TESTS_DATA_FOLDER.rglob("*.json")
-    if json.loads(json_file.read_bytes()).get("timetable_version") == 2
+    if json.loads(json_file.read_text()).get("timetable_version") == 2
 ]
 
 
-def _load_infra(editoast_url: str, infra_id):
+"""
+This file runs the regression tests generated by the fuzzer.
+"""
+
+
+def _load_infra(editoast_url: str, infra_id: int):
     """
     Send a request to preload the infra, raise a RuntimeError on failure
     """
@@ -29,7 +34,7 @@ def _load_infra(editoast_url: str, infra_id):
         raise RuntimeError(f"Infra load {r.status_code}: {r.content}")
 
 
-def _schedule_with_payload(editoast_url: str, payload, accept_400, scenario: Scenario):
+def _schedule_with_payload(editoast_url: str, payload: Dict, accept_400: bool, scenario: Scenario) -> Optional[int]:
     """
     Send a schedule request with the given payload, raises an error if the request failed (unless we accept 400s).
     Returns the schedule id.
@@ -42,7 +47,7 @@ def _schedule_with_payload(editoast_url: str, payload, accept_400, scenario: Sce
     return r.json()[0]["id"]
 
 
-def _stdcm_with_payload(editoast_url: str, payload, scenario: Scenario):
+def _stdcm_with_payload(editoast_url: str, payload: Dict, scenario: Scenario):
     """
     Send a stdcm request with the given payload, raises an error if the request failed.
     """