parking_solution/apps/trainer/tests/test_server.py

"""The job API: readiness, one job at a time, subprocess jobs with persisted logs, versions."""

from __future__ import annotations

import json
import threading
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer
from pathlib import Path

import pytest

from trainer.server import Handler, Jobs, readiness, versions, wait_idle


@pytest.fixture
def api(collector_dir: Path, tmp_path: Path):  # type: ignore[no-untyped-def]
    out = tmp_path / "out"
    Handler.jobs = Jobs(collector_dir, out, "https://example.invalid/pkg", "tok")
    httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    t = threading.Thread(target=httpd.serve_forever, daemon=True)
    t.start()
    base = f"http://127.0.0.1:{httpd.server_address[1]}"

    def call(method: str, path: str, body: dict | None = None):  # type: ignore[no-untyped-def]
        req = urllib.request.Request(base + path, method=method)
        data = None
        if body is not None:
            data = json.dumps(body).encode()
            req.add_header("Content-Type", "application/json")
        try:
            with urllib.request.urlopen(req, data=data, timeout=10) as r:
                raw = r.read()
                return r.status, (
                    json.loads(raw) if r.headers.get_content_type() == "application/json" else raw.decode()
                )
        except urllib.error.HTTPError as e:
            return e.code, json.loads(e.read() or b"{}")

    yield call, out
    httpd.shutdown()
    httpd.server_close()


def test_readiness_and_empty_versions(api) -> None:  # type: ignore[no-untyped-def]
    call, _ = api
    code, r = call("GET", "/readiness")
    assert code == 200 and r["ready"] is True and r["run"]["classes"] == ["sedan", "suv", "van"]
    assert r["defaults"]["minAccuracy"] == 0.85 and "finetune" in r["modes"]
    assert call("GET", "/versions") == (200, {"versions": []})
    assert call("GET", "/health")[1]["busy"] is False
    assert readiness(Path("/nonexistent"))["ready"] is False


def test_evaluate_job_runs_as_a_subprocess_and_is_recorded(api) -> None:  # type: ignore[no-untyped-def]
    call, out = api
    code, job = call("POST", "/jobs", {"kind": "evaluate", "version": "nope"})
    assert code == 202 and job["status"] == "running" and job["kind"] == "evaluate"
    wait_idle(Handler.jobs)
    code, j = call("GET", f"/jobs/{job['id']}")
    assert code == 200 and j["status"] == "failed" and j["exitCode"] == 1
    assert "evaluate --data" in j["log"] and "nope" in j["log"]
    assert (out / "jobs" / f"{job['id']}.json").is_file() and (out / "jobs" / f"{job['id']}.log").is_file()
    code, lst = call("GET", "/jobs")
    assert code == 200 and lst["jobs"][0]["id"] == job["id"] and lst["current"] is None


def test_bad_requests(api) -> None:  # type: ignore[no-untyped-def]
    call, _ = api
    assert call("POST", "/jobs", {"kind": "nuke"})[0] == 400
    assert call("POST", "/jobs", {"kind": "train", "mode": "magic"})[0] == 400
    assert call("POST", "/jobs", {"kind": "evaluate", "version": "../etc"})[0] == 400
    assert call("POST", "/jobs", {"kind": "publish", "version": "v1", "url": "ftp://x"})[0] == 400
    assert call("GET", "/versions/../x/report")[0] == 400
    assert call("GET", "/versions/v9/report")[0] == 404
    assert call("GET", "/jobs/nope")[0] == 404
    assert call("GET", "/nothing")[0] == 404


def test_train_job_then_versions_and_report(api) -> None:  # type: ignore[no-untyped-def]
    pytest.importorskip("torch")
    call, out = api
    body = {"kind": "train", "mode": "features", "minAccuracy": 0.0, "epochs": 100, "version": "vapi"}
    # The test-only flags are not offered by the API; inject them via the CLI args the runner builds.
    orig = Jobs._argv

    def patched(self, kind, a):  # type: ignore[no-untyped-def]
        argv = orig(self, kind, a)
        return argv + ["--no-pretrained", "--input-size", "64", "--no-cache"] if kind == "train" else argv

    Jobs._argv = patched  # type: ignore[method-assign]
    try:
        code, job = call("POST", "/jobs", body)
        assert code == 202
        assert call("POST", "/jobs", {"kind": "evaluate", "version": "vapi"})[0] == 409  # one at a time
        wait_idle(Handler.jobs, 120)
    finally:
        Jobs._argv = orig  # type: ignore[method-assign]
    code, j = call("GET", f"/jobs/{job['id']}")
    assert j["status"] == "done" and "MODEL WRITTEN" in j["log"]
    code, v = call("GET", "/versions")
    assert code == 200 and v["versions"][0]["version"] == "vapi" and v["versions"][0]["written"] is True
    assert v["versions"][0]["classes"] == ["sedan", "suv", "van"] and v["versions"][0]["accuracy"] >= 0.9
    code, report = call("GET", "/versions/vapi/report")
    assert code == 200 and report.startswith("# Body-type classifier vapi")
    assert versions(out)[0]["floor"] == 0.0
    # evaluate on the written model now succeeds
    code, job2 = call("POST", "/jobs", {"kind": "evaluate", "version": "vapi"})
    wait_idle(Handler.jobs)
    assert call("GET", f"/jobs/{job2['id']}")[1]["status"] == "done"