517 lines
16 KiB
Python
517 lines
16 KiB
Python
import json
|
|
|
|
import pytest
|
|
import yaml
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.adapters.evidence import FilesystemEvidenceSource, HttpManifestEvidenceSource
|
|
from tht.adapters.factory import build_evidence_sources
|
|
from tht.cli import app
|
|
from tht.config import (
|
|
ConfigError,
|
|
PgvectorDirectConfig,
|
|
PostgresDwhConfig,
|
|
QdrantConfig,
|
|
ThothRestDwhConfig,
|
|
ThothVectorHttpConfig,
|
|
load_config,
|
|
workspace_id_for_config,
|
|
)
|
|
|
|
|
|
def test_direct_vector_passwords_load_from_file_references(monkeypatch, tmp_path):
|
|
reader = tmp_path / "reader"
|
|
writer = tmp_path / "writer"
|
|
reader.write_text("reader-secret")
|
|
writer.write_text("writer-secret")
|
|
monkeypatch.setenv("READER_FILE", str(reader))
|
|
monkeypatch.setenv("WRITER_FILE", str(writer))
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text("""
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {database: d, schema: public, user: u, password: p}
|
|
vectors:
|
|
type: pgvector_direct
|
|
reader: {database: d, schema: vectors, user: r, password_file: '${READER_FILE}'}
|
|
writer: {database: d, schema: vectors, user: w, password_file: '${WRITER_FILE}'}
|
|
""")
|
|
config = load_config(workspace)
|
|
assert config.vectors.reader.password == "reader-secret"
|
|
assert config.vectors.writer.password == "writer-secret"
|
|
|
|
|
|
def test_direct_vector_secret_file_rejects_whitespace(tmp_path):
|
|
secret = tmp_path / "reader"
|
|
secret.write_text("bad secret")
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text(f"""
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {{database: d, schema: public, user: u, password: p}}
|
|
vectors:
|
|
type: pgvector_direct
|
|
reader: {{database: d, schema: vectors, user: r, password_file: {secret}}}
|
|
""")
|
|
with pytest.raises(ConfigError, match="secret file"):
|
|
load_config(workspace)
|
|
|
|
|
|
def test_loads_discriminated_dwh_and_vector_resources(tmp_path):
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text(
|
|
"""
|
|
dwh:
|
|
type: thoth_rest
|
|
database:
|
|
database: analytics
|
|
schema: mart
|
|
endpoint:
|
|
base_url: https://dwh.example.test/
|
|
api_key: dwh-reader
|
|
vectors:
|
|
type: thoth_vector_http
|
|
reader:
|
|
base_url: https://vectors.example.test/
|
|
api_key: vector-reader
|
|
writer:
|
|
base_url: https://vectors.example.test/
|
|
api_key: vector-writer
|
|
roots:
|
|
artifacts: build/artifacts
|
|
indexes: build/indexes
|
|
sessions: build/sessions
|
|
"""
|
|
)
|
|
|
|
cfg = load_config(workspace)
|
|
|
|
assert isinstance(cfg.dwh, ThothRestDwhConfig)
|
|
assert cfg.dwh.database.db_schema == "mart"
|
|
assert isinstance(cfg.vectors, ThothVectorHttpConfig)
|
|
assert cfg.vectors.writer.api_key == "vector-writer"
|
|
assert cfg.roots.sessions.as_posix() == "build/sessions"
|
|
|
|
|
|
def test_runtime_handoff_preserves_canonical_identity_and_durable_roots(monkeypatch, tmp_path):
|
|
data_root = tmp_path / "data"
|
|
runtime_root = data_root / "sessions" / "psd-clinical"
|
|
workspace = tmp_path / "runtime-random-uuid.yaml"
|
|
workspace.write_text(f"""
|
|
runtime_identity:
|
|
workspace_id: psd-clinical
|
|
workspace_revision: {'a' * 40}
|
|
source_identity: workspace://psd-clinical
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {{database: analytics, schema: mart, user: reader, password: secret}}
|
|
vectors:
|
|
type: pgvector_direct
|
|
connection: {{database: analytics, schema: vectors, user: vector, password: secret}}
|
|
roots:
|
|
sessions: {runtime_root / 'sessions'}
|
|
artifacts: {runtime_root / 'artifacts'}
|
|
indexes: {runtime_root / 'indexes'}
|
|
embeddings: {{provider: ollama_internal, base_url: http://embedding:11434, model: qwen3-embedding:0.6b, dim: 1024}}
|
|
""")
|
|
monkeypatch.setenv("THT_DATA_ROOT", str(data_root))
|
|
|
|
cfg = load_config(workspace)
|
|
|
|
assert cfg._workspace_id == "psd-clinical"
|
|
assert cfg._workspace_revision == "a" * 40
|
|
assert cfg._config_source == "workspace://psd-clinical"
|
|
assert cfg.paths.sessions == runtime_root / "sessions"
|
|
assert cfg.paths.artifacts == runtime_root / "artifacts"
|
|
assert cfg.paths.indexes == runtime_root / "indexes"
|
|
|
|
|
|
def test_runtime_identity_is_authoritative_over_runtime_filename(tmp_path):
|
|
workspace = tmp_path / "runtime-random-uuid.yaml"
|
|
workspace.write_text(f"""
|
|
runtime_identity:
|
|
workspace_id: psd-clinical
|
|
workspace_revision: {'a' * 40}
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {{database: analytics, schema: mart, user: reader, password: secret}}
|
|
roots: {{sessions: sessions, artifacts: artifacts, indexes: indexes}}
|
|
""")
|
|
|
|
cfg = load_config(workspace)
|
|
|
|
assert workspace_id_for_config(cfg, workspace) == "psd-clinical"
|
|
|
|
|
|
def test_runtime_identity_rejects_a_source_for_another_workspace(tmp_path):
|
|
workspace = tmp_path / "runtime-random-uuid.yaml"
|
|
workspace.write_text(f"""
|
|
runtime_identity:
|
|
workspace_id: psd-clinical
|
|
workspace_revision: {'a' * 40}
|
|
source_identity: workspace://another-workspace
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {{database: analytics, schema: mart, user: reader, password: secret}}
|
|
roots: {{sessions: sessions, artifacts: artifacts, indexes: indexes}}
|
|
""")
|
|
|
|
with pytest.raises(ConfigError, match="source_identity"):
|
|
load_config(workspace)
|
|
|
|
|
|
def test_load_config_rejects_executable_yaml_tags_without_running_them(tmp_path):
|
|
marker = tmp_path / "must-not-exist"
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text(f"dwh: !command touch {marker}\n")
|
|
|
|
with pytest.raises(ConfigError, match="YAML|configurazione"):
|
|
load_config(workspace)
|
|
|
|
assert not marker.exists()
|
|
|
|
|
|
def test_loads_direct_discriminated_resources(tmp_path):
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text(
|
|
"""
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: &database
|
|
host: db
|
|
database: analytics
|
|
schema: mart
|
|
user: reader
|
|
password: secret
|
|
vectors:
|
|
type: pgvector_direct
|
|
connection:
|
|
<<: *database
|
|
schema: vectors
|
|
"""
|
|
)
|
|
|
|
cfg = load_config(workspace)
|
|
|
|
assert isinstance(cfg.dwh, PostgresDwhConfig)
|
|
assert cfg.database.transport == "direct"
|
|
assert isinstance(cfg.vectors, PgvectorDirectConfig)
|
|
assert cfg.vector_db.db_schema == "vectors"
|
|
|
|
|
|
def test_loads_writer_only_http_vector_resource(tmp_path):
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text(
|
|
"""
|
|
dwh:
|
|
type: thoth_rest
|
|
database: {database: analytics, schema: mart}
|
|
endpoint: {base_url: https://dwh.test/, api_key: reader}
|
|
vectors:
|
|
type: thoth_vector_http
|
|
writer: {base_url: https://vectors.test/, api_key: writer}
|
|
embeddings: {provider: ollama_internal, base_url: http://embedding:11434, model: qwen3-embedding:0.6b, dim: 1024}
|
|
"""
|
|
)
|
|
|
|
cfg = load_config(workspace)
|
|
assert cfg.vectors.reader is None
|
|
assert cfg.vectors.writer.api_key == "writer"
|
|
|
|
|
|
def test_accepts_only_internal_ollama_embedding_contract(tmp_path):
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text(
|
|
"""
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {database: analytics, schema: mart, user: reader, password: secret}
|
|
resources:
|
|
embeddings:
|
|
provider: ollama_internal
|
|
base_url: http://embedding:11434
|
|
model: qwen3-embedding:0.6b
|
|
dimensions: 1024
|
|
"""
|
|
)
|
|
|
|
cfg = load_config(workspace)
|
|
|
|
assert cfg.embeddings.provider == "ollama_internal"
|
|
assert cfg.embeddings.base_url == "http://embedding:11434"
|
|
assert cfg.embeddings.model == "qwen3-embedding:0.6b"
|
|
assert cfg.embeddings.dim == 1024
|
|
|
|
|
|
def test_accepts_internal_qdrant_resource_contract(tmp_path):
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text(
|
|
"""
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {database: analytics, schema: mart, user: reader, password: secret}
|
|
resources:
|
|
vector:
|
|
engine: qdrant
|
|
base_url: http://qdrant:6333
|
|
collection: psd-clinical
|
|
embeddings:
|
|
provider: ollama_internal
|
|
base_url: http://embedding:11434
|
|
model: qwen3-embedding:0.6b
|
|
dimensions: 1024
|
|
"""
|
|
)
|
|
|
|
cfg = load_config(workspace)
|
|
|
|
assert isinstance(cfg.vectors, QdrantConfig)
|
|
assert cfg.vectors.base_url == "http://qdrant:6333"
|
|
assert cfg.vectors.collection == "psd-clinical"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("snippet", "pattern"),
|
|
[
|
|
(
|
|
"""
|
|
resources:
|
|
embeddings:
|
|
provider: openai_compatible
|
|
base_url: http://embedding:11434
|
|
model: qwen3-embedding:0.6b
|
|
dimensions: 1024
|
|
""",
|
|
"ollama_internal|provider",
|
|
),
|
|
(
|
|
"""
|
|
resources:
|
|
embeddings:
|
|
provider: ollama_internal
|
|
base_url: http://embedding:11434
|
|
model: qwen3-embedding:0.6b
|
|
dimensions: 1024
|
|
api_key: secret
|
|
""",
|
|
"api_key|extra",
|
|
),
|
|
(
|
|
"""
|
|
resources:
|
|
embeddings:
|
|
provider: ollama_internal
|
|
base_url: https://embedding:11434
|
|
model: qwen3-embedding:0.6b
|
|
dimensions: 1024
|
|
""",
|
|
"base_url|internal|private|host",
|
|
),
|
|
(
|
|
"""
|
|
resources:
|
|
embeddings:
|
|
provider: ollama_internal
|
|
base_url: http://example.com:11434
|
|
model: qwen3-embedding:0.6b
|
|
dimensions: 1024
|
|
""",
|
|
"base_url|internal|private|host",
|
|
),
|
|
],
|
|
)
|
|
def test_rejects_external_embedding_configuration(tmp_path, snippet, pattern):
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text(
|
|
"""
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {database: analytics, schema: mart, user: reader, password: secret}
|
|
"""
|
|
+ snippet
|
|
)
|
|
|
|
with pytest.raises(ConfigError, match=pattern):
|
|
load_config(workspace)
|
|
|
|
|
|
def test_rejects_external_top_level_embedding_configuration(tmp_path):
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text(
|
|
"""
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {database: analytics, schema: mart, user: reader, password: secret}
|
|
embeddings:
|
|
provider: openai_compatible
|
|
base_url: https://embedding.example.test
|
|
model: text-embedding-3-large
|
|
dim: 3072
|
|
"""
|
|
)
|
|
|
|
with pytest.raises(ConfigError, match="ollama_internal|provider|base_url|model|1024"):
|
|
load_config(workspace)
|
|
|
|
|
|
def test_builds_typed_evidence_sources_and_keeps_legacy_compatible(tmp_path):
|
|
common = """
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {database: d, schema: public, user: u, password: p}
|
|
"""
|
|
modern = tmp_path / "modern.yaml"
|
|
modern.write_text(common + f"""
|
|
evidence:
|
|
sources:
|
|
- type: filesystem
|
|
root: {tmp_path}
|
|
max_bytes: 123
|
|
- type: http
|
|
urls: ['https://example.test/doc.md']
|
|
""")
|
|
cfg = load_config(modern)
|
|
assert "example.test" not in repr(cfg.evidence)
|
|
assert "example.test" not in cfg.evidence.model_dump_json()
|
|
assert cfg.evidence.sources[1].allow_private_hosts is False
|
|
sources = build_evidence_sources(cfg)
|
|
assert isinstance(sources[0], FilesystemEvidenceSource)
|
|
assert isinstance(sources[1], HttpManifestEvidenceSource)
|
|
assert "example.test" not in repr(sources[1])
|
|
|
|
legacy = tmp_path / "legacy.yaml"
|
|
(tmp_path / "curated").mkdir()
|
|
legacy.write_text(common + f"""
|
|
evidence:
|
|
source_root: {tmp_path}
|
|
evidence_dir: curated
|
|
""")
|
|
legacy_source = build_evidence_sources(load_config(legacy))[0]
|
|
assert isinstance(legacy_source, FilesystemEvidenceSource)
|
|
assert legacy_source.root == (tmp_path / "curated").resolve()
|
|
|
|
|
|
|
|
def test_safe_validation_formatter_keeps_location_and_type_without_rejected_input(tmp_path):
|
|
workspace = tmp_path / "workspace.yaml"
|
|
workspace.write_text("""
|
|
dwh:
|
|
type: postgres_direct
|
|
connection: {database: analytics, schema: public, user: reader}
|
|
""")
|
|
|
|
with pytest.raises(ConfigError) as caught:
|
|
load_config(workspace)
|
|
|
|
message = str(caught.value)
|
|
assert "dwh.postgres_direct.connection.password" in message
|
|
assert "missing" in message
|
|
|
|
|
|
@pytest.mark.parametrize("resources", [None, [], "malformed", 7])
|
|
def test_raw_resources_non_mapping_is_a_safe_config_error(tmp_path, resources):
|
|
values = {
|
|
"dwh": {
|
|
"type": "postgres_direct",
|
|
"connection": {"database": "d", "schema": "public", "user": "u", "password": "p"},
|
|
},
|
|
"resources": resources,
|
|
}
|
|
path = tmp_path / "invalid-resources.yaml"
|
|
path.write_text(yaml.safe_dump(values))
|
|
|
|
with pytest.raises(ConfigError, match="resources"):
|
|
load_config(path)
|
|
config_result = CliRunner().invoke(app, ["config", "check", "--config", str(path)])
|
|
assert config_result.exit_code == 1
|
|
assert "Traceback" not in config_result.stderr
|
|
vector_result = CliRunner().invoke(
|
|
app, ["vector", "index-schema", "--json", "-c", str(path)]
|
|
)
|
|
assert vector_result.exit_code == 1
|
|
assert vector_result.stderr == ""
|
|
assert json.loads(vector_result.stdout) == {"status": "failed", "code": "invalid_configuration"}
|
|
|
|
|
|
@pytest.mark.parametrize("vector", [None, [], "malformed", 7])
|
|
def test_raw_resources_vector_non_mapping_is_a_safe_config_error(tmp_path, vector):
|
|
values = {
|
|
"dwh": {
|
|
"type": "postgres_direct",
|
|
"connection": {"database": "d", "schema": "public", "user": "u", "password": "p"},
|
|
},
|
|
"resources": {"vector": vector},
|
|
}
|
|
path = tmp_path / "invalid-resources-vector.yaml"
|
|
path.write_text(yaml.safe_dump(values))
|
|
|
|
with pytest.raises(ConfigError, match="resources.vector"):
|
|
load_config(path)
|
|
config_result = CliRunner().invoke(app, ["config", "check", "--config", str(path)])
|
|
assert config_result.exit_code == 1
|
|
assert "Traceback" not in config_result.stderr
|
|
vector_result = CliRunner().invoke(
|
|
app, ["vector", "index-schema", "--json", "-c", str(path)]
|
|
)
|
|
assert vector_result.exit_code == 1
|
|
assert vector_result.stderr == ""
|
|
assert json.loads(vector_result.stdout) == {"status": "failed", "code": "invalid_configuration"}
|
|
|
|
|
|
@pytest.mark.parametrize("embeddings", [None, [], "malformed", 7])
|
|
def test_raw_resources_embeddings_non_mapping_is_not_silently_accepted(tmp_path, embeddings):
|
|
values = {
|
|
"dwh": {
|
|
"type": "postgres_direct",
|
|
"connection": {"database": "d", "schema": "public", "user": "u", "password": "p"},
|
|
},
|
|
"resources": {"embeddings": embeddings},
|
|
}
|
|
path = tmp_path / "invalid-resources-embeddings.yaml"
|
|
path.write_text(yaml.safe_dump(values))
|
|
|
|
with pytest.raises(ConfigError, match="resources.embeddings"):
|
|
load_config(path)
|
|
config_result = CliRunner().invoke(app, ["config", "check", "--config", str(path)])
|
|
assert config_result.exit_code == 1
|
|
assert "Traceback" not in config_result.stderr
|
|
vector_result = CliRunner().invoke(
|
|
app, ["vector", "index-schema", "--json", "-c", str(path)]
|
|
)
|
|
assert vector_result.exit_code == 1
|
|
assert vector_result.stderr == ""
|
|
assert json.loads(vector_result.stdout) == {"status": "failed", "code": "invalid_configuration"}
|
|
|
|
|
|
@pytest.mark.parametrize("mutator", [
|
|
lambda values: values.update({1: "not-a-string-key"}),
|
|
lambda values: values["resources"].update({1: {"provider": "bad"}}),
|
|
lambda values: values["resources"].update({"embeddings": {1: "bad"}}),
|
|
])
|
|
def test_raw_non_string_mapping_keys_are_safe_config_errors(tmp_path, mutator):
|
|
values = {
|
|
"dwh": {
|
|
"type": "postgres_direct",
|
|
"connection": {"database": "d", "schema": "public", "user": "u", "password": "p"},
|
|
},
|
|
"resources": {"embeddings": {
|
|
"provider": "ollama_internal", "base_url": "http://embedding:11434",
|
|
"model": "qwen3-embedding:0.6b", "dimensions": 1024,
|
|
}},
|
|
}
|
|
mutator(values)
|
|
path = tmp_path / "invalid-mapping-key.yaml"
|
|
path.write_text(yaml.safe_dump(values))
|
|
|
|
with pytest.raises(ConfigError, match="mapping key"):
|
|
load_config(path)
|
|
config_result = CliRunner().invoke(app, ["config", "check", "--config", str(path)])
|
|
assert config_result.exit_code == 1
|
|
assert "Traceback" not in config_result.stderr
|
|
vector_result = CliRunner().invoke(
|
|
app, ["vector", "index-schema", "--json", "-c", str(path)]
|
|
)
|
|
assert vector_result.exit_code == 1
|
|
assert vector_result.stderr == ""
|
|
assert json.loads(vector_result.stdout) == {"status": "failed", "code": "invalid_configuration"}
|