| """Tests of the pending run's phase: the rule table of `run_phase` over the run block and the submission entries of |
| the run's `status.json`, the pending block `GET /runs` and `GET /me?run=<pending run>` serve with `intake_done` and |
| `phase`, the published runs' blocks without either, and one read of `status.json` per request.""" |
| from __future__ import annotations |
|
|
| import json |
| from pathlib import Path |
|
|
| import httpx |
| import pytest |
| from fastapi.testclient import TestClient |
| from huggingface_hub.errors import HfHubHTTPError |
|
|
| import app as leaderboard_api |
| from conftest import PENDING_RUN, RUN_BLOCKS, UNPULLED_RUN, settings, write |
|
|
| RUN = {**PENDING_RUN, "intake_done": True} |
| STATUS_PATH = "week_03/status.json" |
|
|
|
|
| def entries(org: str, *codes) -> dict: |
| """`status.json` entries of one organization's submission, one per model of `PENDING_RUN` with the given codes.""" |
| return {f"submission-{org}/{model.replace('/', '__')}": {"kind": "submission", "status": code} |
| for model, code in zip(PENDING_RUN["models"], codes)} |
|
|
|
|
| def bearer(token: str) -> dict: |
| return {"Authorization": f"Bearer {token}"} |
|
|
|
|
| class CountingArchive(leaderboard_api.DirectoryArchive): |
| """The fixture archive directory, counting the reads of each path, with the reads of `failing` raising an error.""" |
|
|
| reads: dict[str, int] = {} |
| failing: str | None = None |
|
|
| def read(self, path: str) -> bytes | None: |
| CountingArchive.reads[path] = CountingArchive.reads.get(path, 0) + 1 |
| if path == CountingArchive.failing: |
| response = httpx.Response(503, request=httpx.Request("GET", "https://huggingface.co/api/datasets/x")) |
| raise HfHubHTTPError("the archive cannot be read", response=response) |
| return super().read(path) |
|
|
|
|
| @pytest.fixture() |
| def counting(archive: Path, userinfo_url: str, monkeypatch) -> TestClient: |
| """A client over the fixture archive with `RUN` as the pending block, reading through `CountingArchive`.""" |
| CountingArchive.reads, CountingArchive.failing = {}, None |
| monkeypatch.setattr(leaderboard_api, "DirectoryArchive", CountingArchive) |
| write(archive, "leaderboard/pending.json", {"schema": 1, "run": RUN}) |
| return TestClient(leaderboard_api.create_app(settings(archive, userinfo_url))) |
|
|
|
|
| @pytest.mark.parametrize(("run", "status", "phase"), [ |
| (UNPULLED_RUN, entries("team-a", "in_progress", "completed"), "upcoming"), |
| (RUN, None, "checked"), |
| ({**RUN, "intake_done": False}, None, "pulled"), |
| (RUN, {**entries("team-a", "not_started", "not_started"), **entries("team-b", "not_started")}, "checked"), |
| ({**RUN, "intake_done": False}, entries("team-a", "not_started", "not_started"), "pulled"), |
| (RUN, {**entries("team-a", "in_progress", "not_started"), **entries("team-b", "not_started")}, "evaluating"), |
| ({**RUN, "intake_done": False}, entries("team-a", "completed", "not_started"), "evaluating"), |
| (RUN, {**entries("team-a", "completed", "error"), **entries("team-b", "completed")}, "finishing"), |
| (RUN, {**entries("team-a", "completed", "completed"), **entries("team-b", "in_progress")}, "evaluating"), |
| (RUN, {"elicit/m__one": {"status": "in_progress"}, "unsteered/m__one": {"status": "completed"}}, "checked"), |
| ({**RUN, "intake_done": False}, {"unsteered/m__one": {"status": "in_progress"}}, "pulled"), |
| (RUN, {"submission-team-a/m__one": "in_progress", "submission-team-a/m__two": {"status": "running"}}, "checked"), |
| (RUN, {"submission-team-a/m__one": {"status": "completed"}, "submission-team-a/m__two": {"status": None}}, |
| "evaluating"), |
| (RUN, ["not", "a", "mapping"], "checked"), |
| (PENDING_RUN, None, "pulled"), |
| ({**RUN, "intake_done": "true"}, entries("team-a", "not_started"), "pulled"), |
| ]) |
| def test_run_phase_follows_the_rule_table(run: dict, status, phase: str) -> None: |
| assert leaderboard_api.run_phase(run, status) == phase and phase in leaderboard_api.PHASES |
|
|
|
|
| def test_pending_block_adds_the_phase_and_reads_intake_done_as_a_boolean() -> None: |
| block = leaderboard_api.pending_block(RUN, entries("team-a", "in_progress")) |
| assert block == {**RUN, "phase": "evaluating"} and list(block)[-1] == "phase" |
| for value in (None, "true", 1): |
| assert leaderboard_api.pending_block({**RUN, "intake_done": value}, None)["intake_done"] is False |
| assert leaderboard_api.pending_block(PENDING_RUN, None) == {**PENDING_RUN, "intake_done": False, |
| "phase": "pulled"} |
|
|
|
|
| def test_runs_gives_the_phase_from_the_status_file(counting: TestClient, archive: Path) -> None: |
| """The fixture's status file has a submission entry in progress; the phase follows the file across the week, |
| each request reads it once, and the published runs' entries get no phase.""" |
| payload = counting.get("/runs").json() |
| assert payload["pending"] == {**RUN, "phase": "evaluating"} and CountingArchive.reads[STATUS_PATH] == 1 |
| assert all("phase" not in entry and "intake_done" not in entry for entry in payload["runs"]) |
| for status, phase in (({**entries("team-a", "not_started", "not_started")}, "checked"), |
| ({**entries("team-a", "completed", "error"), **entries("team-b", "completed")}, |
| "finishing")): |
| write(archive, STATUS_PATH, status) |
| assert counting.get("/runs").json()["pending"]["phase"] == phase |
| (archive / STATUS_PATH).unlink() |
| assert counting.get("/runs").json()["pending"]["phase"] == "checked" |
| write(archive, "leaderboard/pending.json", {"schema": 1, "run": {**RUN, "intake_done": False}}) |
| assert counting.get("/runs").json()["pending"] == {**RUN, "intake_done": False, "phase": "pulled"} |
|
|
|
|
| def test_runs_reads_no_status_file_before_the_pull(counting: TestClient, archive: Path) -> None: |
| write(archive, "leaderboard/pending.json", {"schema": 1, "run": UNPULLED_RUN}) |
| assert counting.get("/runs").json()["pending"] == {**UNPULLED_RUN, "intake_done": False, "phase": "upcoming"} |
| assert STATUS_PATH not in CountingArchive.reads |
|
|
|
|
| def test_runs_answers_503_when_the_status_file_cannot_be_read(counting: TestClient) -> None: |
| CountingArchive.failing = STATUS_PATH |
| response = counting.get("/runs") |
| assert response.status_code == 503 and response.json() == {"error": "archive_unavailable"} |
|
|
|
|
| @pytest.mark.parametrize("token", ["participant-token", "organizer-token"]) |
| def test_me_gives_the_pending_block_with_its_phase_and_reads_the_status_once(counting: TestClient, archive: Path, |
| token: str) -> None: |
| status = json.loads((archive / STATUS_PATH).read_text(encoding="utf-8")) |
| CountingArchive.reads.clear() |
| payload = counting.get("/me", params={"run": "week_03"}, headers=bearer(token)).json() |
| assert payload["pending"] is True and payload["run"] == {**RUN, "phase": "evaluating"} |
| assert CountingArchive.reads[STATUS_PATH] == 1 |
| teams = payload["teams"] if token == "participant-token" else list(payload["teams"].values()) |
| assert teams and all(team["run"] == PENDING_RUN for team in teams) |
| for team in teams: |
| assert team["evaluation"] == leaderboard_api.evaluation(status, team["org"], PENDING_RUN["models"]) |
| published = counting.get("/me", headers=bearer(token)).json() |
| assert published["run"] == RUN_BLOCKS["week_02"] and "pending" not in published |
|
|
|
|
| def test_me_reads_no_status_file_before_the_pull(counting: TestClient, archive: Path) -> None: |
| write(archive, "leaderboard/pending.json", {"schema": 1, "run": UNPULLED_RUN}) |
| for token in ("participant-token", "organizer-token"): |
| payload = counting.get("/me", params={"run": "week_03"}, headers=bearer(token)).json() |
| assert payload["run"] == {**UNPULLED_RUN, "intake_done": False, "phase": "upcoming"} and not payload["teams"] |
| assert STATUS_PATH not in CountingArchive.reads |
|
|
