813bd586d6
v0.3 milestone merged to main. Mastery scoring + competency rubrics + verifiable credentials (formative-tier) shipped. 13/13 REQ-IDs covered. Next milestone: v0.4 (operator tier — cohort dashboard + auth + Postgres). ---ci--- project: praxis phase: 2 milestone: v0.3 status: complete milestone_complete: true milestone_merged_to_main: true ---/ci---
301 lines
9.6 KiB
Python
301 lines
9.6 KiB
Python
"""Unit tests for the scenario library (SLICE-02, TASK-02-04)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
from pydantic import ValidationError
|
|
|
|
from server.scenarios.library import (
|
|
CoverageError,
|
|
IndexEntry,
|
|
IndexManifest,
|
|
ScenarioLibrary,
|
|
)
|
|
from server.scenarios.loader import load
|
|
from server.scenarios.schema import RubricMapping, Scenario
|
|
|
|
_REPO_SCENARIOS_DIR = Path(__file__).resolve().parent.parent / "scenarios"
|
|
|
|
|
|
def test_v01_scenario_still_loads():
|
|
s = load("customer_service_refund_ca_v01")
|
|
assert s.id == "cs_refund_ca_v01"
|
|
# SLICE-06 extended v01 with rubric_criteria; the v0.1 backward-compat
|
|
# contract (empty rubric_criteria) is superseded once SLICE-06 lands.
|
|
assert len(s.rubric_criteria) == 4
|
|
assert s.irt_target_p == 0.7
|
|
assert s.version == "1.0.0"
|
|
assert s.generated_from is None
|
|
assert s.intent_hash is None
|
|
assert s.branch_by_id("accept_resolution") is not None
|
|
|
|
|
|
def test_library_loads_index():
|
|
lib = ScenarioLibrary()
|
|
manifest = lib.load()
|
|
assert isinstance(manifest, IndexManifest)
|
|
ids = [e.id for e in manifest.scenarios]
|
|
assert "cs_refund_ca_v01" in ids
|
|
|
|
|
|
def test_list_by_path_customer_service():
|
|
lib = ScenarioLibrary()
|
|
entries = lib.list_by_path("customer_service")
|
|
assert len(entries) >= 1
|
|
assert all(e.id for e in entries)
|
|
s = lib.get(entries[0].id)
|
|
assert s.path == "customer_service"
|
|
|
|
|
|
def test_list_by_difficulty_range():
|
|
lib = ScenarioLibrary()
|
|
entries = lib.list_by_difficulty(1, 2)
|
|
assert all(1 <= e.difficulty <= 2 for e in entries)
|
|
assert any(e.id == "cs_refund_ca_v01" for e in entries)
|
|
none = lib.list_by_difficulty(4, 5)
|
|
assert all(e.difficulty >= 4 for e in none)
|
|
|
|
|
|
def test_get_caches_and_validates():
|
|
lib = ScenarioLibrary()
|
|
s1 = lib.get("cs_refund_ca_v01")
|
|
s2 = lib.get("cs_refund_ca_v01")
|
|
assert s1 is s2
|
|
assert isinstance(s1, Scenario)
|
|
|
|
|
|
def test_get_unknown_id_raises():
|
|
lib = ScenarioLibrary()
|
|
with pytest.raises(KeyError):
|
|
lib.get("does_not_exist")
|
|
|
|
|
|
def test_select_for_theta_returns_closest():
|
|
lib = ScenarioLibrary()
|
|
import math
|
|
target_p = 0.7
|
|
theta = 0.0
|
|
expected_target_b = theta - math.log(target_p / (1.0 - target_p))
|
|
s = lib.select_for_theta(theta, "customer_service", target_p=target_p)
|
|
assert s is not None
|
|
assert s.path == "customer_service"
|
|
entries = lib.list_by_path("customer_service")
|
|
dists = {e.id: abs(float(e.difficulty) - expected_target_b) for e in entries}
|
|
assert s.id == min(dists, key=dists.get)
|
|
|
|
|
|
def test_select_for_theta_empty_path_returns_none():
|
|
lib = ScenarioLibrary()
|
|
assert lib.select_for_theta(0.0, "no_such_path") is None
|
|
|
|
|
|
def test_check_coverage_under_minimum_raises():
|
|
lib = ScenarioLibrary()
|
|
entries = lib.list_by_path("customer_service")
|
|
criterion_counts: dict[str, int] = {}
|
|
for e in entries:
|
|
for cid in e.rubric_criteria:
|
|
criterion_counts[cid] = criterion_counts.get(cid, 0) + 1
|
|
if any(n < ScenarioLibrary.MIN_COVERAGE for n in criterion_counts.values()):
|
|
with pytest.raises(CoverageError):
|
|
lib.check_coverage("customer_service")
|
|
else:
|
|
counts = lib.check_coverage("customer_service")
|
|
assert all(n >= ScenarioLibrary.MIN_COVERAGE for n in counts.values())
|
|
|
|
|
|
def test_check_coverage_passes_with_enough_scenarios(tmp_path: Path):
|
|
scenarios_dir = tmp_path / "scenarios"
|
|
scenarios_dir.mkdir()
|
|
base_scenario = {
|
|
"id": "cs_a",
|
|
"path": "customer_service",
|
|
"market": "CA",
|
|
"language": "en-CA",
|
|
"title": "A",
|
|
"difficulty": 1,
|
|
"failure_mode": "escalates_unresolved",
|
|
"persona": {"voice_id": "v", "character": "Customer (A)"},
|
|
"setup": {"system_prompt": "x", "opening_line": "y"},
|
|
"success_criteria": ["a"],
|
|
"common_mistakes": ["b"],
|
|
"branches": [
|
|
{
|
|
"id": "accept",
|
|
"trigger": {"learner_signals": ["empathy"]},
|
|
"outcome": "success",
|
|
"debrief_focus": "f",
|
|
}
|
|
],
|
|
"debrief": {"model": "deepseek-v4-flash:cloud", "mode": "no_think", "prompt_template": "debrief/default"},
|
|
}
|
|
for i, sid in enumerate(["cs_a", "cs_b"]):
|
|
sc = dict(base_scenario)
|
|
sc["id"] = sid
|
|
sc["title"] = sid
|
|
sc["persona"]["character"] = f"Customer ({sid})"
|
|
with (scenarios_dir / f"{sid}.yaml").open("w") as f:
|
|
yaml.safe_dump(sc, f)
|
|
index = {
|
|
"version": "1.0.0",
|
|
"scenarios": [
|
|
{
|
|
"id": "cs_a",
|
|
"path": "cs_a.yaml",
|
|
"title": "A",
|
|
"difficulty": 1,
|
|
"failure_mode": "escalates_unresolved",
|
|
"rubric_criteria": ["empathy", "resolution"],
|
|
"version": "1.0.0",
|
|
"author": "expert",
|
|
"generated_from": None,
|
|
},
|
|
{
|
|
"id": "cs_b",
|
|
"path": "cs_b.yaml",
|
|
"title": "B",
|
|
"difficulty": 2,
|
|
"failure_mode": "policy_rigid",
|
|
"rubric_criteria": ["empathy", "resolution"],
|
|
"version": "1.0.0",
|
|
"author": "expert",
|
|
"generated_from": None,
|
|
},
|
|
],
|
|
}
|
|
with (scenarios_dir / "index.yaml").open("w") as f:
|
|
yaml.safe_dump(index, f)
|
|
lib = ScenarioLibrary(scenarios_dir=scenarios_dir)
|
|
counts = lib.check_coverage("customer_service")
|
|
assert counts == {"empathy": 2, "resolution": 2}
|
|
|
|
|
|
def test_reject_invalid_semver_in_schema():
|
|
bad = {
|
|
"id": "x",
|
|
"path": "customer_service",
|
|
"market": "CA",
|
|
"title": "T",
|
|
"difficulty": 1,
|
|
"failure_mode": "escalates_unresolved",
|
|
"persona": {"voice_id": "v", "character": "C"},
|
|
"setup": {"system_prompt": "s", "opening_line": "o"},
|
|
"success_criteria": ["a"],
|
|
"common_mistakes": ["b"],
|
|
"branches": [
|
|
{
|
|
"id": "accept",
|
|
"trigger": {"learner_signals": ["empathy"]},
|
|
"outcome": "success",
|
|
"debrief_focus": "f",
|
|
}
|
|
],
|
|
"debrief": {"model": "deepseek-v4-flash:cloud", "mode": "no_think", "prompt_template": "debrief/default"},
|
|
"version": "not-a-semver",
|
|
}
|
|
with pytest.raises(ValidationError):
|
|
Scenario.model_validate(bad)
|
|
|
|
|
|
def test_reject_invalid_semver_in_index_entry():
|
|
with pytest.raises(ValidationError):
|
|
IndexEntry(
|
|
id="x",
|
|
path="x.yaml",
|
|
title="T",
|
|
difficulty=1,
|
|
failure_mode="f",
|
|
rubric_criteria=["empathy"],
|
|
version="1.0",
|
|
)
|
|
|
|
|
|
def test_rubric_mapping_defaults():
|
|
m = RubricMapping(criterion_id="empathy")
|
|
assert m.criterion_id == "empathy"
|
|
assert m.weight is None
|
|
assert m.evidence_required is True
|
|
|
|
|
|
def test_ai_variation_backref_validation(tmp_path: Path):
|
|
scenarios_dir = tmp_path / "scenarios"
|
|
scenarios_dir.mkdir()
|
|
parent = {
|
|
"id": "cs_parent",
|
|
"path": "customer_service",
|
|
"market": "CA",
|
|
"language": "en-CA",
|
|
"title": "Parent",
|
|
"difficulty": 2,
|
|
"failure_mode": "escalates_unresolved",
|
|
"persona": {"voice_id": "v", "character": "Customer (P)"},
|
|
"setup": {"system_prompt": "s", "opening_line": "o"},
|
|
"success_criteria": ["a"],
|
|
"common_mistakes": ["b"],
|
|
"branches": [
|
|
{
|
|
"id": "accept",
|
|
"trigger": {"learner_signals": ["empathy"]},
|
|
"outcome": "success",
|
|
"debrief_focus": "f",
|
|
}
|
|
],
|
|
"debrief": {"model": "deepseek-v4-flash:cloud", "mode": "no_think", "prompt_template": "debrief/default"},
|
|
"version": "1.0.0",
|
|
}
|
|
child = dict(parent)
|
|
child["id"] = "cs_child"
|
|
child["title"] = "Child"
|
|
child["generated_from"] = "cs_parent"
|
|
child["persona"] = {"voice_id": "v", "character": "Customer (C)"}
|
|
with (scenarios_dir / "cs_parent.yaml").open("w") as f:
|
|
yaml.safe_dump(parent, f)
|
|
with (scenarios_dir / "cs_child.yaml").open("w") as f:
|
|
yaml.safe_dump(child, f)
|
|
index = {
|
|
"version": "1.0.0",
|
|
"scenarios": [
|
|
{
|
|
"id": "cs_parent",
|
|
"path": "cs_parent.yaml",
|
|
"title": "Parent",
|
|
"difficulty": 2,
|
|
"failure_mode": "escalates_unresolved",
|
|
"rubric_criteria": [],
|
|
"version": "1.0.0",
|
|
"author": "expert",
|
|
"generated_from": None,
|
|
},
|
|
{
|
|
"id": "cs_child",
|
|
"path": "cs_child.yaml",
|
|
"title": "Child",
|
|
"difficulty": 2,
|
|
"failure_mode": "escalates_unresolved",
|
|
"rubric_criteria": [],
|
|
"version": "1.0.0",
|
|
"author": "ai",
|
|
"generated_from": "cs_parent",
|
|
},
|
|
],
|
|
}
|
|
with (scenarios_dir / "index.yaml").open("w") as f:
|
|
yaml.safe_dump(index, f)
|
|
lib = ScenarioLibrary(scenarios_dir=scenarios_dir)
|
|
parent_s = lib.get("cs_parent")
|
|
child_s = lib.get("cs_child")
|
|
assert parent_s.generated_from is None
|
|
assert child_s.generated_from == "cs_parent"
|
|
child_entry = next(e for e in lib.entries() if e.id == "cs_child")
|
|
assert child_entry.generated_from == "cs_parent"
|
|
ids = {e.id for e in lib.entries()}
|
|
assert child_s.generated_from in ids
|
|
|
|
|
|
def test_index_manifest_default_version():
|
|
m = IndexManifest()
|
|
assert m.version == "1.0.0"
|
|
assert m.scenarios == [] |