585 lines
21 KiB
Python
585 lines
21 KiB
Python
import json
|
|
import subprocess
|
|
import traceback
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
from pydantic import SecretStr
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.adapters.evidence import HttpManifestEvidenceSource
|
|
from tht.adapters.factory import build_evidence_sources, build_vector_store
|
|
from tht.cli import app
|
|
from tht.config import ConfigError, load_config
|
|
from tht.jobs.dwh_pipeline import config_dwh_binding
|
|
from tht.ports.vector import VectorStoreError
|
|
|
|
SIGNED_CANARY = "SIGNED-CANARY-QUERY"
|
|
ACCESS_CANARY = "ACCESS-CANARY"
|
|
SECRET_CANARY = "SECRET-CANARY"
|
|
TOKEN_CANARY = "TOKEN-CANARY"
|
|
ALL_CANARIES = (SIGNED_CANARY, ACCESS_CANARY, SECRET_CANARY, TOKEN_CANARY)
|
|
|
|
|
|
def raw_runtime(source, *, vector=None):
|
|
value = {
|
|
"dwh": {
|
|
"type": "postgres_direct",
|
|
"connection": {
|
|
"database": "analytics", "schema": "public", "user": "reader",
|
|
"password": "not-a-canary",
|
|
},
|
|
},
|
|
"evidence": {"sources": [source]},
|
|
}
|
|
if vector is not None:
|
|
value["vector"] = vector
|
|
return value
|
|
|
|
|
|
def write_config(tmp_path, source, *, vector=None):
|
|
path = tmp_path / "runtime.yaml"
|
|
path.write_text(yaml.safe_dump(raw_runtime(source, vector=vector)))
|
|
return path
|
|
|
|
|
|
def assert_no_canaries(value):
|
|
text = str(value)
|
|
for canary in ALL_CANARIES:
|
|
assert canary not in text
|
|
|
|
|
|
def test_filesystem_config_does_not_touch_a_declared_source_root(tmp_path):
|
|
missing = tmp_path / "deliberately-missing"
|
|
cfg = load_config(write_config(tmp_path, {"type": "filesystem", "root": str(missing)}))
|
|
|
|
assert cfg.evidence.sources[0].root == missing
|
|
assert cfg.evidence.sources[0].patterns == ["**/*.md"]
|
|
assert cfg.evidence.sources[0].max_bytes == 10 * 1024 * 1024
|
|
assert not missing.exists()
|
|
|
|
|
|
def test_public_http_urls_are_secret_typed_without_adapter_construction(tmp_path):
|
|
cfg = load_config(write_config(tmp_path, {
|
|
"type": "http", "urls": ["https://evidence.example.test/guide.md"],
|
|
}))
|
|
source = cfg.evidence.sources[0]
|
|
|
|
assert isinstance(source.urls[0], SecretStr)
|
|
assert source.transport_urls() == ["https://evidence.example.test/guide.md"]
|
|
assert source.connect_timeout == 5
|
|
assert source.read_timeout == 30
|
|
assert source.max_bytes == 10 * 1024 * 1024
|
|
assert source.max_redirects == 5
|
|
assert source.allow_private_hosts is False
|
|
assert source.max_cache_bytes == 64 * 1024 * 1024
|
|
|
|
|
|
def test_signed_http_file_resolves_in_memory_and_preserves_provenance_order(tmp_path):
|
|
signed = [
|
|
f"https://evidence.example.test/guide.md?token={SIGNED_CANARY}",
|
|
"https://evidence.example.test/runbook.md?signature=second",
|
|
]
|
|
secret_file = tmp_path / "signed-urls.json"
|
|
secret_file.write_text(json.dumps(signed))
|
|
cfg = load_config(write_config(tmp_path, {
|
|
"type": "http",
|
|
"provenance_urls": [
|
|
"https://evidence.example.test/guide.md",
|
|
"https://evidence.example.test/runbook.md",
|
|
],
|
|
"signed_urls_file": str(secret_file),
|
|
"connect_timeout": 7,
|
|
"read_timeout": 41,
|
|
"max_bytes": 1234,
|
|
"max_redirects": 2,
|
|
"allow_private_hosts": True,
|
|
"max_cache_bytes": 5678,
|
|
}))
|
|
source = cfg.evidence.sources[0]
|
|
|
|
assert all(isinstance(url, SecretStr) for url in source.urls)
|
|
assert source.transport_urls() == signed
|
|
assert source.provenance_urls == [
|
|
"https://evidence.example.test/guide.md",
|
|
"https://evidence.example.test/runbook.md",
|
|
]
|
|
assert (source.connect_timeout, source.read_timeout) == (7, 41)
|
|
assert (source.max_bytes, source.max_redirects, source.max_cache_bytes) == (1234, 2, 5678)
|
|
assert source.allow_private_hosts is True
|
|
assert "signed_urls_file" not in repr(source)
|
|
assert_no_canaries(repr(cfg))
|
|
assert_no_canaries(cfg.model_dump_json())
|
|
|
|
adapter = build_evidence_sources(cfg)[0]
|
|
assert isinstance(adapter, HttpManifestEvidenceSource)
|
|
assert_no_canaries(repr(adapter))
|
|
|
|
|
|
def test_signed_http_file_requires_explicit_provenance_urls(tmp_path):
|
|
secret_file = tmp_path / "signed-urls.json"
|
|
secret_file.write_text(json.dumps([
|
|
f"https://evidence.example.test/guide.md?token={SIGNED_CANARY}",
|
|
]))
|
|
path = write_config(tmp_path, {
|
|
"type": "http", "signed_urls_file": str(secret_file),
|
|
})
|
|
|
|
with pytest.raises(ConfigError) as caught:
|
|
load_config(path)
|
|
assert "provenance" in str(caught.value).lower()
|
|
assert_no_canaries(caught.value)
|
|
|
|
|
|
def test_signed_http_file_rejects_explicit_null_provenance(tmp_path):
|
|
secret_file = tmp_path / "signed-urls.json"
|
|
secret_file.write_text(json.dumps(["https://evidence.example.test/guide.md"]))
|
|
path = write_config(tmp_path, {
|
|
"type": "http", "provenance_urls": None, "signed_urls_file": str(secret_file),
|
|
})
|
|
|
|
with pytest.raises(ConfigError) as caught:
|
|
load_config(path)
|
|
assert "provenance" in str(caught.value).lower()
|
|
|
|
|
|
def test_public_http_rejects_inline_query_bearing_transport_urls(tmp_path):
|
|
path = write_config(tmp_path, {
|
|
"type": "http",
|
|
"urls": [f"https://evidence.example.test/guide.md?token={SIGNED_CANARY}"],
|
|
})
|
|
|
|
with pytest.raises(ConfigError) as caught:
|
|
load_config(path)
|
|
assert "evidence.sources.0" in str(caught.value)
|
|
assert_no_canaries(caught.value)
|
|
|
|
|
|
@pytest.mark.parametrize("contents", [
|
|
"{malformed", json.dumps({"url": "https://evidence.example.test/guide.md"}),
|
|
json.dumps([]), json.dumps(["https://evidence.example.test/guide.md", 3]),
|
|
])
|
|
def test_signed_http_rejects_malformed_non_list_empty_or_non_string_files(tmp_path, contents):
|
|
secret_file = tmp_path / "signed-urls.json"
|
|
secret_file.write_text(contents)
|
|
path = write_config(tmp_path, {
|
|
"type": "http",
|
|
"provenance_urls": ["https://evidence.example.test/guide.md"],
|
|
"signed_urls_file": str(secret_file),
|
|
})
|
|
|
|
with pytest.raises(ConfigError) as caught:
|
|
load_config(path)
|
|
assert "signed URL file" in str(caught.value)
|
|
assert_no_canaries(caught.value)
|
|
|
|
|
|
def test_signed_http_rejects_missing_and_oversized_files_without_disclosure(tmp_path):
|
|
missing = tmp_path / "missing.json"
|
|
path = write_config(tmp_path, {
|
|
"type": "http",
|
|
"provenance_urls": ["https://evidence.example.test/guide.md"],
|
|
"signed_urls_file": str(missing),
|
|
})
|
|
with pytest.raises(ConfigError, match="signed URL file"):
|
|
load_config(path)
|
|
|
|
oversized = tmp_path / "oversized.json"
|
|
oversized.write_bytes(b"x" * (1024 * 1024 + 1))
|
|
path = write_config(tmp_path, {
|
|
"type": "http",
|
|
"provenance_urls": ["https://evidence.example.test/guide.md"],
|
|
"signed_urls_file": str(oversized),
|
|
})
|
|
with pytest.raises(ConfigError, match="signed URL file") as caught:
|
|
load_config(path)
|
|
assert_no_canaries(caught.value)
|
|
|
|
|
|
@pytest.mark.parametrize("provenance,signed", [
|
|
(
|
|
["https://evidence.example.test/a.md", "https://evidence.example.test/b.md"],
|
|
["https://evidence.example.test/b.md?sig=1", "https://evidence.example.test/a.md?sig=2"],
|
|
),
|
|
(["https://evidence.example.test/a.md"], [
|
|
"https://evidence.example.test/a.md?sig=1", "https://evidence.example.test/b.md?sig=2",
|
|
]),
|
|
(["https://evidence.example.test/a.md"], ["https://evidence.example.test/b.md"]),
|
|
(["https://evidence.example.test/a.md"], [f"https://user:{SIGNED_CANARY}@evidence.example.test/a.md"]),
|
|
(
|
|
["https://evidence.example.test/a.md", "https://evidence.example.test/a.md"],
|
|
["https://evidence.example.test/a.md?sig=1", "https://evidence.example.test/a.md?sig=2"],
|
|
),
|
|
])
|
|
def test_signed_http_rejects_reordered_extra_mismatch_userinfo_and_duplicate_provenance(
|
|
tmp_path, provenance, signed,
|
|
):
|
|
secret_file = tmp_path / "signed-urls.json"
|
|
secret_file.write_text(json.dumps(signed))
|
|
path = write_config(tmp_path, {
|
|
"type": "http", "provenance_urls": provenance, "signed_urls_file": str(secret_file),
|
|
})
|
|
|
|
with pytest.raises(ConfigError) as caught:
|
|
load_config(path)
|
|
assert "evidence.sources.0" in str(caught.value)
|
|
assert_no_canaries(caught.value)
|
|
|
|
|
|
def test_s3_rejects_inline_credentials_and_never_discloses_them(tmp_path):
|
|
path = write_config(tmp_path, {
|
|
"type": "s3", "bucket": "clinical-evidence",
|
|
"access_key": ACCESS_CANARY,
|
|
"secret_key": SECRET_CANARY,
|
|
"session_token": TOKEN_CANARY,
|
|
})
|
|
|
|
with pytest.raises(ConfigError) as caught:
|
|
load_config(path)
|
|
assert "file" in str(caught.value).lower()
|
|
assert_no_canaries(caught.value)
|
|
assert_no_canaries("".join(traceback.format_exception(caught.value)))
|
|
|
|
|
|
def test_s3_scalar_secret_files_are_bounded(tmp_path):
|
|
access = tmp_path / "oversized-access-key"
|
|
access.write_bytes(b"A" * (64 * 1024 + 1))
|
|
secret = tmp_path / "secret-key"
|
|
secret.write_text("bounded-secret")
|
|
path = write_config(tmp_path, {
|
|
"type": "s3", "bucket": "clinical-evidence",
|
|
"access_key_file": str(access), "secret_key_file": str(secret),
|
|
})
|
|
|
|
with pytest.raises(ConfigError) as caught:
|
|
load_config(path)
|
|
assert "secret file" in str(caught.value).lower()
|
|
assert "bounded-secret" not in str(caught.value)
|
|
|
|
|
|
def test_s3_ambient_and_static_file_credentials_are_secret_typed(tmp_path):
|
|
ambient = load_config(write_config(tmp_path, {
|
|
"type": "s3", "bucket": "clinical-evidence", "prefix": "published/",
|
|
})).evidence.sources[0]
|
|
assert ambient.access_key is None
|
|
assert ambient.secret_key is None
|
|
assert ambient.session_token is None
|
|
assert ambient.max_bytes == 10 * 1024 * 1024
|
|
assert ambient.max_objects == 10_000
|
|
assert ambient.max_pages == 100
|
|
assert ambient.page_size == 1000
|
|
|
|
files = {}
|
|
for name, canary in [
|
|
("access_key", ACCESS_CANARY), ("secret_key", SECRET_CANARY),
|
|
("session_token", TOKEN_CANARY),
|
|
]:
|
|
path = tmp_path / name
|
|
path.write_text(canary)
|
|
files[f"{name}_file"] = str(path)
|
|
cfg = load_config(write_config(tmp_path, {
|
|
"type": "s3", "bucket": "clinical-evidence", "prefix": "published/",
|
|
**files,
|
|
"endpoint_url": "https://s3.example.test",
|
|
"region": "eu-west-1",
|
|
"trusted_endpoint": True,
|
|
"allow_private_endpoint": True,
|
|
"allow_insecure_endpoint": False,
|
|
"max_bytes": 222,
|
|
"max_objects": 33,
|
|
"max_pages": 4,
|
|
"page_size": 5,
|
|
}))
|
|
source = cfg.evidence.sources[0]
|
|
assert all(isinstance(value, SecretStr) for value in (
|
|
source.access_key, source.secret_key, source.session_token,
|
|
))
|
|
assert (source.max_bytes, source.max_objects, source.max_pages, source.page_size) == (222, 33, 4, 5)
|
|
assert_no_canaries(repr(cfg))
|
|
assert_no_canaries(cfg.model_dump_json())
|
|
|
|
|
|
def test_evidence_policy_defaults_non_defaults_and_unknown_keys(tmp_path):
|
|
default = load_config(write_config(tmp_path, {
|
|
"type": "filesystem", "root": str(tmp_path / "missing"),
|
|
}))
|
|
assert default.vector.max_chunk_chars == 4000
|
|
assert default.vector.retain_published_generations == 3
|
|
|
|
explicit = load_config(write_config(tmp_path, {
|
|
"type": "filesystem", "root": str(tmp_path / "missing"),
|
|
"patterns": ["docs/*.md"], "max_bytes": 99,
|
|
}, vector={"max_chunk_chars": 123, "retain_published_generations": 7}))
|
|
assert explicit.evidence.sources[0].patterns == ["docs/*.md"]
|
|
assert explicit.vector.max_chunk_chars == 123
|
|
assert explicit.vector.retain_published_generations == 7
|
|
|
|
for mutation in [
|
|
{"type": "filesystem", "root": str(tmp_path), "unknown": SIGNED_CANARY},
|
|
{"type": "http", "urls": ["https://evidence.example.test/a"], "unknown": SIGNED_CANARY},
|
|
{"type": "s3", "bucket": "bucket-name", "unknown": SIGNED_CANARY},
|
|
]:
|
|
with pytest.raises(ConfigError) as caught:
|
|
load_config(write_config(tmp_path, mutation))
|
|
assert "extra_forbidden" in str(caught.value)
|
|
assert_no_canaries(caught.value)
|
|
|
|
|
|
def test_validation_repr_cli_and_exception_output_never_disclose_transport_secrets(tmp_path):
|
|
secret_file = tmp_path / "signed-urls.json"
|
|
secret_file.write_text(json.dumps([
|
|
f"https://evidence.example.test/other.md?token={SIGNED_CANARY}",
|
|
]))
|
|
path = write_config(tmp_path, {
|
|
"type": "http",
|
|
"provenance_urls": ["https://evidence.example.test/guide.md"],
|
|
"signed_urls_file": str(secret_file),
|
|
})
|
|
with pytest.raises(ConfigError) as caught:
|
|
load_config(path)
|
|
assert_no_canaries(caught.value)
|
|
assert_no_canaries("".join(traceback.format_exception(caught.value)))
|
|
|
|
valid_file = tmp_path / "valid-signed-urls.json"
|
|
valid_file.write_text(json.dumps([
|
|
f"https://evidence.example.test/guide.md?token={SIGNED_CANARY}",
|
|
]))
|
|
valid = write_config(tmp_path, {
|
|
"type": "http",
|
|
"provenance_urls": ["https://evidence.example.test/guide.md"],
|
|
"signed_urls_file": str(valid_file),
|
|
})
|
|
result = CliRunner().invoke(app, ["config", "check", "--config", str(valid)])
|
|
assert result.exit_code == 0
|
|
assert_no_canaries(result.stdout)
|
|
assert_no_canaries(result.stderr)
|
|
|
|
|
|
def _render_registry_runtime_config(tmp_path) -> str:
|
|
"""Render the production schema-v3 runtime rather than duplicating its YAML."""
|
|
tmp_path.mkdir(parents=True, exist_ok=True)
|
|
backend = Path(__file__).resolve().parents[2] / "backend"
|
|
tsx = backend / "node_modules/.bin/tsx"
|
|
if not tsx.exists():
|
|
pytest.skip("cross-runtime integration requires backend/node_modules/.bin/tsx")
|
|
password_file = tmp_path / "dwh-password"
|
|
password_file.write_text("not-a-canary")
|
|
script = r"""
|
|
import { parseWorkspaceYaml } from "__SCHEMA__";
|
|
import { renderRuntimeConfig } from "__RENDERER__";
|
|
const workspace = parseWorkspaceYaml(`workspace:
|
|
schema_version: 3
|
|
id: psd-clinical
|
|
name: Runtime test
|
|
language: en
|
|
dwh:
|
|
engine: postgres
|
|
database: analytics
|
|
schema: mart
|
|
supported_transports: [postgres_direct]
|
|
semantic_index:
|
|
vector_store:
|
|
engine: qdrant
|
|
collection: workspace-semantic
|
|
dimensions: 1024
|
|
distance: cosine
|
|
embedding:
|
|
provider: ollama_internal
|
|
model: qwen3-embedding:0.6b
|
|
dimensions: 1024
|
|
llm_policy:
|
|
allowed: [zai/glm-5.2]
|
|
`);
|
|
const bindings = {
|
|
dwh: { transport: "postgres_direct", missing: [], values: {
|
|
THT_WS_PSD_CLINICAL_DWH_HOST: "dwh.internal",
|
|
THT_WS_PSD_CLINICAL_DWH_PORT: "5432",
|
|
THT_WS_PSD_CLINICAL_DWH_USER: "reader",
|
|
THT_WS_PSD_CLINICAL_DWH_PASSWORD_FILE: process.argv[2],
|
|
} },
|
|
evidence: { missing: [], values: {} },
|
|
};
|
|
process.stdout.write(renderRuntimeConfig(workspace, bindings, {
|
|
sessions: "/tmp/sessions", artifacts: "/tmp/artifacts", indexes: "/tmp/indexes",
|
|
}, { workspaceId: "psd-clinical", workspaceRevision: "a".repeat(40) }));
|
|
"""
|
|
script = script.replace(
|
|
"__SCHEMA__", str(backend / "src/workspaces/schema.ts"),
|
|
).replace("__RENDERER__", str(backend / "src/workspaces/runtime-renderer.ts"))
|
|
script_path = tmp_path / "render-runtime.mts"
|
|
script_path.write_text(script)
|
|
result = subprocess.run(
|
|
[str(tsx), str(script_path), str(password_file)],
|
|
cwd=backend, check=True, capture_output=True, text=True,
|
|
)
|
|
return result.stdout
|
|
|
|
|
|
def _render_registry_runtime_configs(tmp_path) -> tuple[Path, Path]:
|
|
session = tmp_path / "session.yaml"
|
|
maintenance = tmp_path / "maintenance.yaml"
|
|
session.write_text(_render_registry_runtime_config(tmp_path))
|
|
maintenance.write_text(_render_registry_runtime_config(tmp_path))
|
|
return session, maintenance
|
|
|
|
|
|
def _qdrant_config_yaml(tmp_path, *, registry: bool) -> dict:
|
|
value = {
|
|
"dwh": {
|
|
"type": "postgres_direct",
|
|
"connection": {
|
|
"database": "analytics", "schema": "public", "user": "reader",
|
|
"password": "not-a-canary",
|
|
},
|
|
},
|
|
"vectors": {
|
|
"type": "qdrant", "base_url": "http://localhost:6333",
|
|
"collection": "workspace-semantic",
|
|
},
|
|
"embeddings": {
|
|
"base_url": "http://localhost:11434", "model": "qwen3-embedding:0.6b", "dim": 1024,
|
|
},
|
|
"evidence": {"source_root": str(tmp_path / "evidence")},
|
|
}
|
|
if registry:
|
|
value["runtime_identity"] = {
|
|
"workspace_id": "demo-workspace", "workspace_revision": "a" * 40,
|
|
}
|
|
return value
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.parametrize("collection_state", ["missing", "incompatible"])
|
|
def test_registry_render_chain_produces_require_existing_qdrant_configs(
|
|
tmp_path, monkeypatch, collection_state,
|
|
):
|
|
session_path, maintenance_path = _render_registry_runtime_configs(tmp_path)
|
|
session_cfg = load_config(session_path)
|
|
maintenance_cfg = load_config(maintenance_path)
|
|
|
|
assert session_cfg.vectors.collection_lifecycle == "require_existing"
|
|
assert maintenance_cfg.vectors.collection_lifecycle == "require_existing"
|
|
assert yaml.safe_load(session_path.read_text())["vectors"]["collection_lifecycle"] == "require_existing"
|
|
assert yaml.safe_load(maintenance_path.read_text())["vectors"]["collection_lifecycle"] == "require_existing"
|
|
assert session_path.read_bytes() == maintenance_path.read_bytes()
|
|
|
|
from qdrant_test_helpers import FakeQdrantHttp, _write_record
|
|
|
|
for cfg in (session_cfg, maintenance_cfg):
|
|
fake = FakeQdrantHttp(dimension=384) if collection_state == "incompatible" else FakeQdrantHttp()
|
|
if collection_state == "incompatible":
|
|
fake.collection = {"vectors": {"size": 384, "distance": "Cosine"}}
|
|
monkeypatch.setattr("requests.request", fake.request)
|
|
store = build_vector_store(cfg, require_write=True)
|
|
with pytest.raises(VectorStoreError, match="semantic_index_incompatible"):
|
|
store.upsert("memory", [_write_record("memory:1", "memory")])
|
|
assert not [call for call in fake.calls if call[0] == "PUT"]
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_registry_rendered_session_and_maintenance_configs_bind_equally(tmp_path):
|
|
session_path, maintenance_path = _render_registry_runtime_configs(tmp_path)
|
|
|
|
assert config_dwh_binding(load_config(session_path)) == config_dwh_binding(
|
|
load_config(maintenance_path)
|
|
)
|
|
|
|
|
|
def test_loader_rejects_split_brain_qdrant_compatibility_resources(tmp_path):
|
|
values = _qdrant_config_yaml(tmp_path, registry=True)
|
|
values["vectors"]["collection"] = "workspace-a"
|
|
values["resources"] = {
|
|
"vector": {
|
|
"engine": "qdrant",
|
|
"base_url": "http://qdrant:6333/",
|
|
"collection": "workspace-b",
|
|
}
|
|
}
|
|
path = tmp_path / "split-brain.yaml"
|
|
path.write_text(yaml.safe_dump(values))
|
|
|
|
with pytest.raises(ConfigError, match="vectors.*resources.vector|disagree"):
|
|
load_config(path)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value"),
|
|
[
|
|
("base_url", {}),
|
|
("base_url", []),
|
|
("base_url", None),
|
|
("base_url", 6333),
|
|
("collection", {}),
|
|
("collection", []),
|
|
("collection", None),
|
|
("collection", 7),
|
|
],
|
|
)
|
|
def test_raw_qdrant_consistency_rejects_untrusted_top_level_values(tmp_path, field, value):
|
|
values = _qdrant_config_yaml(tmp_path, registry=True)
|
|
values["vectors"][field] = value
|
|
values["resources"] = {
|
|
"vector": {
|
|
"engine": "qdrant",
|
|
"base_url": "http://localhost:6333",
|
|
"collection": "workspace-semantic",
|
|
}
|
|
}
|
|
path = tmp_path / "invalid-qdrant.yaml"
|
|
path.write_text(yaml.safe_dump(values))
|
|
|
|
with pytest.raises(ConfigError):
|
|
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"
|
|
}
|
|
|
|
|
|
def test_loader_rejects_lifecycle_policy_in_compatibility_resource(tmp_path):
|
|
values = _qdrant_config_yaml(tmp_path, registry=True)
|
|
values["resources"] = {
|
|
"vector": {
|
|
"engine": "qdrant",
|
|
"base_url": "http://qdrant:6333",
|
|
"collection": "workspace-semantic",
|
|
"collection_lifecycle": "require_existing",
|
|
}
|
|
}
|
|
path = tmp_path / "lifecycle.yaml"
|
|
path.write_text(yaml.safe_dump(values))
|
|
|
|
with pytest.raises(ConfigError, match="collection_lifecycle"):
|
|
load_config(path)
|
|
|
|
|
|
def test_legacy_runtime_config_keeps_create_capable_qdrant_default(tmp_path):
|
|
path = tmp_path / "legacy.yaml"
|
|
path.write_text(yaml.safe_dump(_qdrant_config_yaml(tmp_path, registry=False)))
|
|
|
|
cfg = load_config(path)
|
|
|
|
assert cfg.vectors.collection_lifecycle == "create_if_missing"
|
|
|
|
|
|
def test_registry_session_and_maintenance_bindings_are_equal(tmp_path):
|
|
values = _qdrant_config_yaml(tmp_path, registry=True)
|
|
session_path = tmp_path / "session.yaml"
|
|
maintenance_path = tmp_path / "maintenance.yaml"
|
|
session_path.write_text(yaml.safe_dump(values))
|
|
maintenance_path.write_text(yaml.safe_dump(values))
|
|
|
|
assert config_dwh_binding(load_config(session_path)) == config_dwh_binding(
|
|
load_config(maintenance_path)
|
|
)
|