| """Tests for lightweight Hub reads used by the Space UI.""" |
|
|
| from __future__ import annotations |
|
|
| from pathlib import Path |
|
|
| import pandas as pd |
| import pytest |
|
|
| from src import hub_io |
|
|
|
|
| @pytest.fixture(autouse=True) |
| def _reset_leaderboard_cache(monkeypatch: pytest.MonkeyPatch) -> None: |
| monkeypatch.setenv("RESULTS_REPO", "InteligentDD/molgenbench-results") |
| monkeypatch.setenv("LEADERBOARD_CACHE_TTL_SECONDS", "60") |
| monkeypatch.setenv("HF_HUB_ETAG_TIMEOUT", "3") |
| monkeypatch.setattr(hub_io, "_LEADERBOARD_CACHE_FRAME", None) |
| monkeypatch.setattr(hub_io, "_LEADERBOARD_CACHE_UPDATED_AT", 0.0) |
|
|
|
|
| def _write_parquet(path: Path, source_type: str = "online_submission") -> Path: |
| pd.DataFrame( |
| [ |
| { |
| "source_type": source_type, |
| "target_split": "all", |
| "submission_id": "sub-1", |
| "model_name": "Model", |
| "task": "denovo", |
| "evaluation_level": "2d", |
| } |
| ] |
| ).to_parquet(path) |
| return path |
|
|
|
|
| def test_load_leaderboard_uses_memory_cache_for_repeated_ui_reads( |
| tmp_path: Path, |
| monkeypatch: pytest.MonkeyPatch, |
| ) -> None: |
| """Repeated tab events should not each hit the Hub.""" |
|
|
| parquet_path = _write_parquet(tmp_path / "leaderboard.parquet") |
| baseline_path = _write_parquet(tmp_path / "paper_baselines.parquet", "paper_baseline") |
| calls: list[dict[str, object]] = [] |
|
|
| def fake_download(**kwargs: object) -> str: |
| calls.append(kwargs) |
| if kwargs["filename"] == "paper_baselines.parquet": |
| return str(baseline_path) |
| return str(parquet_path) |
|
|
| monkeypatch.setattr(hub_io, "hf_hub_download", fake_download) |
|
|
| first = hub_io.load_leaderboard() |
| second = hub_io.load_leaderboard() |
|
|
| assert first["submission_id"].tolist() == ["sub-1", "sub-1"] |
| assert second["submission_id"].tolist() == ["sub-1", "sub-1"] |
| assert first["source_type"].tolist() == ["online_submission", "paper_baseline"] |
| assert len(calls) == 2 |
| assert calls[0]["etag_timeout"] == 3.0 |
|
|
|
|
| def test_load_leaderboard_tolerates_missing_paper_baselines( |
| tmp_path: Path, |
| monkeypatch: pytest.MonkeyPatch, |
| ) -> None: |
| """The Space remains compatible while the new baseline file is being deployed.""" |
|
|
| parquet_path = _write_parquet(tmp_path / "leaderboard.parquet") |
|
|
| def fake_download(**kwargs: object) -> str: |
| if kwargs["filename"] == "paper_baselines.parquet": |
| raise FileNotFoundError("not deployed yet") |
| return str(parquet_path) |
|
|
| monkeypatch.setattr(hub_io, "hf_hub_download", fake_download) |
|
|
| frame = hub_io.load_leaderboard() |
|
|
| assert frame["submission_id"].tolist() == ["sub-1"] |
| assert frame["source_type"].tolist() == ["online_submission"] |
|
|
|
|
| def test_load_leaderboard_falls_back_to_existing_cache_when_hub_is_slow( |
| monkeypatch: pytest.MonkeyPatch, |
| ) -> None: |
| """A stale cache is still better than freezing the UI on a failed Hub read.""" |
|
|
| cached = pd.DataFrame([{"submission_id": "cached", "task": "denovo"}]) |
| monkeypatch.setattr(hub_io, "_LEADERBOARD_CACHE_FRAME", cached) |
| monkeypatch.setattr(hub_io, "_LEADERBOARD_CACHE_UPDATED_AT", 0.0) |
| monkeypatch.setenv("LEADERBOARD_CACHE_TTL_SECONDS", "0") |
|
|
| def failing_download(**_: object) -> str: |
| raise TimeoutError("hub timeout") |
|
|
| monkeypatch.setattr(hub_io, "hf_hub_download", failing_download) |
|
|
| frame = hub_io.load_leaderboard() |
|
|
| assert frame["submission_id"].tolist() == ["cached"] |
|
|
