Antoine/ATOCore
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add project registry refresh foundation

Browse Source
This commit is contained in:
Anto01
2026-04-06 08:02:13 -04:00
parent 0f95415530
commit 8293099025
12 changed files with 526 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
tests/test_api_storage.py
View File

@@ -46,3 +46,107 @@ def test_health_endpoint_exposes_machine_paths_and_source_readiness(tmp_data_dir
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"

6
tests/test_config.py
View File

@@ -12,6 +12,9 @@ def test_settings_resolve_canonical_directories(tmp_path, monkeypatch):
monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(tmp_path / "drive-source"))
monkeypatch.setenv("ATOCORE_LOG_DIR", str(tmp_path / "logs"))
monkeypatch.setenv("ATOCORE_BACKUP_DIR", str(tmp_path / "backups"))
monkeypatch.setenv(
"ATOCORE_PROJECT_REGISTRY_PATH", str(tmp_path / "config" / "project-registry.json")
)
settings = config.Settings()
@@ -24,6 +27,9 @@ def test_settings_resolve_canonical_directories(tmp_path, monkeypatch):
assert settings.resolved_log_dir == (tmp_path / "logs").resolve()
assert settings.resolved_backup_dir == (tmp_path / "backups").resolve()
assert settings.resolved_run_dir == (tmp_path / "run").resolve()
assert settings.resolved_project_registry_path == (
tmp_path / "config" / "project-registry.json"
).resolve()
def test_settings_keep_legacy_db_path_when_present(tmp_path, monkeypatch):

152
tests/test_project_registry.py Normal file
View File

@@ -0,0 +1,152 @@
"""Tests for project registry resolution and refresh behavior."""
import json
import atocore.config as config
from atocore.projects.registry import (
get_registered_project,
list_registered_projects,
refresh_registered_project,
)
def test_project_registry_lists_projects_with_resolved_roots(tmp_path, monkeypatch):
vault_dir = tmp_path / "vault"
drive_dir = tmp_path / "drive"
config_dir = tmp_path / "config"
vault_dir.mkdir()
drive_dir.mkdir()
config_dir.mkdir()
(vault_dir / "incoming" / "projects" / "p04-gigabit").mkdir(parents=True)
registry_path = config_dir / "project-registry.json"
registry_path.write_text(
json.dumps(
{
"projects": [
{
"id": "p04-gigabit",
"aliases": ["p04", "gigabit"],
"description": "P04 docs",
"ingest_roots": [
{
"source": "vault",
"subpath": "incoming/projects/p04-gigabit",
"label": "P04 staged docs",
}
],
}
]
}
),
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))
original_settings = config.settings
try:
config.settings = config.Settings()
projects = list_registered_projects()
finally:
config.settings = original_settings
assert len(projects) == 1
assert projects[0]["id"] == "p04-gigabit"
assert projects[0]["ingest_roots"][0]["exists"] is True
def test_project_registry_resolves_alias(tmp_path, monkeypatch):
vault_dir = tmp_path / "vault"
drive_dir = tmp_path / "drive"
config_dir = tmp_path / "config"
vault_dir.mkdir()
drive_dir.mkdir()
config_dir.mkdir()
registry_path = config_dir / "project-registry.json"
registry_path.write_text(
json.dumps(
{
"projects": [
{
"id": "p05-interferometer",
"aliases": ["p05", "interferometer"],
"ingest_roots": [
{"source": "vault", "subpath": "incoming/projects/p05-interferometer"}
],
}
]
}
),
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))
original_settings = config.settings
try:
config.settings = config.Settings()
project = get_registered_project("p05")
finally:
config.settings = original_settings
assert project is not None
assert project.project_id == "p05-interferometer"
def test_refresh_registered_project_ingests_registered_roots(tmp_path, monkeypatch):
vault_dir = tmp_path / "vault"
drive_dir = tmp_path / "drive"
config_dir = tmp_path / "config"
project_dir = vault_dir / "incoming" / "projects" / "p06-polisher"
project_dir.mkdir(parents=True)
drive_dir.mkdir()
config_dir.mkdir()
registry_path = config_dir / "project-registry.json"
registry_path.write_text(
json.dumps(
{
"projects": [
{
"id": "p06-polisher",
"aliases": ["p06", "polisher"],
"description": "P06 docs",
"ingest_roots": [
{"source": "vault", "subpath": "incoming/projects/p06-polisher"}
],
}
]
}
),
encoding="utf-8",
)
calls = []
def fake_ingest_folder(path, purge_deleted=True):
calls.append((str(path), purge_deleted))
return [{"file": str(path / "README.md"), "status": "ingested"}]
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))
original_settings = config.settings
try:
config.settings = config.Settings()
monkeypatch.setattr("atocore.projects.registry.ingest_folder", fake_ingest_folder)
result = refresh_registered_project("polisher")
finally:
config.settings = original_settings
assert result["project"] == "p06-polisher"
assert len(calls) == 1
assert calls[0][0].endswith("p06-polisher")
assert calls[0][1] is False
assert result["roots"][0]["status"] == "ingested"