Spaces:
Sleeping
Sleeping
File size: 5,459 Bytes
aa0019f 21cabbe aa0019f 21cabbe aa0019f 21cabbe f12561b 4f1b3fa 21cabbe 4f1b3fa 21cabbe 4f1b3fa 21cabbe fd2d750 21cabbe aa0019f 21cabbe aa0019f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | import pytest
from fastapi.testclient import TestClient
from app import app
from environment.models import AvigilanceReward, REWARD_FLOAT_FIELDS
client = TestClient(app)
def _assert_reward_fields_within_open_interval(reward: dict) -> None:
for field_name in REWARD_FLOAT_FIELDS:
value = reward[field_name]
assert 0 < value < 1, f"{field_name} escaped the open interval: {value}"
def _build_action(task_id: str, observation: dict) -> dict:
if task_id == "task1":
return {
"task_id": "task1",
"fto_grade_action": {
"grade": "B",
"total_score": 70,
"risk_flags": [],
"recommended_action": "self_assessment_required",
"justification": "Contract test action from pytest.",
},
}
if task_id == "task2":
incident_ids = [item["incident_id"] for item in observation["incident_batch"]]
return {
"task_id": "task2",
"incident_priority_action": {
"priority_ranking": incident_ids,
"top_3_rationale": "Contract test rationale from pytest.",
"defer_list": incident_ids[3:],
"escalate_immediately": incident_ids[:2],
"pattern_detected": False,
"pattern_description": None,
},
}
incident_ids = [item["incident_id"] for item in observation["incident_queue"]]
fto_ids = [item["fto_id"] for item in observation["fto_audit_queue"]]
assignments = {
"inspector_1": (incident_ids + fto_ids)[:2],
"inspector_2": (incident_ids + fto_ids)[2:4],
}
return {
"task_id": "task3",
"resource_allocation_action": {
"inspector_assignments": assignments,
"deferred_items": (incident_ids + fto_ids)[4:],
"priority_rationale": "Contract test allocation rationale from pytest.",
"predicted_risk_reduction": 0.55,
"abstain": False,
"abstain_reason": None,
},
}
def test_root_serves_space_frontend() -> None:
response = client.get("/")
assert response.status_code == 200
assert "text/html" in response.headers["content-type"]
assert "AvigilanceEnv" in response.text
assert "Reset Episode" in response.text
assert "Avigilance Mission Console" in response.text
assert "task-card" in response.text
assert "<style>" in response.text
assert "<script>" in response.text
def test_root_contains_self_contained_frontend_logic() -> None:
response = client.get("/")
assert response.status_code == 200
assert "pushTimeline" in response.text
assert "defaultAction" in response.text
assert "--accent" in response.text
def test_frontend_fallback_route_serves_space_app() -> None:
response = client.get("/walkthrough", follow_redirects=True)
assert response.status_code == 200
assert "text/html" in response.headers["content-type"]
assert "AvigilanceEnv" in response.text
def test_openenv_endpoints_round_trip() -> None:
reset = client.post("/reset", params={"task_id": "task1", "seed": 42})
assert reset.status_code == 200
observation = reset.json()
assert observation["task_id"] == "task1"
action = {
"task_id": "task1",
"fto_grade_action": {
"grade": "B",
"total_score": 70,
"risk_flags": [],
"recommended_action": "self_assessment_required",
"justification": "Contract test action from pytest.",
},
}
step = client.post("/step", json=action)
assert step.status_code == 200
payload = step.json()
_assert_reward_fields_within_open_interval(payload["reward"])
assert payload["observation"]["task_id"] == "task1"
state = client.get("/state", params={"task_id": "task1"})
assert state.status_code == 200
assert state.json()["task_id"] == "task1"
def test_reward_schema_is_exclusive() -> None:
schema = AvigilanceReward.model_json_schema()
assert schema["properties"]["score"]["exclusiveMinimum"] == 0
assert schema["properties"]["score"]["exclusiveMaximum"] == 1
def test_metadata_points_to_frontend_walkthrough_anchor() -> None:
response = client.get("/metadata")
assert response.status_code == 200
assert response.json()["walkthrough"] == "/#walkthrough"
@pytest.mark.parametrize("task_id", ["task1", "task2", "task3"])
def test_all_task_step_rewards_stay_inside_open_interval(task_id: str) -> None:
reset = client.post("/reset", params={"task_id": task_id, "seed": 42})
assert reset.status_code == 200
action = _build_action(task_id, reset.json())
step = client.post("/step", json=action)
assert step.status_code == 200
_assert_reward_fields_within_open_interval(step.json()["reward"])
@pytest.mark.parametrize(
("task_id", "invalid_payload"),
[
("task1", {"task_id": "task1"}),
("task2", {"task_id": "task2"}),
("task3", {"task_id": "task3"}),
],
)
def test_missing_task_actions_return_open_interval_reward_fields(task_id: str, invalid_payload: dict) -> None:
reset = client.post("/reset", params={"task_id": task_id, "seed": 42})
assert reset.status_code == 200
step = client.post("/step", json=invalid_payload)
assert step.status_code == 200
_assert_reward_fields_within_open_interval(step.json()["reward"]) |