63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
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"
|
|
)
|