109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
from datetime import UTC, datetime
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app
|
|
from tht.cli.schema_cmd import _add_examples
|
|
from tht.config import ExamplesConfig
|
|
from tht.mschema.models import ColumnPhysical, PhysicalSchema, TablePhysical
|
|
|
|
|
|
def _write_catalog(tmp_path):
|
|
phys = PhysicalSchema(
|
|
database="d", schema="s", introspected_at=datetime(2026, 1, 1, tzinfo=UTC),
|
|
tables={
|
|
"dim_patient": TablePhysical(
|
|
comment="Anagrafica",
|
|
columns={
|
|
"cod_paz": ColumnPhysical(type="bigint", pk=True, comment="Codice paziente"),
|
|
},
|
|
)
|
|
},
|
|
)
|
|
out = tmp_path / "artifacts" / "mschema" / "physical.yaml"
|
|
phys.to_yaml(out)
|
|
return out
|
|
|
|
|
|
def _write_config(tmp_path):
|
|
cfg = tmp_path / "workspace.yaml"
|
|
cfg.write_text(
|
|
"database: {database: d, schema: s, user: u, password: p, transport: direct}\n"
|
|
f"paths: {{artifacts: {tmp_path/'artifacts'}, indexes: {tmp_path/'i'}, sessions: {tmp_path/'s'}}}\n"
|
|
)
|
|
return cfg
|
|
|
|
|
|
def test_introspect_rejects_unbound_legacy_cache(tmp_path):
|
|
catalog = _write_catalog(tmp_path)
|
|
cfg = _write_config(tmp_path)
|
|
res = CliRunner().invoke(app, ["schema", "introspect", "-c", str(cfg)])
|
|
assert res.exit_code == 1
|
|
assert "legacy artifacts are unbound" in res.output
|
|
assert catalog.exists()
|
|
|
|
|
|
def test_introspect_fresh_root_initializes_through_writer_job(tmp_path, monkeypatch):
|
|
import tht.cli.schema_cmd as module
|
|
|
|
cfg = _write_config(tmp_path)
|
|
physical = PhysicalSchema(
|
|
database="d", schema="s", introspected_at=datetime(2026, 1, 1, tzinfo=UTC),
|
|
tables={"dim_patient": TablePhysical(columns={"id": ColumnPhysical(type="bigint")})},
|
|
)
|
|
|
|
def refresh(_cfg, *, output_path=None, **_kwargs):
|
|
physical.to_yaml(output_path)
|
|
return physical
|
|
|
|
monkeypatch.setattr(module, "refresh_catalog", refresh)
|
|
res = CliRunner().invoke(app, ["schema", "introspect", "-c", str(cfg)])
|
|
assert res.exit_code == 0, res.output
|
|
assert (tmp_path / ".tht-dwh" / "OWNER.json").is_file()
|
|
assert "1 tabelle" in res.output
|
|
|
|
|
|
def test_introspect_refresh_bypasses_cache(tmp_path):
|
|
_write_catalog(tmp_path)
|
|
cfg = _write_config(tmp_path)
|
|
res = CliRunner().invoke(app, ["schema", "introspect", "-c", str(cfg), "--refresh"])
|
|
assert res.exit_code != 0
|
|
assert "OK (cache)" not in res.output
|
|
|
|
|
|
def test_introspect_corrupt_catalog_falls_through(tmp_path):
|
|
catalog = tmp_path / "artifacts" / "mschema" / "physical.yaml"
|
|
catalog.parent.mkdir(parents=True)
|
|
catalog.write_text("{not: [valid")
|
|
cfg = _write_config(tmp_path)
|
|
res = CliRunner().invoke(app, ["schema", "introspect", "-c", str(cfg)])
|
|
assert res.exit_code != 0
|
|
assert "OK (cache)" not in res.output
|
|
|
|
|
|
def test_render_without_schema_source_fails_closed(tmp_path):
|
|
cfg = _write_config(tmp_path)
|
|
res = CliRunner().invoke(app, ["schema", "render", "-c", str(cfg)])
|
|
assert res.exit_code == 1
|
|
assert "physical schema is missing" in res.output
|
|
|
|
|
|
def test_examples_skip_one_unreadable_column_and_continue(caplog):
|
|
physical = PhysicalSchema(
|
|
database="d", schema="s", introspected_at=datetime(2026, 1, 1, tzinfo=UTC),
|
|
tables={"t": TablePhysical(columns={
|
|
"bad": ColumnPhysical(type="text"), "good": ColumnPhysical(type="text")
|
|
})},
|
|
)
|
|
|
|
class Dwh:
|
|
def sample_column(self, table, column, *, limit):
|
|
if column == "bad":
|
|
raise RuntimeError("denied")
|
|
return ["kept"]
|
|
|
|
_add_examples(Dwh(), physical, ExamplesConfig(max_per_column=3))
|
|
assert physical.tables["t"].columns["bad"].examples == []
|
|
assert physical.tables["t"].columns["good"].examples == ["kept"]
|
|
assert "Campionamento saltato" in caplog.text
|