91 lines
3.8 KiB
Python
91 lines
3.8 KiB
Python
|
|
import pytest
|
|
|
|
from tht.config import (
|
|
canonical_effective_config_json,
|
|
effective_config_fingerprint,
|
|
effective_config_identity,
|
|
effective_config_input_fingerprint,
|
|
)
|
|
|
|
|
|
def _write_cfg(tmp_path, raw):
|
|
import yaml as _yaml
|
|
|
|
from tht.config import load_config
|
|
path = tmp_path / "config.yaml"
|
|
path.write_text(_yaml.safe_dump(raw), encoding="utf-8")
|
|
cfg = load_config(path)
|
|
cfg._workspace_id = raw.get("workspace", {}).get("id", "psd")
|
|
cfg._config_source = f"workspace://{cfg._workspace_id}"
|
|
return cfg
|
|
|
|
|
|
def _cfg(tmp_path, *, transport="thoth_rest", base_url="http://dwh.example.invalid", collection="psd", model="qwen3-embedding:0.6b"):
|
|
return {
|
|
"schemaVersion": 1,
|
|
"workspace": {"schema_version": 3, "id": "psd", "name": "PSD", "language": "it"},
|
|
"dwh": {
|
|
"type": transport,
|
|
"database": {"database": "warehouse", "schema": "dw"},
|
|
"endpoint": {"base_url": base_url, "api_key": "secret"},
|
|
} if transport == "thoth_rest" else {
|
|
"type": "postgres_direct",
|
|
"connection": {"host": "h", "port": 5432, "database": "warehouse", "schema": "dw", "user": "reader", "password": "secret"},
|
|
},
|
|
"vectors": {"type": "qdrant", "base_url": "http://qdrant:6333", "collection": collection, "collection_lifecycle": "self_heal"},
|
|
"embeddings": {"provider": "ollama_internal", "base_url": "http://embedding:11434", "model": model, "dimensions": 1024},
|
|
"roots": {"artifacts": str(tmp_path / "artifacts"), "indexes": str(tmp_path / "indexes")},
|
|
"paths": {"artifacts": str(tmp_path / "artifacts"), "indexes": str(tmp_path / "indexes"), "sessions": str(tmp_path / "sessions")},
|
|
}
|
|
|
|
|
|
@pytest.fixture()
|
|
def cfg(tmp_path):
|
|
return _write_cfg(tmp_path, _cfg(tmp_path))
|
|
|
|
|
|
def test_canonical_json_is_deterministic_and_key_ordered(cfg):
|
|
doc = canonical_effective_config_json(cfg)
|
|
assert doc == canonical_effective_config_json(cfg)
|
|
keys = list(__import__("json").loads(doc).keys())
|
|
assert keys == ["schemaVersion", "dwh", "vector", "embedding", "roots"]
|
|
|
|
|
|
def test_canonical_excludes_credentials_and_evidence(cfg):
|
|
doc = canonical_effective_config_json(cfg)
|
|
assert "secret" not in doc
|
|
assert "password" not in doc
|
|
assert "evidence" not in doc
|
|
assert "session_storage" not in doc
|
|
assert "runtime_identity" not in doc
|
|
|
|
|
|
def test_identity_and_fingerprints_format(cfg):
|
|
ident = effective_config_identity("psd", cfg)
|
|
assert ident == "workspace://psd@v1:" + effective_config_fingerprint(cfg)[len("sha256:"):]
|
|
assert effective_config_input_fingerprint("psd", cfg).startswith("sha256:")
|
|
assert len(effective_config_fingerprint(cfg)) == len("sha256:") + 64
|
|
|
|
|
|
def test_content_only_change_keeps_identity(tmp_path):
|
|
base = _write_cfg(tmp_path, _cfg(tmp_path))
|
|
changed = _cfg(tmp_path)
|
|
# A non-DWH-affecting change (collection lifecycle policy) must not alter the identity;
|
|
# the canonical document excludes it. Evidence changes are proven end-to-end by acceptance.
|
|
changed["vectors"]["collection_lifecycle"] = "require_existing"
|
|
changed_cfg = _write_cfg(tmp_path, changed)
|
|
assert effective_config_identity("psd", base) == effective_config_identity("psd", changed_cfg)
|
|
|
|
|
|
def test_dwh_affecting_change_alters_identity(tmp_path):
|
|
base = _write_cfg(tmp_path, _cfg(tmp_path))
|
|
other = _write_cfg(tmp_path, _cfg(tmp_path, base_url="http://other.example.invalid"))
|
|
assert effective_config_identity("psd", base) != effective_config_identity("psd", other)
|
|
|
|
|
|
def test_transport_change_alters_identity(tmp_path):
|
|
rest = _write_cfg(tmp_path, _cfg(tmp_path, transport="thoth_rest"))
|
|
direct = _write_cfg(tmp_path, _cfg(tmp_path, transport="postgres_direct"))
|
|
assert effective_config_identity("psd", rest) != effective_config_identity("psd", direct)
|