93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
from datetime import datetime
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app
|
|
from tht.mschema.models import ColumnPhysical, PhysicalSchema, TablePhysical
|
|
from tht.config import ExamplesConfig
|
|
from tht.cli.schema_cmd import _add_examples
|
|
|
|
|
|
def _write_catalog(tmp_path):
|
|
phys = PhysicalSchema(
|
|
database="d", schema="s", introspected_at=datetime(2026, 1, 1),
|
|
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_cache_hit_skips_dwh(tmp_path):
|
|
# Le credenziali sono fasulle: se la guardia non scattasse PRIMA del branch
|
|
# transport, il comando tenterebbe la connessione e fallirebbe.
|
|
catalog = _write_catalog(tmp_path)
|
|
before = catalog.read_bytes()
|
|
cfg = _write_config(tmp_path)
|
|
res = CliRunner().invoke(app, ["schema", "introspect", "-c", str(cfg)])
|
|
assert res.exit_code == 0, res.output
|
|
assert "OK (cache)" in res.output
|
|
assert "1 tabelle" in res.output
|
|
assert catalog.read_bytes() == before
|
|
|
|
|
|
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_catalog_guides_fallback(tmp_path):
|
|
cfg = _write_config(tmp_path)
|
|
res = CliRunner().invoke(app, ["schema", "render", "-c", str(cfg)])
|
|
assert res.exit_code == 1
|
|
assert "Esegui prima" 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),
|
|
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
|