wip: adapt fuzzer to stdcm v2

Author: eckter

tests/fuzzer/fuzzer_v2.py

@@ -1,13 +1,14 @@
 import datetime
 import json
+import math
 import random
 import time
 from collections import defaultdict
 from dataclasses import dataclass, field
 from enum import Enum, IntEnum
 from functools import cache
 from pathlib import Path
-from typing import Dict, Iterable, List, Optional, Set, Tuple
+from typing import Dict, Iterable, List, Optional, Tuple
 
 import requests
 from osrd_schemas.switch_type import builtin_node_types
@@ -119,9 +120,9 @@ def run_test(
         if rolling_stock_name is None
         else get_rolling_stock(editoast_url, rolling_stock_name)
     )
-    path, path_length = make_valid_path(infra)
-    if random.randint(0, 1) == 0:
-        test_new_train(editoast_url, scenario, rolling_stock.name, path_length, infra_name, path, prelude)
+    path = make_valid_path(infra)
+    if random.randint(0, 0) == 1:
+        test_new_train(editoast_url, scenario, rolling_stock.name, infra_name, path, prelude)
     else:
         test_stdcm(editoast_url, scenario, rolling_stock.id, infra_name, path, prelude)
 
@@ -130,7 +131,6 @@ def test_new_train(
     editoast_url: str,
     scenario: Scenario,
     rolling_stock: str,
-    path_length: float,
     infra_name: str,
     path: List[Tuple[str, float]],
     prelude: List,
@@ -179,8 +179,11 @@ def test_stdcm(
     Not finding a path isn't considered as an error, we only look for 500 codes here.
     """
     print("testing stdcm")
-    stdcm_payload = make_stdcm_payload(scenario, path, rolling_stock)
-    r = post_with_timeout(editoast_url + "stdcm/", json=stdcm_payload)
+    stdcm_payload = make_stdcm_payload(path, rolling_stock)
+    r = post_with_timeout(
+        editoast_url + f"v2/timetable/{scenario.timetable}/stdcm/?infra={scenario.infra}",
+        json=stdcm_payload
+    )
     if r.status_code // 100 != 2:
         content = r.content.decode("utf-8")
         if r.status_code // 100 == 4 and "No path could be found" in content:
@@ -190,29 +193,24 @@ def test_stdcm(
     print("test PASSED")
 
 
-def make_stdcm_payload(scenario: Scenario, path: List[Tuple[str, float]], rolling_stock: int) -> Dict:
+def make_stdcm_payload(path: List[Tuple[str, float]], rolling_stock: int) -> Dict:
     """
     Creates a payload for an STDCM request
     """
     res = {
-        "infra_id": scenario.infra,
         "rolling_stock_id": rolling_stock,
-        "timetable_id": scenario.timetable,
-        "start_time": random.randint(0, 3600 * 24),
-        "maximum_departure_delay": random.randint(0, 3600 * 4),
-        "maximum_run_time": random.randint(3600 * 5, 3600 * 10),
-        "margin_before": random.randint(0, 600),
-        "margin_after": random.randint(0, 600),
-        "steps": [convert_stop(stop) for stop in path],
+        "start_time": make_random_time(),
+        "maximum_departure_delay": random.randint(0, 3_600_000 * 4),
+        "maximum_run_time": random.randint(3_600_000 * 5, 3_600_000 * 10),
+        "time_gap_before": random.randint(0, 600_000),
+        "time_gap_after": random.randint(0, 600_000),
+        "steps": [convert_stop_stdcm(stop) for stop in path],
         "comfort": "STANDARD",
-        "standard_allowance": {
-            "value_type": "percentage",
-            "percentage": 0,
-        },
+        "margin": None,
     }
     res["steps"][-1]["duration"] = 1  # Force a stop at the end
-    allowance_value = make_random_allowance_value(0)
-    if allowance_value["value_type"] != "time" and random.randint(0, 2) == 0:
+    allowance_value = make_random_margin_value()
+    if random.randint(0, 2) == 0:
         res["standard_allowance"] = allowance_value
     return res
 
@@ -298,15 +296,6 @@ def get_random_rolling_stock(editoast_url: str) -> RollingStock:
     return RollingStock(rolling_stock["name"], rolling_stock["id"])
 
 
-def format_route_node(waypoint_id, direction):
-    """
-    Formats a waypoint + track direction into a string, to be used in dicts
-    :param waypoint_id: waypoint id
-    :param direction: direction on the track section the waypoint is placed on
-    """
-    return f"{waypoint_id};{direction}"
-
-
 def make_graph(editoast_url: str, infra: int) -> InfraGraph:
     """
     Makes a graph from the infra
@@ -338,20 +327,6 @@ def random_set_element(s: Iterable):
     return random.choice(list(s))
 
 
-def check_tracks_are_unseen(seen_track_sections: Set[str], route: Dict):
-    """
-    Checks all tracks in the route, returns true if a track is already part of the path, adds them otherwise
-    :param seen_track_sections: Set of track sections on the path
-    :param route: Route to check
-    """
-    for track_range in route["path"]:
-        track_id = track_range["track"]
-        if track_id in seen_track_sections:
-            return False
-        seen_track_sections.add(track_id)
-    return True
-
-
 def make_steps_on_track(track: Dict, endpoint: Endpoint, number: float) -> List[float]:
     """
     Generates a random list of steps on a route
@@ -420,7 +395,7 @@ def make_path(infra: InfraGraph) -> Tuple[List[Tuple[str, float]], float]:
     return res, total_path_length
 
 
-def make_valid_path(infra: InfraGraph) -> Tuple[List[Tuple[str, float]], float]:
+def make_valid_path(infra: InfraGraph) -> List[Tuple[str, float]]:
     """
     Generates a path with at least two steps
     :param infra: infra graph
@@ -429,29 +404,30 @@ def make_valid_path(infra: InfraGraph) -> Tuple[List[Tuple[str, float]], float]:
     while True:
         path, path_length = make_path(infra)
         if len(path) > 1 and path_length > 0:
-            return path, path_length
+            return path
 
 
-def convert_stop(stop: Tuple[str, float]) -> Dict:
+def convert_stop_stdcm(stop: Tuple[str, float]) -> Dict:
     """
-    Converts a stop to be in the schedule payload
+    Converts a stop to be in the stdcm payload
     :param stop: (track, offset)
     """
     track_section, offset = stop
-    duration = 0 if random.randint(0, 1) == 0 else random.random() * 1000
+    duration = None if random.randint(0, 1) == 0 else to_ms(random.random() * 1_000)
     return {
         "duration": duration,
-        "waypoints": [{"track_section": track_section, "offset": offset}],
+        "location": {"track": track_section, "offset": to_mm(offset)},
     }
 
 
-def convert_stop_v2(stop: Tuple[str, float], i: int) -> Dict:
+def convert_stop(stop: Tuple[str, float], i: int) -> Dict:
     """
     Converts a stop to be in the schedule payload
     :param stop: (track, offset)
+    :param i: index of the stop, included as an id
     """
     track_section, offset = stop
-    return {"offset": round(offset * 1_000), "track": track_section, "id": str(i)}
+    return {"offset": to_mm(offset), "track": track_section, "id": str(i)}
 
 
 def create_scenario(editoast_url: str, infra_id: int) -> Scenario:
@@ -517,7 +493,7 @@ def reset_scenario(editoast_url: str, scenario: Scenario) -> Scenario:
     return Scenario(project_id, study_id, new_id, infra_id, new_timetable_id)
 
 
-def make_random_allowance_value_v2() -> str:
+def make_random_margin_value() -> str:
     if random.randint(0, 3) == 0:
         return f"{random.randint(3, 20)}%"
     if random.randint(0, 3) == 0:
@@ -534,7 +510,7 @@ def make_random_margins(n_steps: int) -> Dict:
             i += 2  # Avoids constraints on very short ranges
     return {
         "boundaries": transitions,
-        "values": [make_random_allowance_value_v2() for _ in range(len(transitions) + 1)],
+        "values": [make_random_margin_value() for _ in range(len(transitions) + 1)],
     }
 
 
@@ -551,7 +527,7 @@ def make_payload_schedule(
     return [
         {
             "comfort": "STANDARD",
-            "path": [convert_stop_v2(stop, i) for i, stop in enumerate(path)],
+            "path": [convert_stop(stop, i) for i, stop in enumerate(path)],
             "initial_speed": 0,
             "labels": [],
             "constraint_distribution": random.choice(["STANDARD", "MARECO"]),
@@ -566,24 +542,10 @@ def make_payload_schedule(
     ]
 
 
-def make_random_allowance_value(allowance_length) -> Dict:
-    if random.randint(0, 3) == 0:
-        return {
-            "value_type": "percentage",
-            "percentage": random.randint(3, 20) + random.random(),
-        }
-    if random.randint(0, 3) == 0:
-        return {
-            "value_type": "time_per_distance",
-            "minutes": random.randint(3, 7) + random.random(),
-        }
-    return {
-        "value_type": "time",
-        "seconds": (random.randint(3, 7) + random.random()) * 60 * allowance_length / 100000,
-    }
-
-
 def make_random_time():
+    """
+    Generate a random datetime. All values will be within the same 24h
+    """
     start = datetime.datetime(year=2024, month=1, day=1, tzinfo=datetime.timezone.utc)  # Arbitrary date
     date = start + datetime.timedelta(seconds=(random.randint(0, 3600 * 24)))
     return date.isoformat()
@@ -641,6 +603,13 @@ def get_infra(editoast_url: str, infra_name: str) -> int:
     raise ValueError(f"Unable to find infra {infra_name}")
 
 
+def to_mm(distance: float) -> int:
+    return math.floor(distance * 1_000)
+
+
+to_ms = to_mm
+
+
 if __name__ == "__main__":
     infra_id = get_infra(EDITOAST_URL, INFRA_NAME)
     new_scenario = create_scenario(EDITOAST_URL, infra_id)
@@ -653,7 +622,7 @@ def get_infra(editoast_url: str, infra_name: str) -> int:
         EDITOAST_URL,
         new_scenario,
         scenario_ttl=20,
-        n_test=1000,
+        n_test=10,
         log_folder=Path(__file__).parent / "errors",
         infra_name=INFRA_NAME,
         rolling_stock_name=ROLLING_STOCK_NAME,