feat(storage): resolve portable workspace roots

This commit is contained in:
2026-07-11 21:21:06 +02:00
parent 3e2ad7955c
commit eacb139e22
6 changed files with 299 additions and 0 deletions
+96
View File
@@ -0,0 +1,96 @@
import json
from pathlib import Path
from typer.testing import CliRunner
from tht.cli import app
runner = CliRunner()
def _config(path: Path, *, absolute_sessions: Path | None = None) -> Path:
sessions = absolute_sessions or Path("sessions")
path.write_text(
"dwh:\n"
" type: postgres_direct\n"
" connection:\n"
" database: patient_db\n"
" schema: private_schema\n"
" user: pii_user\n"
" password: super-secret\n"
"roots:\n"
" artifacts: artifacts\n"
" indexes: indexes\n"
f" sessions: {sessions}\n"
)
return path
def test_doctor_json_reports_portable_path_statuses_without_secrets(monkeypatch, tmp_path):
cfg = _config(tmp_path / "demo.yaml")
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
result = runner.invoke(app, ["doctor", "--json", "--config", str(cfg)])
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload == {
"ok": True,
"components": {
"config": {"status": "ok"},
"data_root": {"status": "ok"},
"workspace_paths": {"status": "ok", "legacy_absolute": []},
},
}
assert "super-secret" not in result.stdout
assert "patient_db" not in result.stdout
assert "pii_user" not in result.stdout
assert result.stderr == ""
def test_doctor_json_flags_absolute_legacy_paths(monkeypatch, tmp_path):
cfg = _config(tmp_path / "demo.yaml", absolute_sessions=tmp_path / "old-sessions")
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
result = runner.invoke(app, ["doctor", "--json", "--config", str(cfg)])
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["components"]["workspace_paths"] == {
"status": "warning",
"legacy_absolute": ["sessions"],
}
assert str(tmp_path) not in result.stdout
def test_doctor_json_returns_structured_config_error(monkeypatch, tmp_path):
cfg = _config(tmp_path / "demo.yaml")
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
cfg.write_text(cfg.read_text().replace("sessions: sessions", "sessions: ../../private"))
result = runner.invoke(app, ["doctor", "--json", "--config", str(cfg)])
assert result.exit_code == 1
payload = json.loads(result.stdout)
assert payload["ok"] is False
assert payload["components"]["workspace_paths"]["status"] == "error"
assert "outside workspace root" in payload["components"]["workspace_paths"]["message"]
assert result.stderr == ""
def test_doctor_json_does_not_echo_invalid_config_values(monkeypatch, tmp_path):
cfg = _config(tmp_path / "demo.yaml")
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
cfg.write_text(cfg.read_text().replace("database: patient_db", ""))
result = runner.invoke(app, ["doctor", "--json", "--config", str(cfg)])
assert result.exit_code == 1
payload = json.loads(result.stdout)
assert payload["components"]["config"] == {
"status": "error",
"message": "configuration is invalid",
}
assert "super-secret" not in result.stdout
assert "pii_user" not in result.stdout
+77
View File
@@ -0,0 +1,77 @@
from pathlib import Path
import pytest
from tht.config import ConfigError, load_config
from tht.paths import resolve_workspace_paths
def _write_config(path: Path, *, sessions: str = "sessions", absolute: Path | None = None) -> Path:
root = absolute or Path("artifacts")
path.write_text(
"dwh:\n"
" type: postgres_direct\n"
" connection:\n"
" database: db\n"
" schema: public\n"
" user: user\n"
" password: secret\n"
"roots:\n"
f" artifacts: {root}\n"
f" indexes: {root if absolute else 'indexes'}\n"
f" sessions: {absolute if absolute else sessions}\n"
)
return path
def test_relative_paths_resolve_under_workspace_root(tmp_path):
cfg_path = _write_config(tmp_path / "demo.yaml")
cfg = load_config(cfg_path)
resolved = resolve_workspace_paths(cfg_path, cfg, tmp_path / "data")
assert resolved.workspace == tmp_path / "data/workspaces/demo"
assert resolved.sessions == tmp_path / "data/workspaces/demo/sessions"
assert resolved.artifacts == tmp_path / "data/workspaces/demo/artifacts"
assert resolved.indexes == tmp_path / "data/workspaces/demo/indexes"
assert resolved.corpus == tmp_path / "data/workspaces/demo/corpus"
def test_absolute_legacy_paths_are_preserved(tmp_path):
legacy = tmp_path / "existing-workspace"
cfg_path = _write_config(tmp_path / "demo.yaml", absolute=legacy)
cfg = load_config(cfg_path)
resolved = resolve_workspace_paths(cfg_path, cfg, tmp_path / "data")
assert resolved.sessions == legacy
assert resolved.artifacts == legacy
assert resolved.indexes == legacy
def test_path_escape_is_rejected(tmp_path):
cfg_path = _write_config(tmp_path / "demo.yaml", sessions="../../private")
cfg = load_config(cfg_path)
with pytest.raises(ConfigError, match="outside workspace root"):
resolve_workspace_paths(cfg_path, cfg, tmp_path / "data")
def test_data_root_environment_activates_portable_paths(monkeypatch, tmp_path):
cfg_path = _write_config(tmp_path / "demo.yaml")
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
cfg = load_config(cfg_path)
assert cfg.paths.sessions == tmp_path / "data/workspaces/demo/sessions"
assert cfg.paths.artifacts == tmp_path / "data/workspaces/demo/artifacts"
def test_no_data_root_preserves_legacy_relative_paths(monkeypatch, tmp_path):
cfg_path = _write_config(tmp_path / "demo.yaml")
monkeypatch.delenv("THT_DATA_ROOT", raising=False)
cfg = load_config(cfg_path)
assert cfg.paths.sessions == Path("sessions")
assert cfg.paths.artifacts == Path("artifacts")