Spaces:
Running
Running
| """Synchronisation de la file sky-trajectories avec le Hub : envoi, archivage, échecs sans perte, secrets protégés.""" | |
| import json | |
| import pytest | |
| from fastapi import FastAPI | |
| from fastapi.testclient import TestClient | |
| from mouh.geometry import sync | |
| from mouh.geometry.api import make_router | |
| from mouh.geometry.mesh import STAGING_NAME, export_trajectory | |
| CUBE_V = [(0, 0, 0), (1, 0, 0), (1, 1, 0), (0, 1, 0), (0, 0, 1), (1, 0, 1), (1, 1, 1), (0, 1, 1)] | |
| CUBE_F = [(0, 2, 1), (0, 3, 2), (4, 5, 6), (4, 6, 7), (0, 1, 5), (0, 5, 4), | |
| (3, 6, 2), (3, 7, 6), (0, 4, 7), (0, 7, 3), (1, 2, 6), (1, 6, 5)] | |
| HEADERS = {"x-admin-key": "secret"} | |
| SECRET = "hf_TEST_SECRET_NE_PAS_AFFICHER" | |
| class FakeApi: | |
| def __init__(self, fail: Exception | None = None): | |
| self.fail = fail | |
| self.created = [] | |
| self.uploads = [] | |
| def create_repo(self, **kwargs): | |
| self.created.append(kwargs) | |
| def upload_file(self, **kwargs): | |
| if self.fail is not None: | |
| raise self.fail | |
| self.uploads.append(kwargs) | |
| def queue_with_one_geometry(tmp_path): | |
| queue = tmp_path / "sky-trajectories" | |
| rep = export_trajectory({"vertices": CUBE_V, "faces": CUBE_F, "instruction": "Cube unitaire"}, queue) | |
| assert rep.ok and rep.stats["written"] | |
| return queue | |
| def test_empty_queue_sends_nothing(tmp_path): | |
| api = FakeApi() | |
| result = sync.push_pending(tmp_path / "vide", api, "RDS777/sky-trajectories") | |
| assert result["sent"] == 0 and api.uploads == [] and api.created == [] | |
| def test_successful_push_creates_a_private_dataset_and_archives_the_lines(tmp_path): | |
| queue = queue_with_one_geometry(tmp_path) | |
| api = FakeApi() | |
| result = sync.push_pending(queue, api, "RDS777/sky-trajectories") | |
| assert result["sent"] == 1 and result["pending"] == 0 | |
| assert api.created[0]["private"] is True and api.created[0]["repo_type"] == "dataset" | |
| assert api.created[0]["exist_ok"] is True | |
| upload = api.uploads[0] | |
| assert upload["path_in_repo"].startswith("geometry/") and upload["path_in_repo"].endswith(".jsonl") | |
| sent_line = json.loads(upload["path_or_fileobj"].decode("utf-8").splitlines()[0]) | |
| assert sent_line["instruction"] == "Cube unitaire" and len(sent_line["id"]) == 64 | |
| assert (queue / STAGING_NAME).read_text(encoding="utf-8") == "" | |
| assert sync.sent_batches(queue) == 1 | |
| def test_a_failed_upload_leaves_the_queue_untouched(tmp_path): | |
| queue = queue_with_one_geometry(tmp_path) | |
| before = (queue / STAGING_NAME).read_text(encoding="utf-8") | |
| with pytest.raises(RuntimeError): | |
| sync.push_pending(queue, FakeApi(fail=RuntimeError("réseau coupé")), "RDS777/sky-trajectories") | |
| assert (queue / STAGING_NAME).read_text(encoding="utf-8") == before | |
| assert sync.sent_batches(queue) == 0 | |
| def test_dataset_name_comes_from_the_environment_with_a_default(monkeypatch): | |
| monkeypatch.delenv("SKY_DATASET_REPO", raising=False) | |
| assert sync.dataset_repo() == "RDS777/sky-trajectories" | |
| monkeypatch.setenv("SKY_DATASET_REPO", "autre/depot") | |
| assert sync.dataset_repo() == "autre/depot" | |
| def make_client(tmp_path, admin="secret", sync_fn=None): | |
| app = FastAPI() | |
| app.include_router(make_router(tmp_path / "q", lambda key: key == admin, sync=sync_fn)) | |
| return TestClient(app) | |
| def test_sync_route_requires_the_admin_key(tmp_path): | |
| client = make_client(tmp_path, sync_fn=lambda: {"sent": 0}) | |
| assert client.post("/api/symbioza/sync").status_code == 401 | |
| assert client.post("/api/symbioza/sync", headers={"x-admin-key": "faux"}).status_code == 401 | |
| def test_sync_route_reports_503_when_no_token_is_configured(tmp_path): | |
| def unavailable(): | |
| raise sync.SyncUnavailable("HF_TOKEN absent : synchronisation désactivée") | |
| client = make_client(tmp_path, sync_fn=unavailable) | |
| r = client.post("/api/symbioza/sync", headers=HEADERS) | |
| assert r.status_code == 503 and "HF_TOKEN absent" in r.json()["error"] | |
| def test_sync_route_never_echoes_a_secret_from_an_upstream_error(tmp_path): | |
| def leaky(): | |
| raise RuntimeError(f"401 Unauthorized pour {SECRET}") | |
| client = make_client(tmp_path, sync_fn=leaky) | |
| r = client.post("/api/symbioza/sync", headers=HEADERS) | |
| assert r.status_code == 502 | |
| assert SECRET not in r.text | |
| assert r.json()["type"] == "RuntimeError" | |
| def test_sync_route_without_a_configured_callable_is_503(tmp_path): | |
| client = make_client(tmp_path, sync_fn=None) | |
| assert client.post("/api/symbioza/sync", headers=HEADERS).status_code == 503 | |
| def test_status_route_counts_without_exposing_content(tmp_path): | |
| client = make_client(tmp_path) | |
| r = client.get("/api/symbioza/status") | |
| assert r.status_code == 200 and r.json() == {"pending": 0, "sent_batches": 0} | |
| def scaled_cube(k): | |
| return [(x * k, y * k, z * k) for x, y, z in CUBE_V] | |
| def export_named(queue, k, name): | |
| rep = export_trajectory({"vertices": scaled_cube(k), "faces": CUBE_F, "instruction": name}, queue) | |
| assert rep.ok and rep.stats["written"] | |
| return rep | |
| def test_geometries_route_is_empty_on_an_empty_queue(tmp_path): | |
| client = make_client(tmp_path) | |
| assert client.get("/api/symbioza/geometries").json() == {"geometries": []} | |
| def test_geometries_route_lists_newest_first_within_the_limit(tmp_path): | |
| queue = tmp_path / "q" | |
| for k, name in ((1, "A"), (2, "B"), (3, "C")): | |
| export_named(queue, k, name) | |
| client = make_client(tmp_path) | |
| got = client.get("/api/symbioza/geometries", params={"limit": 2}).json()["geometries"] | |
| assert [g["instruction"] for g in got] == ["C", "B"] | |
| assert got[0]["checks"]["self_intersections"] == 0 and len(got[0]["vertices"]) == 8 | |
| def test_geometries_keep_listing_after_a_sync_from_the_archive(tmp_path): | |
| queue = tmp_path / "q" | |
| export_named(queue, 1, "A") | |
| sync.push_pending(queue, FakeApi(), "RDS777/sky-trajectories") | |
| export_named(queue, 2, "B") | |
| assert [g["instruction"] for g in sync.recent_entries(queue)] == ["B", "A"] | |
| def test_recent_entries_ignores_unreadable_lines(tmp_path): | |
| queue = tmp_path / "q" | |
| export_named(queue, 1, "A") | |
| with (queue / STAGING_NAME).open("a", encoding="utf-8") as fh: | |
| fh.write("ceci n'est pas du JSON\n") | |
| assert [g["instruction"] for g in sync.recent_entries(queue)] == ["A"] | |
| def test_recent_entries_stops_at_the_face_budget(tmp_path): | |
| queue = tmp_path / "q" | |
| for k, name in ((1, "A"), (2, "B")): | |
| export_named(queue, k, name) | |
| # Chaque cube compte 12 faces : un budget de 12 n'en laisse passer qu'une seule. | |
| assert [g["instruction"] for g in sync.recent_entries(queue, limit=5, face_budget=12)] == ["B"] | |
| # Un budget inférieur au plus petit maillage ne laisse rien passer. | |
| assert sync.recent_entries(queue, limit=5, face_budget=11) == [] | |
| def test_geometries_route_caps_the_limit(tmp_path): | |
| queue = tmp_path / "q" | |
| for k in range(1, 6): | |
| export_named(queue, k, f"G{k}") | |
| client = make_client(tmp_path) | |
| assert len(client.get("/api/symbioza/geometries", params={"limit": 3}).json()["geometries"]) == 3 | |
| assert len(client.get("/api/symbioza/geometries", params={"limit": 0}).json()["geometries"]) == 1 | |
| assert len(client.get("/api/symbioza/geometries", params={"limit": 999}).json()["geometries"]) == 5 | |