"""Tests for storage-related API readiness endpoints.""" from contextlib import contextmanager from fastapi.testclient import TestClient import atocore.config as config from atocore.main import app def test_sources_endpoint_reports_configured_sources(tmp_data_dir, monkeypatch): vault_dir = tmp_data_dir / "vault-source" drive_dir = tmp_data_dir / "drive-source" vault_dir.mkdir() drive_dir.mkdir() monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir)) monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir)) config.settings = config.Settings() client = TestClient(app) response = client.get("/sources") assert response.status_code == 200 body = response.json() assert body["vault_enabled"] is True assert body["drive_enabled"] is True assert len(body["sources"]) == 2 assert all(source["read_only"] for source in body["sources"]) def test_health_endpoint_exposes_machine_paths_and_source_readiness(tmp_data_dir, monkeypatch): vault_dir = tmp_data_dir / "vault-source" drive_dir = tmp_data_dir / "drive-source" vault_dir.mkdir() drive_dir.mkdir() monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir)) monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir)) config.settings = config.Settings() client = TestClient(app) response = client.get("/health") assert response.status_code == 200 body = response.json() assert body["status"] == "ok" assert body["sources_ready"] is True assert "db_path" in body["machine_paths"] assert "run_dir" in body["machine_paths"] def test_health_endpoint_reports_code_version_from_module(tmp_data_dir): """The /health response must include code_version reflecting atocore.__version__, so deployment drift detection works.""" from atocore import __version__ client = TestClient(app) response = client.get("/health") assert response.status_code == 200 body = response.json() assert body["version"] == __version__ assert body["code_version"] == __version__ def test_health_endpoint_reports_build_metadata_from_env(tmp_data_dir, monkeypatch): """The /health response must include build_sha, build_time, and build_branch from the ATOCORE_BUILD_* env vars, so deploy.sh can detect precise drift via SHA comparison instead of relying on the coarse code_version field. Regression test for the codex finding from 2026-04-08: code_version 0.2.0 is too coarse to trust as a 'live is current' signal because it only changes on manual bumps. The build_sha field changes per commit and is set by deploy.sh. """ monkeypatch.setenv("ATOCORE_BUILD_SHA", "abc1234567890fedcba0987654321") monkeypatch.setenv("ATOCORE_BUILD_TIME", "2026-04-09T01:23:45Z") monkeypatch.setenv("ATOCORE_BUILD_BRANCH", "main") client = TestClient(app) response = client.get("/health") assert response.status_code == 200 body = response.json() assert body["build_sha"] == "abc1234567890fedcba0987654321" assert body["build_time"] == "2026-04-09T01:23:45Z" assert body["build_branch"] == "main" def test_health_endpoint_reports_unknown_when_build_env_unset(tmp_data_dir, monkeypatch): """When deploy.sh hasn't set the build env vars (e.g. someone ran `docker compose up` directly), /health reports 'unknown' for all three build fields. This is a clear signal to the operator that the deploy provenance is missing and they should re-run via deploy.sh.""" monkeypatch.delenv("ATOCORE_BUILD_SHA", raising=False) monkeypatch.delenv("ATOCORE_BUILD_TIME", raising=False) monkeypatch.delenv("ATOCORE_BUILD_BRANCH", raising=False) client = TestClient(app) response = client.get("/health") assert response.status_code == 200 body = response.json() assert body["build_sha"] == "unknown" assert body["build_time"] == "unknown" assert body["build_branch"] == "unknown" def test_projects_endpoint_reports_registered_projects(tmp_data_dir, monkeypatch): vault_dir = tmp_data_dir / "vault-source" drive_dir = tmp_data_dir / "drive-source" config_dir = tmp_data_dir / "config" project_dir = vault_dir / "incoming" / "projects" / "p04-gigabit" project_dir.mkdir(parents=True) drive_dir.mkdir() config_dir.mkdir() registry_path = config_dir / "project-registry.json" registry_path.write_text( """ { "projects": [ { "id": "p04-gigabit", "aliases": ["p04"], "description": "P04 docs", "ingest_roots": [ {"source": "vault", "subpath": "incoming/projects/p04-gigabit"} ] } ] } """.strip(), encoding="utf-8", ) monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir)) monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir)) monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path)) config.settings = config.Settings() client = TestClient(app) response = client.get("/projects") assert response.status_code == 200 body = response.json() assert body["projects"][0]["id"] == "p04-gigabit" assert body["projects"][0]["ingest_roots"][0]["exists"] is True def test_project_refresh_endpoint_uses_registered_roots(tmp_data_dir, monkeypatch): vault_dir = tmp_data_dir / "vault-source" drive_dir = tmp_data_dir / "drive-source" config_dir = tmp_data_dir / "config" project_dir = vault_dir / "incoming" / "projects" / "p05-interferometer" project_dir.mkdir(parents=True) drive_dir.mkdir() config_dir.mkdir() registry_path = config_dir / "project-registry.json" registry_path.write_text( """ { "projects": [ { "id": "p05-interferometer", "aliases": ["p05"], "description": "P05 docs", "ingest_roots": [ {"source": "vault", "subpath": "incoming/projects/p05-interferometer"} ] } ] } """.strip(), encoding="utf-8", ) calls = [] def fake_refresh_registered_project(project_name, purge_deleted=False): calls.append((project_name, purge_deleted)) return { "project": "p05-interferometer", "aliases": ["p05"], "description": "P05 docs", "purge_deleted": purge_deleted, "status": "ingested", "roots_ingested": 1, "roots_skipped": 0, "roots": [ { "source": "vault", "subpath": "incoming/projects/p05-interferometer", "path": str(project_dir), "status": "ingested", "results": [], } ], } monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir)) monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir)) monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path)) config.settings = config.Settings() monkeypatch.setattr("atocore.api.routes.refresh_registered_project", fake_refresh_registered_project) client = TestClient(app) response = client.post("/projects/p05/refresh") assert response.status_code == 200 assert calls == [("p05", False)] assert response.json()["project"] == "p05-interferometer" def test_project_refresh_endpoint_serializes_ingestion(tmp_data_dir, monkeypatch): config.settings = config.Settings() events = [] @contextmanager def fake_lock(): events.append("enter") try: yield finally: events.append("exit") def fake_refresh_registered_project(project_name, purge_deleted=False): events.append(("refresh", project_name, purge_deleted)) return { "project": "p05-interferometer", "aliases": ["p05"], "description": "P05 docs", "purge_deleted": purge_deleted, "status": "nothing_to_ingest", "roots_ingested": 0, "roots_skipped": 0, "roots": [], } monkeypatch.setattr("atocore.api.routes.exclusive_ingestion", fake_lock) monkeypatch.setattr("atocore.api.routes.refresh_registered_project", fake_refresh_registered_project) client = TestClient(app) response = client.post("/projects/p05/refresh") assert response.status_code == 200 assert events == ["enter", ("refresh", "p05", False), "exit"] def test_projects_template_endpoint_returns_template(tmp_data_dir, monkeypatch): config.settings = config.Settings() client = TestClient(app) response = client.get("/projects/template") assert response.status_code == 200 body = response.json() assert body["allowed_sources"] == ["vault", "drive"] assert body["template"]["projects"][0]["id"] == "p07-example" def test_project_proposal_endpoint_returns_normalized_preview(tmp_data_dir, monkeypatch): vault_dir = tmp_data_dir / "vault-source" drive_dir = tmp_data_dir / "drive-source" config_dir = tmp_data_dir / "config" staged = vault_dir / "incoming" / "projects" / "p07-example" staged.mkdir(parents=True) drive_dir.mkdir() config_dir.mkdir() registry_path = config_dir / "project-registry.json" registry_path.write_text('{"projects": []}', encoding="utf-8") monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir)) monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir)) monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path)) config.settings = config.Settings() client = TestClient(app) response = client.post( "/projects/proposal", json={ "project_id": "p07-example", "aliases": ["p07", "example-project", "p07"], "description": "Example project", "ingest_roots": [ { "source": "vault", "subpath": "incoming/projects/p07-example", "label": "Primary docs", } ], }, ) assert response.status_code == 200 body = response.json() assert body["project"]["aliases"] == ["p07", "example-project"] assert body["resolved_ingest_roots"][0]["exists"] is True assert body["valid"] is True def test_project_register_endpoint_persists_entry(tmp_data_dir, monkeypatch): vault_dir = tmp_data_dir / "vault-source" drive_dir = tmp_data_dir / "drive-source" config_dir = tmp_data_dir / "config" staged = vault_dir / "incoming" / "projects" / "p07-example" staged.mkdir(parents=True) drive_dir.mkdir() config_dir.mkdir() registry_path = config_dir / "project-registry.json" registry_path.write_text('{"projects": []}', encoding="utf-8") monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir)) monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir)) monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path)) config.settings = config.Settings() client = TestClient(app) response = client.post( "/projects/register", json={ "project_id": "p07-example", "aliases": ["p07", "example-project"], "description": "Example project", "ingest_roots": [ { "source": "vault", "subpath": "incoming/projects/p07-example", "label": "Primary docs", } ], }, ) assert response.status_code == 200 body = response.json() assert body["status"] == "registered" assert body["project"]["id"] == "p07-example" assert '"p07-example"' in registry_path.read_text(encoding="utf-8") def test_project_register_endpoint_rejects_collisions(tmp_data_dir, monkeypatch): vault_dir = tmp_data_dir / "vault-source" drive_dir = tmp_data_dir / "drive-source" config_dir = tmp_data_dir / "config" vault_dir.mkdir() drive_dir.mkdir() config_dir.mkdir() registry_path = config_dir / "project-registry.json" registry_path.write_text( """ { "projects": [ { "id": "p05-interferometer", "aliases": ["p05", "interferometer"], "ingest_roots": [ {"source": "vault", "subpath": "incoming/projects/p05-interferometer"} ] } ] } """.strip(), encoding="utf-8", ) monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir)) monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir)) monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path)) config.settings = config.Settings() client = TestClient(app) response = client.post( "/projects/register", json={ "project_id": "p07-example", "aliases": ["interferometer"], "ingest_roots": [ { "source": "vault", "subpath": "incoming/projects/p07-example", } ], }, ) assert response.status_code == 400 assert "collisions" in response.json()["detail"] def test_project_update_endpoint_persists_changes(tmp_data_dir, monkeypatch): vault_dir = tmp_data_dir / "vault-source" drive_dir = tmp_data_dir / "drive-source" config_dir = tmp_data_dir / "config" project_dir = vault_dir / "incoming" / "projects" / "p04-gigabit" project_dir.mkdir(parents=True) drive_dir.mkdir() config_dir.mkdir() registry_path = config_dir / "project-registry.json" registry_path.write_text( """ { "projects": [ { "id": "p04-gigabit", "aliases": ["p04", "gigabit"], "description": "Old description", "ingest_roots": [ {"source": "vault", "subpath": "incoming/projects/p04-gigabit"} ] } ] } """.strip(), encoding="utf-8", ) monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir)) monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir)) monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path)) config.settings = config.Settings() client = TestClient(app) response = client.put( "/projects/p04", json={ "aliases": ["p04", "gigabit", "gigabit-project"], "description": "Updated P04 docs", }, ) assert response.status_code == 200 body = response.json() assert body["status"] == "updated" assert body["project"]["aliases"] == ["p04", "gigabit", "gigabit-project"] assert body["project"]["description"] == "Updated P04 docs" def test_project_update_endpoint_rejects_collisions(tmp_data_dir, monkeypatch): vault_dir = tmp_data_dir / "vault-source" drive_dir = tmp_data_dir / "drive-source" config_dir = tmp_data_dir / "config" vault_dir.mkdir() drive_dir.mkdir() config_dir.mkdir() registry_path = config_dir / "project-registry.json" registry_path.write_text( """ { "projects": [ { "id": "p04-gigabit", "aliases": ["p04", "gigabit"], "ingest_roots": [ {"source": "vault", "subpath": "incoming/projects/p04-gigabit"} ] }, { "id": "p05-interferometer", "aliases": ["p05", "interferometer"], "ingest_roots": [ {"source": "vault", "subpath": "incoming/projects/p05-interferometer"} ] } ] } """.strip(), encoding="utf-8", ) monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir)) monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir)) monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path)) config.settings = config.Settings() client = TestClient(app) response = client.put( "/projects/p04", json={ "aliases": ["p04", "interferometer"], }, ) assert response.status_code == 400 assert "collisions" in response.json()["detail"] def test_admin_backup_create_without_chroma(tmp_data_dir, monkeypatch): config.settings = config.Settings() captured = {} def fake_create_runtime_backup(timestamp=None, include_chroma=False): captured["include_chroma"] = include_chroma return { "created_at": "2026-04-06T23:00:00+00:00", "backup_root": "/tmp/fake", "db_snapshot_path": "/tmp/fake/db/atocore.db", "db_size_bytes": 0, "registry_snapshot_path": "", "chroma_snapshot_path": "", "chroma_snapshot_bytes": 0, "chroma_snapshot_files": 0, "chroma_snapshot_included": False, "vector_store_note": "skipped", } monkeypatch.setattr("atocore.api.routes.create_runtime_backup", fake_create_runtime_backup) client = TestClient(app) response = client.post("/admin/backup", json={}) assert response.status_code == 200 assert captured == {"include_chroma": False} body = response.json() assert body["chroma_snapshot_included"] is False def test_admin_backup_create_with_chroma_holds_lock(tmp_data_dir, monkeypatch): config.settings = config.Settings() events = [] @contextmanager def fake_lock(): events.append("enter") try: yield finally: events.append("exit") def fake_create_runtime_backup(timestamp=None, include_chroma=False): events.append(("backup", include_chroma)) return { "created_at": "2026-04-06T23:30:00+00:00", "backup_root": "/tmp/fake", "db_snapshot_path": "/tmp/fake/db/atocore.db", "db_size_bytes": 0, "registry_snapshot_path": "", "chroma_snapshot_path": "/tmp/fake/chroma", "chroma_snapshot_bytes": 4, "chroma_snapshot_files": 1, "chroma_snapshot_included": True, "vector_store_note": "included", } monkeypatch.setattr("atocore.api.routes.exclusive_ingestion", fake_lock) monkeypatch.setattr("atocore.api.routes.create_runtime_backup", fake_create_runtime_backup) client = TestClient(app) response = client.post("/admin/backup", json={"include_chroma": True}) assert response.status_code == 200 assert events == ["enter", ("backup", True), "exit"] assert response.json()["chroma_snapshot_included"] is True def test_admin_backup_list_and_validate_endpoints(tmp_data_dir, monkeypatch): config.settings = config.Settings() def fake_list_runtime_backups(): return [ { "stamp": "20260406T220000Z", "path": "/tmp/fake/snapshots/20260406T220000Z", "has_metadata": True, "metadata": {"db_snapshot_path": "/tmp/fake/snapshots/20260406T220000Z/db/atocore.db"}, } ] def fake_validate_backup(stamp): if stamp == "missing": return { "stamp": stamp, "path": f"/tmp/fake/snapshots/{stamp}", "exists": False, "errors": ["snapshot_directory_missing"], } return { "stamp": stamp, "path": f"/tmp/fake/snapshots/{stamp}", "exists": True, "db_ok": True, "registry_ok": True, "chroma_ok": None, "valid": True, "errors": [], } monkeypatch.setattr("atocore.api.routes.list_runtime_backups", fake_list_runtime_backups) monkeypatch.setattr("atocore.api.routes.validate_backup", fake_validate_backup) client = TestClient(app) listing = client.get("/admin/backup") assert listing.status_code == 200 listing_body = listing.json() assert "backup_dir" in listing_body assert listing_body["backups"][0]["stamp"] == "20260406T220000Z" valid = client.get("/admin/backup/20260406T220000Z/validate") assert valid.status_code == 200 assert valid.json()["valid"] is True missing = client.get("/admin/backup/missing/validate") assert missing.status_code == 404 def test_query_endpoint_accepts_project_hint(monkeypatch): def fake_retrieve(prompt, top_k=10, filter_tags=None, project_hint=None): assert prompt == "architecture" assert top_k == 3 assert project_hint == "p04-gigabit" return [] monkeypatch.setattr("atocore.api.routes.retrieve", fake_retrieve) client = TestClient(app) response = client.post( "/query", json={ "prompt": "architecture", "top_k": 3, "project": "p04-gigabit", }, ) assert response.status_code == 200 assert response.json()["results"] == []