feat: revision-qualified annotations root for pinned runtimes (P5)

2026-08-13 05:02:19 +02:00
parent 259d5a0374
commit 5249798c03
6 changed files with 98 additions and 3 deletions
@@ -0,0 +1,62 @@
+from pathlib import Path
+
+from tht.cli.schema_cmd import annotations_path
+from tht.config import load_config
+
+REVISION = "a" * 40
+
+
+def _config(tmp_path: Path, *, with_annotations_root: bool = True) -> Path:
+    annotations = tmp_path / "revisions" / REVISION / "artifacts"
+    roots = f"""
+roots:
+  sessions: {tmp_path / 'sessions'}
+  artifacts: {tmp_path / 'artifacts'}
+  indexes: {tmp_path / 'indexes'}
+  memory: {tmp_path / 'memory'}
+"""
+    if with_annotations_root:
+        roots += f"  annotations_root: {annotations}\n"
+    path = tmp_path / "runtime.yaml"
+    path.write_text(f"""
+runtime_identity:
+  workspace_id: demo
+  workspace_revision: {REVISION}
+  source_identity: workspace://demo
+dwh:
+  type: postgres_direct
+  connection: {{database: warehouse, schema: analytics, user: reader, password: secret}}
+{roots}
+""")
+    return path
+
+
+def test_annotations_path_prefers_revision_root(tmp_path, monkeypatch):
+    monkeypatch.delenv("THT_HOME", raising=False)
+    monkeypatch.delenv("THT_DATA_ROOT", raising=False)
+    cfg = load_config(_config(tmp_path))
+
+    assert cfg.paths.annotations_root == tmp_path / "revisions" / REVISION / "artifacts"
+    assert annotations_path(cfg) == (
+        tmp_path / "revisions" / REVISION / "artifacts" / "mschema" / "annotations.yaml"
+    )
+
+
+def test_annotations_path_falls_back_to_legacy_root(tmp_path, monkeypatch):
+    monkeypatch.delenv("THT_HOME", raising=False)
+    monkeypatch.delenv("THT_DATA_ROOT", raising=False)
+    cfg = load_config(_config(tmp_path, with_annotations_root=False))
+
+    assert cfg.paths.annotations_root is None
+    assert annotations_path(cfg) == tmp_path / "artifacts" / "mschema" / "annotations.yaml"
+
+
+def test_load_config_preserves_annotations_root_with_data_root(tmp_path, monkeypatch):
+    monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
+    monkeypatch.delenv("THT_HOME", raising=False)
+    cfg = load_config(_config(tmp_path))
+
+    assert cfg.paths.annotations_root == tmp_path / "revisions" / REVISION / "artifacts"
+    assert annotations_path(cfg) == (
+        tmp_path / "revisions" / REVISION / "artifacts" / "mschema" / "annotations.yaml"
+    )