419 lines
13 KiB
Python
419 lines
13 KiB
Python
"""Tests for storage-related API readiness endpoints."""
|
|
|
|
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_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,
|
|
"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_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_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"] == []
|