Praxis CI
58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""Scenario loader — YAML → Pydantic Scenario (D-018).
|
|
|
|
Loads a scenario by id from the scenarios/ directory, validates it against the
|
|
Pydantic schema, and returns a typed Scenario object. Used by the pipeline
|
|
(TASK-03-07) and the e2e smoke test.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from server.scenarios.schema import Scenario, ValidationError
|
|
|
|
_DEFAULT_SCENARIOS_DIR = Path(__file__).resolve().parent.parent.parent / "scenarios"
|
|
|
|
|
|
def load(scenario_id: str, scenarios_dir: Path | None = None) -> Scenario:
|
|
"""Load and validate a scenario by id.
|
|
|
|
Args:
|
|
scenario_id: e.g. 'customer_service_refund_ca_v01' (the YAML filename stem).
|
|
scenarios_dir: override the scenarios directory (default: repo /scenarios).
|
|
|
|
Returns:
|
|
A validated Scenario object.
|
|
|
|
Raises:
|
|
FileNotFoundError: if the YAML file doesn't exist.
|
|
ValidationError: if the YAML fails schema validation (typed Pydantic error).
|
|
"""
|
|
base = scenarios_dir or _DEFAULT_SCENARIOS_DIR
|
|
path = base / f"{scenario_id}.yaml"
|
|
if not path.exists():
|
|
# Try the id-with-cs-prefix alias (RESEARCH example used 'cs_refund_ca_v01').
|
|
path = base / f"{scenario_id.replace('cs_', 'customer_service_')}.yaml"
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Scenario YAML not found: {scenario_id} in {base}")
|
|
|
|
with path.open("r", encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
|
|
return Scenario.model_validate(raw)
|
|
|
|
|
|
def load_all(scenarios_dir: Path | None = None) -> list[Scenario]:
|
|
"""Load all scenarios in the directory (for the future scenario library)."""
|
|
base = scenarios_dir or _DEFAULT_SCENARIOS_DIR
|
|
out: list[Scenario] = []
|
|
for p in sorted(base.glob("*.yaml")):
|
|
with p.open("r", encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
out.append(Scenario.model_validate(raw))
|
|
return out
|
|
|
|
|
|
__all__ = ["load", "load_all", "ValidationError"] |