| """Unit tests for the admin promote / demote helpers. |
| |
| C5 contract: the admin gate distinguishes in-set from out-of-set |
| users, ``promote_row`` and ``demote_row`` flip the two validation |
| fields on the right row, and a repeat promotion is idempotent. Every |
| ``results.jsonl`` read and write is mocked, so the suite makes zero |
| Hub calls. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import os |
| from types import SimpleNamespace |
|
|
| import pytest |
|
|
| |
| |
| |
| os.environ.setdefault("CADGENBENCH_DISABLE_BOOT_SWEEP", "1") |
|
|
| import admin |
| import submit |
|
|
| SEED_ROWS = [ |
| { |
| "submission_id": "alpha", |
| "validation_status": "unvalidated", |
| "validation_method": None, |
| "aggregate_score": 0.5, |
| }, |
| { |
| "submission_id": "beta", |
| "validation_status": "validated", |
| "validation_method": "code", |
| "aggregate_score": 0.9, |
| }, |
| ] |
|
|
|
|
| def _jsonl(rows: list[dict]) -> str: |
| return "\n".join(json.dumps(r) for r in rows) + "\n" |
|
|
|
|
| def _row(rows: list[dict], submission_id: str) -> dict: |
| return next(r for r in rows if r["submission_id"] == submission_id) |
|
|
|
|
| @pytest.fixture |
| def hub(monkeypatch): |
| """Mock the results.jsonl read + write that ``_hub_rmw_results`` drives. |
| |
| ``state["rows"]`` starts as a copy of :data:`SEED_ROWS` and is |
| replaced by whatever bytes the helper hands to ``upload_file``, |
| re-parsed back into dicts. ``state["uploads"]`` counts the writes |
| so a test can assert how many commits a call produced. |
| """ |
| state: dict = { |
| "rows": [dict(r) for r in SEED_ROWS], |
| "uploads": 0, |
| "deleted_paths": [], |
| } |
|
|
| def fake_download() -> str: |
| return _jsonl(state["rows"]) |
|
|
| def fake_upload(*, path_or_fileobj, **kwargs) -> None: |
| body = ( |
| path_or_fileobj.decode("utf-8") |
| if isinstance(path_or_fileobj, bytes) |
| else path_or_fileobj |
| ) |
| state["rows"] = [ |
| json.loads(line) for line in body.splitlines() if line.strip() |
| ] |
| state["uploads"] += 1 |
|
|
| def fake_delete_file(*, path_in_repo, **kwargs) -> None: |
| state["deleted_paths"].append(path_in_repo) |
|
|
| monkeypatch.setattr(submit, "_download_results_jsonl", fake_download) |
| monkeypatch.setattr(submit._HF_API, "upload_file", fake_upload) |
| monkeypatch.setattr(submit._HF_API, "delete_file", fake_delete_file) |
| return state |
|
|
|
|
| def test_admin_gate_in_set_vs_out_of_set(monkeypatch): |
| """is_admin admits logged-in users in the set, rejects everyone else.""" |
| monkeypatch.setenv("CADGENBENCH_ADMINS", "michaelr27, lvwerra") |
| assert admin.is_admin(SimpleNamespace(username="michaelr27")) is True |
| assert admin.is_admin(SimpleNamespace(username="lvwerra")) is True |
| assert admin.is_admin(SimpleNamespace(username="someone-else")) is False |
| |
| assert admin.is_admin(None) is False |
| |
| monkeypatch.delenv("CADGENBENCH_ADMINS", raising=False) |
| assert admin.is_admin(SimpleNamespace(username="michaelr27")) is False |
|
|
|
|
| def test_promote_happy_path(hub): |
| """Promoting an unvalidated row sets status + method, leaves others alone.""" |
| admin.promote_row("alpha", "manual") |
| promoted = _row(hub["rows"], "alpha") |
| assert promoted["validation_status"] == "validated" |
| assert promoted["validation_method"] == "manual" |
| |
| assert _row(hub["rows"], "beta")["validation_method"] == "code" |
| assert hub["uploads"] == 1 |
|
|
|
|
| def test_demote_happy_path(hub): |
| """Demoting a validated row clears the method and flips status back.""" |
| admin.demote_row("beta") |
| demoted = _row(hub["rows"], "beta") |
| assert demoted["validation_status"] == "unvalidated" |
| assert demoted["validation_method"] is None |
| assert hub["uploads"] == 1 |
|
|
|
|
| def test_promote_idempotent(hub): |
| """Re-promoting an already-validated row lands the same state.""" |
| admin.promote_row("beta", "code") |
| once = dict(_row(hub["rows"], "beta")) |
| assert once["validation_status"] == "validated" |
| assert once["validation_method"] == "code" |
| |
| admin.promote_row("beta", "code") |
| assert _row(hub["rows"], "beta") == once |
|
|
|
|
| def test_promote_rows_bulk(hub): |
| """A bulk promote flips every listed row in one write.""" |
| admin.promote_rows(["alpha", "beta"], "traces") |
| for sid in ("alpha", "beta"): |
| row = _row(hub["rows"], sid) |
| assert row["validation_status"] == "validated" |
| assert row["validation_method"] == "traces" |
| assert hub["uploads"] == 1 |
|
|
|
|
| def test_demote_rows_bulk(hub): |
| """A bulk demote clears method on every listed row in one write.""" |
| admin.demote_rows(["alpha", "beta"]) |
| for sid in ("alpha", "beta"): |
| row = _row(hub["rows"], sid) |
| assert row["validation_status"] == "unvalidated" |
| assert row["validation_method"] is None |
| assert hub["uploads"] == 1 |
|
|
|
|
| def test_promote_rows_missing_id_raises_without_write(hub): |
| """An unknown id aborts the whole batch before any upload.""" |
| with pytest.raises(LookupError): |
| admin.promote_rows(["alpha", "ghost"], "code") |
| assert hub["uploads"] == 0 |
| |
| assert _row(hub["rows"], "alpha")["validation_status"] == "unvalidated" |
|
|
|
|
| def test_empty_selection_raises(hub): |
| """Bulk helpers reject an empty / all-falsy selection.""" |
| for call in ( |
| lambda: admin.promote_rows([], "code"), |
| lambda: admin.demote_rows([None, ""]), |
| lambda: admin.delete_rows([]), |
| ): |
| with pytest.raises(ValueError): |
| call() |
| assert hub["uploads"] == 0 |
|
|
|
|
| def test_delete_rows_removes_rows_and_artifacts(hub): |
| """Delete drops the rows and best-effort removes their artifacts.""" |
| admin.delete_rows(["alpha"]) |
| remaining = {r["submission_id"] for r in hub["rows"]} |
| assert remaining == {"beta"} |
| |
| assert hub["deleted_paths"] == [ |
| "submissions/alpha.zip", |
| "reports/alpha.html", |
| "reports/alpha.json", |
| ] |
| assert hub["uploads"] == 1 |
|
|
