tests/test_openenv_wrapper.py

from fastapi.testclient import TestClient

from server.app import app


def test_openenv_wrapper_exposes_schema_and_state():
    client = TestClient(app)
    assert client.get("/schema").status_code == 200
    assert client.get("/state").status_code == 200
    web_response = client.get("/web")
    assert web_response.status_code == 200
    assert "GPU Scheduler Console" in web_response.text
    assert "Operate the benchmark scenario like a compact placement console" in web_response.text


def test_openenv_wrapper_accepts_wrapped_action_payload():
    client = TestClient(app)
    reset_response = client.post("/reset", json={})
    assert reset_response.status_code == 200

    step_response = client.post(
        "/step",
        json={
            "action": {
                "action": "defer",
                "rationale": "Defer placement for a better window.",
                "metadata": {},
            }
        },
    )
    assert step_response.status_code == 200
    body = step_response.json()
    assert body["observation"]["current_tick"] == 1
    assert body["done"] is False
    assert "summary" in body["observation"]
    assert "feasible_assignments" in body["observation"]
    assert "action_hint" in body["observation"]
    assert body["observation"]["feasible_assignments"] == []
    assert body["observation"]["suggested_assignments"] == []


def test_web_reset_exposes_readable_summary_fields():
    client = TestClient(app)
    body = client.post("/web/reset").json()
    observation = body["observation"]
    assert observation["summary"].startswith("Tick 0/")
    assert observation["priority_alerts"] == []
    assert observation["suggested_assignments"] == []
    assert observation["action_hint"] == "Inspect pending jobs and eligible GPUs, then choose a placement or defer."


def test_web_state_contains_dashboard_fields():
    client = TestClient(app)
    client.post("/web/reset", json={})
    body = client.get("/web/state").json()
    assert "action_hint" in body
    assert "suggested_assignments" in body
    assert "pending_jobs" in body
    assert body["suggested_assignments"] == []


def test_web_state_matches_web_reset_and_step_session():
    client = TestClient(app)
    client.post("/web/reset", json={"scenario_id": "deadline_crunch"})
    for _ in range(2):
        response = client.post(
            "/web/step",
            json={
                "action": {
                    "action": "defer",
                    "rationale": "Inspect the current pending queue first.",
                    "metadata": {},
                }
            },
        )
        assert response.status_code == 200

    body = client.get("/web/state").json()
    assert body["scenario_id"] == "deadline_crunch"
    assert body["current_tick"] == 2
    assert body["summary"].startswith("Tick 2/")