459 lines
13 KiB
Python
459 lines
13 KiB
Python
import pytest
|
|
|
|
from tht.config import (
|
|
ConfigError,
|
|
PgvectorDirectConfig,
|
|
PostgresDwhConfig,
|
|
QdrantConfig,
|
|
ThothRestDwhConfig,
|
|
ThothVectorHttpConfig,
|
|
load_config,
|
|
workspace_id_for_config,
|
|
)
|
|
from tht.evidence import build_sources
|
|
from tht.evidence.adapters import FilesystemEvidenceSource, HttpManifestEvidenceSource
|
|
|
|
|
|
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_catalog_selected_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
|
|
id: ollama/bge-m3
|
|
model: bge-m3
|
|
dimensions: 1536
|
|
"""
|
|
)
|
|
|
|
cfg = load_config(workspace)
|
|
|
|
assert cfg.embeddings.provider == "ollama_internal"
|
|
assert cfg.embeddings.base_url == "http://embedding:11434"
|
|
assert cfg.embeddings.id == "ollama/bge-m3"
|
|
assert cfg.embeddings.model == "bge-m3"
|
|
assert cfg.embeddings.dim == 1536
|
|
|
|
|
|
def test_installation_embedding_projection_completes_model_free_runtime_source(
|
|
tmp_path, monkeypatch,
|
|
):
|
|
monkeypatch.setenv("THT_INTERNAL_EMBEDDING_ID", "ollama/bge-m3")
|
|
monkeypatch.setenv("THT_INTERNAL_EMBEDDING_MODEL", "bge-m3")
|
|
monkeypatch.setenv("THT_INTERNAL_EMBEDDING_DIMENSIONS", "1536")
|
|
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
|
|
"""
|
|
)
|
|
|
|
cfg = load_config(workspace)
|
|
|
|
assert cfg.embeddings.id == "ollama/bge-m3"
|
|
assert cfg.embeddings.model == "bge-m3"
|
|
assert cfg.embeddings.dim == 1536
|
|
|
|
|
|
def test_workspace_embedding_values_cannot_override_installation_projection(
|
|
tmp_path, monkeypatch,
|
|
):
|
|
monkeypatch.setenv("THT_INTERNAL_EMBEDDING_ID", "ollama/bge-m3")
|
|
monkeypatch.setenv("THT_INTERNAL_EMBEDDING_MODEL", "bge-m3")
|
|
monkeypatch.setenv("THT_INTERNAL_EMBEDDING_DIMENSIONS", "1536")
|
|
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: workspace-owned-model
|
|
"""
|
|
)
|
|
|
|
with pytest.raises(ConfigError, match="proprietà dell'installazione|diverge"):
|
|
load_config(workspace)
|
|
|
|
|
|
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.collections == {
|
|
"reference": "psd-clinical",
|
|
"memory": "psd-clinical-memory",
|
|
}
|
|
|
|
|
|
@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_sources(cfg.evidence)
|
|
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_sources(load_config(legacy).evidence)[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
|