marcopan
eb3bde90e2
Ports the leaf data-layer modules and validates them: - mschema/ (models, eligibility, merge, render), db/ (connection, sampling, introspect, fetch_ca), rest/client.py -- renamed psdwp3->nsp, verbatim. - L0 (testcontainers, real Postgres): db connection read-only enforcement (psd_ro cannot CREATE/INSERT), introspect against a known schema (tables, columns, types, comments, FKs, enum, composite PK), sampling most-frequent values + truncation reporting. 15 tests, ~4s. - L1 (fake data): rest/client RPC contract (mocked transport -- X-API-Key header, payloads, base_url slash handling, HTTP/network error surfacing), mschema/render 3 formats (markdown, mschema-text, schema-dict) + eligibility rules (wide_text excluded, short_text/numeric/enum/temporal/ boolean eligible, annotation override wins). 25 tests. pyproject registers l0/l2 markers + addopts '-m not l2' (L2 opt-in). Deferred to their dependency-porting tasks: test_rrf.py (search needs vectorstore, B3) and the 11 CLI contract tests (need _guards/session, wired when each command lands). 'Not assumed reliable' now has real teeth for the data layer; CLI/search contracts follow.
101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
"""L1: mschema/render — the 3 serialization formats on a fake PhysicalSchema.
|
|
|
|
Catches port breaks in the render layer (markdown reviewer report, mschema-text
|
|
ThothAI style, schema-dict for AV-SQL). Pure logic, no I/O.
|
|
"""
|
|
from datetime import datetime
|
|
|
|
from nsp.mschema.models import (
|
|
Annotations,
|
|
ColumnAnnotation,
|
|
ColumnPhysical,
|
|
ForeignKey,
|
|
PhysicalSchema,
|
|
TablePhysical,
|
|
)
|
|
from nsp.mschema.render import to_markdown, to_mschema_text, to_schema_dict
|
|
|
|
|
|
def _fake_schema() -> PhysicalSchema:
|
|
return PhysicalSchema(
|
|
database="testdb",
|
|
schema="dw",
|
|
introspected_at=datetime(2025, 1, 1, 0, 0, 0),
|
|
tables={
|
|
"dim_pazienti": TablePhysical(
|
|
comment="Anagrafica pazienti",
|
|
row_count=3,
|
|
columns={
|
|
"id_paziente": ColumnPhysical(type="bigint", nullable=False, pk=True),
|
|
"citta": ColumnPhysical(type="varchar(100)", examples=["Milano", "Bergamo"]),
|
|
"note": ColumnPhysical(type="text", eligible=False,
|
|
eligibility_reason="wide_text"),
|
|
},
|
|
foreign_keys=[
|
|
ForeignKey(columns=["fk_col"], ref_table="other", ref_columns=["id"]),
|
|
],
|
|
),
|
|
},
|
|
)
|
|
|
|
|
|
def test_to_markdown_includes_table_and_marks_ignored():
|
|
md = to_markdown(_fake_schema())
|
|
assert "# Schema dw (testdb)" in md
|
|
assert "## dim_pazienti" in md
|
|
assert "Anagrafica pazienti" in md
|
|
# wide_text column is struck through and labeled
|
|
assert "~~note~~" in md
|
|
assert "wide_text" in md
|
|
# foreign key reported
|
|
assert "other" in md
|
|
|
|
|
|
def test_to_mschema_text_style_and_eligibility_filter():
|
|
txt = to_mschema_text(_fake_schema())
|
|
assert "【Schema】" in txt
|
|
assert "【Foreign keys】" in txt
|
|
assert "CREATE TABLE dim_pazienti (" in txt
|
|
# eligible columns appear (column name lowercase, type uppercased)
|
|
assert "id_paziente BIGINT" in txt
|
|
assert "citta VARCHAR(100)" in txt
|
|
# wide_text column is filtered out: "note" must not appear as a CREATE TABLE column line
|
|
assert " NOTE TEXT" not in txt
|
|
assert "note" not in txt.replace("【", "").lower().split()
|
|
# PK marker
|
|
assert "PRIMARY KEY" in txt
|
|
|
|
|
|
def test_to_schema_dict_avsql_shape():
|
|
d = to_schema_dict(_fake_schema())
|
|
assert "dim_pazienti" in d
|
|
entry = d["dim_pazienti"]
|
|
assert "columns_name" in entry
|
|
assert "columns_type" in entry
|
|
assert "primary_keys" in entry
|
|
assert "foreign_keys" in entry
|
|
assert "table_to_tablefullname" in entry
|
|
assert entry["table_to_tablefullname"] == "dw.dim_pazienti"
|
|
# wide_text column filtered out of columns_name
|
|
assert "id_paziente" in entry["columns_name"]
|
|
assert "note" not in entry["columns_name"]
|
|
assert entry["primary_keys"] == ["id_paziente"]
|
|
|
|
|
|
def test_annotations_override_description_in_render():
|
|
schema = _fake_schema()
|
|
from nsp.mschema.models import TableAnnotation
|
|
ann = Annotations(tables={
|
|
"dim_pazienti": TableAnnotation(columns={
|
|
"citta": ColumnAnnotation(description="Comune di residenza"),
|
|
}),
|
|
})
|
|
md = to_markdown(schema, ann)
|
|
assert "Comune di residenza" in md
|
|
|
|
|
|
def test_render_subset_of_tables():
|
|
txt = to_mschema_text(_fake_schema(), tables=["dim_pazienti"])
|
|
# only the requested table appears
|
|
assert "dim_pazienti" in txt
|