Codex
076c9742c5
Add catalog-owned logical relationships and runtime snapshots, extend the database-management UI and validation coverage, and document the updated operational workflow. Keep active sensitive-generation status in a tooltip and indicator, and update the layout E2E to follow the history action in its new database-scoped location.
404 lines
13 KiB
Python
404 lines
13 KiB
Python
import json
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app
|
|
from tht.config import load_config
|
|
from tht.mschema.context import SchemaContextError, load_schema_context
|
|
from tht.mschema.models import (
|
|
Annotations,
|
|
ColumnPhysical,
|
|
ForeignKey,
|
|
PhysicalSchema,
|
|
TableAnnotation,
|
|
TablePhysical,
|
|
)
|
|
from tht.mschema.render import to_mschema_text
|
|
|
|
|
|
REVISION = "a" * 40
|
|
RUNNER = CliRunner()
|
|
|
|
|
|
def _workspace(tmp_path, snapshot: dict) -> object:
|
|
artifacts = tmp_path / "artifacts"
|
|
relationships = tmp_path / "effective-relationships.json"
|
|
relationships.write_text(json.dumps(snapshot))
|
|
physical = PhysicalSchema(
|
|
database="warehouse",
|
|
schema="analytics",
|
|
introspected_at=datetime(2026, 1, 1, tzinfo=UTC),
|
|
tables={
|
|
"users": TablePhysical(
|
|
columns={"id": ColumnPhysical(type="bigint", pk=True)},
|
|
),
|
|
"orders": TablePhysical(
|
|
columns={
|
|
"physical_user_id": ColumnPhysical(type="bigint"),
|
|
"annotated_user_id": ColumnPhysical(type="bigint"),
|
|
"generated_user_id": ColumnPhysical(type="bigint"),
|
|
},
|
|
foreign_keys=[
|
|
ForeignKey(
|
|
columns=["physical_user_id"],
|
|
ref_table="users",
|
|
ref_columns=["id"],
|
|
)
|
|
],
|
|
),
|
|
},
|
|
)
|
|
physical.to_yaml(artifacts / "mschema" / "physical.yaml")
|
|
Annotations(
|
|
tables={
|
|
"orders": TableAnnotation(
|
|
description="Customer orders",
|
|
foreign_keys=[
|
|
ForeignKey(
|
|
columns=["annotated_user_id"],
|
|
ref_table="users",
|
|
ref_columns=["id"],
|
|
)
|
|
],
|
|
)
|
|
}
|
|
).to_yaml(artifacts / "mschema" / "annotations.yaml")
|
|
config = tmp_path / "runtime.yaml"
|
|
config.write_text(
|
|
f"""
|
|
runtime_identity:
|
|
workspace_id: demo
|
|
workspace_revision: {REVISION}
|
|
source_identity: workspace://demo
|
|
database: {{database: warehouse, schema: analytics, user: reader, password: secret, transport: direct}}
|
|
vector_db: {{database: vectors, schema: public, user: reader, password: secret}}
|
|
embeddings: {{base_url: 'http://localhost:11434', model: 'qwen3-embedding:0.6b', dim: 1024}}
|
|
paths:
|
|
artifacts: {artifacts}
|
|
indexes: {tmp_path / 'indexes'}
|
|
sessions: {tmp_path / 'sessions'}
|
|
effective_relationships: {relationships}
|
|
"""
|
|
)
|
|
return load_config(config)
|
|
|
|
|
|
def test_effective_snapshot_is_the_exclusive_relationship_source(tmp_path):
|
|
cfg = _workspace(
|
|
tmp_path,
|
|
{
|
|
"schemaVersion": 1,
|
|
"workspaceId": "demo",
|
|
"relationships": [
|
|
{
|
|
"sourceTable": "orders",
|
|
"sourceColumns": ["generated_user_id"],
|
|
"targetTable": "users",
|
|
"targetColumns": ["id"],
|
|
"origin": "generated",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
context = load_schema_context(cfg)
|
|
rendered = to_mschema_text(
|
|
context.physical,
|
|
context.annotations,
|
|
effective_relationships=context.effective_relationships,
|
|
)
|
|
|
|
assert "-- Customer orders" in rendered
|
|
assert "orders.generated_user_id=users.id" in rendered
|
|
assert "orders.physical_user_id=users.id" not in rendered
|
|
assert "orders.annotated_user_id=users.id" not in rendered
|
|
|
|
|
|
def test_declared_effective_snapshot_must_exist(tmp_path):
|
|
cfg = _workspace(
|
|
tmp_path,
|
|
{"schemaVersion": 1, "workspaceId": "demo", "relationships": []},
|
|
)
|
|
cfg.paths.effective_relationships.unlink()
|
|
|
|
with pytest.raises(SchemaContextError, match="snapshot is missing"):
|
|
load_schema_context(cfg)
|
|
|
|
|
|
def test_data_root_override_preserves_the_runtime_relationship_snapshot_path(
|
|
tmp_path, monkeypatch
|
|
):
|
|
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "mounted-data"))
|
|
monkeypatch.delenv("THT_HOME", raising=False)
|
|
|
|
cfg = _workspace(
|
|
tmp_path,
|
|
{"schemaVersion": 1, "workspaceId": "demo", "relationships": []},
|
|
)
|
|
|
|
assert cfg.paths.effective_relationships == tmp_path / "effective-relationships.json"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("snapshot", "message"),
|
|
[
|
|
(
|
|
{"schemaVersion": 1, "workspaceId": "another", "relationships": []},
|
|
"workspace does not match",
|
|
),
|
|
(
|
|
{
|
|
"schemaVersion": 1,
|
|
"workspaceId": "demo",
|
|
"relationships": [
|
|
{
|
|
"sourceTable": "missing_orders",
|
|
"sourceColumns": ["user_id"],
|
|
"targetTable": "users",
|
|
"targetColumns": ["id"],
|
|
"origin": "generated",
|
|
}
|
|
],
|
|
},
|
|
"endpoint table is absent",
|
|
),
|
|
(
|
|
{
|
|
"schemaVersion": 1,
|
|
"workspaceId": "demo",
|
|
"relationships": [
|
|
{
|
|
"sourceTable": "orders",
|
|
"sourceColumns": ["missing_user_id"],
|
|
"targetTable": "users",
|
|
"targetColumns": ["id"],
|
|
"origin": "manual",
|
|
}
|
|
],
|
|
},
|
|
"endpoint column is absent",
|
|
),
|
|
(
|
|
{
|
|
"schemaVersion": 1,
|
|
"workspaceId": "demo",
|
|
"relationships": [
|
|
{
|
|
"sourceTable": "orders",
|
|
"sourceColumns": ["generated_user_id", "annotated_user_id"],
|
|
"targetTable": "users",
|
|
"targetColumns": ["id"],
|
|
"origin": "physical",
|
|
}
|
|
],
|
|
},
|
|
"snapshot is invalid",
|
|
),
|
|
],
|
|
)
|
|
def test_effective_snapshot_rejects_wrong_workspace_and_orphan_endpoints(
|
|
tmp_path, snapshot, message
|
|
):
|
|
cfg = _workspace(tmp_path, snapshot)
|
|
|
|
with pytest.raises(SchemaContextError, match=message):
|
|
load_schema_context(cfg)
|
|
|
|
|
|
def test_legacy_relationship_merge_remains_available_and_deduplicated(tmp_path):
|
|
cfg = _workspace(
|
|
tmp_path,
|
|
{"schemaVersion": 1, "workspaceId": "demo", "relationships": []},
|
|
)
|
|
cfg.paths.effective_relationships = None
|
|
annotations = Annotations.from_yaml(
|
|
cfg.paths.artifacts / "mschema" / "annotations.yaml"
|
|
)
|
|
annotations.tables["orders"].foreign_keys.append(
|
|
ForeignKey(
|
|
columns=["annotated_user_id"],
|
|
ref_table="users",
|
|
ref_columns=["id"],
|
|
)
|
|
)
|
|
annotations.to_yaml(cfg.paths.artifacts / "mschema" / "annotations.yaml")
|
|
|
|
context = load_schema_context(cfg)
|
|
rendered = to_mschema_text(context.physical, context.annotations)
|
|
|
|
assert context.effective_relationships is None
|
|
assert rendered.count("orders.physical_user_id=users.id") == 1
|
|
assert rendered.count("orders.annotated_user_id=users.id") == 1
|
|
|
|
|
|
def test_schema_render_uses_the_effective_snapshot(tmp_path):
|
|
_workspace(
|
|
tmp_path,
|
|
{
|
|
"schemaVersion": 1,
|
|
"workspaceId": "demo",
|
|
"relationships": [
|
|
{
|
|
"sourceTable": "orders",
|
|
"sourceColumns": ["generated_user_id"],
|
|
"targetTable": "users",
|
|
"targetColumns": ["id"],
|
|
"origin": "manual",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
result = RUNNER.invoke(
|
|
app,
|
|
["schema", "render", "--format", "mschema-text", "-c", str(tmp_path / "runtime.yaml")],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert "orders.generated_user_id=users.id" in result.stdout
|
|
assert "orders.physical_user_id=users.id" not in result.stdout
|
|
assert "orders.annotated_user_id=users.id" not in result.stdout
|
|
|
|
|
|
def test_schema_search_uses_the_effective_snapshot(tmp_path, monkeypatch):
|
|
cfg = _workspace(
|
|
tmp_path,
|
|
{
|
|
"schemaVersion": 1,
|
|
"workspaceId": "demo",
|
|
"relationships": [
|
|
{
|
|
"sourceTable": "orders",
|
|
"sourceColumns": ["generated_user_id"],
|
|
"targetTable": "users",
|
|
"targetColumns": ["id"],
|
|
"origin": "generated",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
physical = cfg.paths.artifacts / "mschema" / "physical.yaml"
|
|
|
|
class Searcher:
|
|
def search(self, _vector, **_kwargs):
|
|
return [
|
|
SimpleNamespace(
|
|
id="orders",
|
|
kind="schema_table",
|
|
ref="orders",
|
|
title="orders",
|
|
similarity=1.0,
|
|
content="orders",
|
|
metadata={},
|
|
)
|
|
]
|
|
|
|
class Embedder:
|
|
def embed_query(self, _text):
|
|
return [1.0, 0.0, 0.0]
|
|
|
|
import tht.cli.search_cmd as search_module
|
|
import tht.cli.vector_cmd as vector_module
|
|
import tht.evidence as evidence_module
|
|
|
|
monkeypatch.setattr(
|
|
search_module,
|
|
"_leased_dwh_snapshot",
|
|
lambda _cfg, _ctx: SimpleNamespace(physical=physical, lsh_dir=tmp_path / "no-lsh"),
|
|
)
|
|
monkeypatch.setattr(vector_module, "require_vector_cfg", lambda _cfg: None)
|
|
monkeypatch.setattr(vector_module, "open_searcher", lambda _cfg: Searcher())
|
|
monkeypatch.setattr(vector_module, "make_embedder", lambda _cfg: Embedder())
|
|
monkeypatch.setattr(evidence_module, "validate_corpus_workspace", lambda *_args: None)
|
|
monkeypatch.setattr(evidence_module, "active_searcher", lambda _cfg, searcher, **_kwargs: searcher)
|
|
|
|
result = RUNNER.invoke(
|
|
app,
|
|
[
|
|
"search",
|
|
"find",
|
|
"orders",
|
|
"--kind",
|
|
"schema",
|
|
"--json",
|
|
"-c",
|
|
str(tmp_path / "runtime.yaml"),
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
payload = json.loads(result.stdout)
|
|
assert "orders.generated_user_id=users.id" in payload["mschema"]
|
|
assert "orders.physical_user_id=users.id" not in payload["mschema"]
|
|
assert "orders.annotated_user_id=users.id" not in payload["mschema"]
|
|
|
|
|
|
def test_schema_search_fails_closed_before_returning_no_candidates(tmp_path, monkeypatch):
|
|
cfg = _workspace(
|
|
tmp_path,
|
|
{"schemaVersion": 1, "workspaceId": "demo", "relationships": []},
|
|
)
|
|
cfg.paths.effective_relationships.unlink()
|
|
physical = cfg.paths.artifacts / "mschema" / "physical.yaml"
|
|
|
|
import tht.cli.search_cmd as search_module
|
|
import tht.cli.vector_cmd as vector_module
|
|
import tht.evidence as evidence_module
|
|
import tht.search as search_core
|
|
|
|
monkeypatch.setattr(
|
|
search_module,
|
|
"_leased_dwh_snapshot",
|
|
lambda _cfg, _ctx: SimpleNamespace(physical=physical, lsh_dir=tmp_path / "no-lsh"),
|
|
)
|
|
monkeypatch.setattr(vector_module, "require_vector_cfg", lambda _cfg: None)
|
|
monkeypatch.setattr(vector_module, "open_searcher", lambda _cfg: object())
|
|
monkeypatch.setattr(vector_module, "make_embedder", lambda _cfg: object())
|
|
monkeypatch.setattr(evidence_module, "validate_corpus_workspace", lambda *_args: None)
|
|
monkeypatch.setattr(evidence_module, "active_searcher", lambda _cfg, searcher, **_kwargs: searcher)
|
|
monkeypatch.setattr(search_core, "combined_search", lambda **_kwargs: [])
|
|
|
|
result = RUNNER.invoke(
|
|
app,
|
|
[
|
|
"search",
|
|
"find",
|
|
"orders",
|
|
"--kind",
|
|
"schema",
|
|
"--json",
|
|
"-c",
|
|
str(tmp_path / "runtime.yaml"),
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 1
|
|
assert "effective relationship snapshot is missing" in result.stderr
|
|
|
|
|
|
def test_suggest_fks_cannot_write_when_relationships_are_catalog_managed(tmp_path):
|
|
_workspace(
|
|
tmp_path,
|
|
{"schemaVersion": 1, "workspaceId": "demo", "relationships": []},
|
|
)
|
|
annotations = tmp_path / "artifacts" / "mschema" / "annotations.yaml"
|
|
before = annotations.read_text()
|
|
|
|
result = RUNNER.invoke(
|
|
app,
|
|
[
|
|
"schema",
|
|
"suggest-fks",
|
|
"--write",
|
|
"-c",
|
|
str(tmp_path / "runtime.yaml"),
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 1
|
|
assert "managed by the catalog" in result.stderr
|
|
assert annotations.read_text() == before
|