7dffb76801
- SQLTable: add column_meta(), field_descriptions(), to_ddl(), _resolve_type() - generate_models.py: read DuckDB column comments, emit Field(description=...) - run_pipeline: validate output columns against SQLTable contract (SchemaError) - Pipeline.run: pass output class through to runner - Tag.col(): new namespace for column-level descriptions - bib/meta: collect_column_comments() and apply_column_comments() for Zotero→DuckDB flow - stack.toml + src/conf/: centralised config loader with attribute access and path() - 100% test coverage on all changed files
478 lines
14 KiB
Python
478 lines
14 KiB
Python
"""Tests for bib.meta — docstring ↔ Zotero bidirectional mapping."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import duckdb
|
|
|
|
from bib.meta import (
|
|
apply_column_comments,
|
|
collect_column_comments,
|
|
expr_fn_tag,
|
|
extract_meta_details,
|
|
extract_meta_tags,
|
|
generate_docstring_refs,
|
|
tag_items_from_pipeline,
|
|
)
|
|
from bib.tag import Tag
|
|
|
|
# ── extract_meta_tags ─────────────────────────────────────────────
|
|
|
|
|
|
class TestExtractMetaTags:
|
|
def test_cclf_ip_section(self):
|
|
def fn():
|
|
"""CCLF IP Section 2.2.1 "Part A Claims Header File" (p.8-9):"""
|
|
|
|
tags = extract_meta_tags(fn)
|
|
assert len(tags) == 1
|
|
assert tags[0].label == "meta:cclf-ip-s2.2.1"
|
|
|
|
def test_multiple_cclf_sections(self):
|
|
def fn():
|
|
"""CCLF IP Section 3.1 "Matching MBIs" (p.14):
|
|
|
|
CCLF IP Section 5.1.1 "Creation of MBI field" (p.19):
|
|
"""
|
|
|
|
tags = extract_meta_tags(fn)
|
|
assert len(tags) == 2
|
|
labels = {t.label for t in tags}
|
|
assert labels == {"meta:cclf-ip-s3.1", "meta:cclf-ip-s5.1.1"}
|
|
|
|
def test_cclf_section_no_title(self):
|
|
def fn():
|
|
"""CCLF IP Section 3.1 (p.14):"""
|
|
|
|
tags = extract_meta_tags(fn)
|
|
assert len(tags) == 1
|
|
assert tags[0].label == "meta:cclf-ip-s3.1"
|
|
|
|
def test_cclf_section_no_page(self):
|
|
def fn():
|
|
"""CCLF IP Section 5.3.1"""
|
|
|
|
tags = extract_meta_tags(fn)
|
|
assert len(tags) == 1
|
|
assert tags[0].label == "meta:cclf-ip-s5.3.1"
|
|
|
|
def test_cfr_reference(self):
|
|
def fn():
|
|
"""See 42 CFR § 425.502 for details."""
|
|
|
|
tags = extract_meta_tags(fn)
|
|
assert len(tags) == 1
|
|
assert tags[0].label == "meta:42-cfr-425.502"
|
|
|
|
def test_cfr_without_section_symbol(self):
|
|
def fn():
|
|
"""See 42 CFR 425.502 for details."""
|
|
|
|
tags = extract_meta_tags(fn)
|
|
assert len(tags) == 1
|
|
assert tags[0].label == "meta:42-cfr-425.502"
|
|
|
|
def test_federal_register(self):
|
|
def fn():
|
|
"""Published at 90 FR 86252."""
|
|
|
|
tags = extract_meta_tags(fn)
|
|
assert len(tags) == 1
|
|
assert tags[0].label == "meta:fr-90-86252"
|
|
|
|
def test_mixed_citations(self):
|
|
def fn():
|
|
"""CCLF IP Section 2.2 "Part A" (p.8):
|
|
|
|
See also 42 CFR § 425.502 and 90 FR 86252.
|
|
"""
|
|
|
|
tags = extract_meta_tags(fn)
|
|
assert len(tags) == 3
|
|
labels = {t.label for t in tags}
|
|
assert "meta:cclf-ip-s2.2" in labels
|
|
assert "meta:42-cfr-425.502" in labels
|
|
assert "meta:fr-90-86252" in labels
|
|
|
|
def test_deduplication(self):
|
|
def fn():
|
|
"""CCLF IP Section 2.2 (p.8):
|
|
Another ref to CCLF IP Section 2.2 (p.8):
|
|
"""
|
|
|
|
tags = extract_meta_tags(fn)
|
|
assert len(tags) == 1
|
|
|
|
def test_no_docstring(self):
|
|
fn = lambda: None # noqa: E731
|
|
assert extract_meta_tags(fn) == []
|
|
|
|
def test_no_citations(self):
|
|
def fn():
|
|
"""Just a regular function."""
|
|
|
|
assert extract_meta_tags(fn) == []
|
|
|
|
def test_real_cclf_function(self):
|
|
from aco.express.cclf import stg_beneficiary_xref
|
|
|
|
tags = extract_meta_tags(stg_beneficiary_xref)
|
|
labels = {t.label for t in tags}
|
|
assert "meta:cclf-ip-s3.1" in labels
|
|
assert "meta:cclf-ip-s5.1.1" in labels
|
|
|
|
|
|
# ── extract_meta_details ──────────────────────────────────────────
|
|
|
|
|
|
class TestExtractMetaDetails:
|
|
def test_cclf_ip_details(self):
|
|
def fn():
|
|
"""CCLF IP Section 2.2.1 "Part A Claims Header File" (p.8-9):"""
|
|
|
|
details = extract_meta_details(fn)
|
|
assert len(details) == 1
|
|
d = details[0]
|
|
assert d["tag"] == "meta:cclf-ip-s2.2.1"
|
|
assert d["source"] == "CCLF Information Packet"
|
|
assert d["title"] == "Part A Claims Header File"
|
|
assert d["section"] == "2.2.1"
|
|
assert d["page"] == "8-9"
|
|
|
|
def test_cfr_details(self):
|
|
def fn():
|
|
"""42 CFR § 425.502"""
|
|
|
|
details = extract_meta_details(fn)
|
|
assert len(details) == 1
|
|
assert details[0]["source"] == "Code of Federal Regulations"
|
|
assert details[0]["title"] == "42 CFR § 425.502"
|
|
|
|
def test_cclf_dedup(self):
|
|
def fn():
|
|
"""CCLF IP Section 2.2 (p.8):
|
|
Again CCLF IP Section 2.2 (p.8):"""
|
|
|
|
details = extract_meta_details(fn)
|
|
assert len(details) == 1
|
|
|
|
def test_cfr_dedup(self):
|
|
def fn():
|
|
"""42 CFR § 425.502 and again 42 CFR § 425.502"""
|
|
|
|
details = extract_meta_details(fn)
|
|
assert len(details) == 1
|
|
|
|
def test_fr_details(self):
|
|
def fn():
|
|
"""Published at 90 FR 86252."""
|
|
|
|
details = extract_meta_details(fn)
|
|
assert len(details) == 1
|
|
d = details[0]
|
|
assert d["source"] == "Federal Register"
|
|
assert d["title"] == "90 FR 86252"
|
|
assert d["page"] == "86252"
|
|
assert d["section"] == ""
|
|
|
|
def test_fr_dedup(self):
|
|
def fn():
|
|
"""90 FR 86252 and again 90 FR 86252"""
|
|
|
|
details = extract_meta_details(fn)
|
|
assert len(details) == 1
|
|
|
|
|
|
# ── expr_fn_tag ───────────────────────────────────────────────────
|
|
|
|
|
|
class TestExprFnTag:
|
|
def test_intermediate_table(self):
|
|
tag = expr_fn_tag("cclf._stg_beneficiary_xref")
|
|
assert tag.label == "fn:cclf.stg_beneficiary_xref"
|
|
|
|
def test_output_table(self):
|
|
tag = expr_fn_tag("cclf.medical_claim")
|
|
assert tag.label == "fn:cclf.medical_claim"
|
|
|
|
def test_double_underscore_internal(self):
|
|
tag = expr_fn_tag("readmissions._int_encounter")
|
|
assert tag.label == "fn:readmissions.int_encounter"
|
|
|
|
|
|
# ── Tag factory methods ───────────────────────────────────────────
|
|
|
|
|
|
class TestTagFactories:
|
|
def test_meta_tag(self):
|
|
t = Tag.meta("cclf-ip-s2.2.1")
|
|
assert t.namespace == "meta"
|
|
assert t.value == "cclf-ip-s2.2.1"
|
|
assert t.label == "meta:cclf-ip-s2.2.1"
|
|
|
|
def test_fn_tag(self):
|
|
t = Tag.fn("cclf.stg_beneficiary_xref")
|
|
assert t.namespace == "fn"
|
|
assert t.value == "cclf.stg_beneficiary_xref"
|
|
assert t.label == "fn:cclf.stg_beneficiary_xref"
|
|
|
|
def test_from_label_meta(self):
|
|
t = Tag.from_label("meta:cclf-ip-s2.2.1")
|
|
assert t.namespace == "meta"
|
|
assert t.value == "cclf-ip-s2.2.1"
|
|
|
|
def test_from_label_fn(self):
|
|
t = Tag.from_label("fn:cclf.stg_beneficiary_xref")
|
|
assert t.namespace == "fn"
|
|
assert t.value == "cclf.stg_beneficiary_xref"
|
|
|
|
|
|
# ── tag_items_from_pipeline ───────────────────────────────────────
|
|
|
|
|
|
class TestTagItemsFromPipeline:
|
|
def test_table_tag_fallback(self):
|
|
"""When no refs match, falls back to table: tag lookup."""
|
|
import narwhals as nw
|
|
|
|
from bib.item import Source
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
item = Source(
|
|
title="Core Encounter Spec",
|
|
tags=["table:test.out"],
|
|
)
|
|
key = store.create(item)
|
|
|
|
@nw.narwhalify
|
|
def fn(df):
|
|
"""CCLF IP Section 9.9 "Fake" (p.1):"""
|
|
return df
|
|
|
|
from aco.express.base import Expr
|
|
from aco.pipe.base import Pipeline
|
|
|
|
expr = Expr(name="test.out", fn=fn, refs=[])
|
|
pipeline = Pipeline(exprs=[expr])
|
|
|
|
report = tag_items_from_pipeline(pipeline, store)
|
|
tagged = {k: v for k, v in report.items() if v}
|
|
assert len(tagged) > 0
|
|
|
|
updated = store.get(key)
|
|
fn_tags = [t for t in updated.tags if t.startswith("fn:")]
|
|
assert len(fn_tags) > 0
|
|
|
|
def test_tags_matched_items(self):
|
|
from bib.item import Source
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
item = Source(
|
|
title="CCLF Information Packet",
|
|
tags=["source:cms-cclf-ip", "module:aco"],
|
|
)
|
|
key = store.create(item)
|
|
|
|
from aco.pipe.cclf import pipeline
|
|
|
|
report = tag_items_from_pipeline(pipeline, store)
|
|
|
|
# At least some expressions should have tagged the item
|
|
tagged = {k: v for k, v in report.items() if v}
|
|
assert len(tagged) > 0
|
|
|
|
# Verify the item now has fn: and meta: tags
|
|
updated = store.get(key)
|
|
fn_tags = [t for t in updated.tags if t.startswith("fn:")]
|
|
meta_tags = [t for t in updated.tags if t.startswith("meta:")]
|
|
assert len(fn_tags) > 0
|
|
assert len(meta_tags) > 0
|
|
|
|
|
|
# ── generate_docstring_refs ───────────────────────────────────────
|
|
|
|
|
|
class TestGenerateDocstringRefs:
|
|
def test_no_items(self):
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
refs = generate_docstring_refs("cclf._stg_beneficiary_xref", store)
|
|
assert refs == ""
|
|
|
|
def test_with_tagged_item(self):
|
|
from bib.item import Source
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
item = Source(
|
|
title="CCLF Information Packet v41.0",
|
|
tags=["fn:cclf.stg_beneficiary_xref"],
|
|
institution="CMS",
|
|
date_published="2025-07-16",
|
|
)
|
|
store.create(item)
|
|
|
|
refs = generate_docstring_refs("cclf._stg_beneficiary_xref", store)
|
|
assert "References" in refs
|
|
assert "CCLF Information Packet" in refs
|
|
|
|
|
|
# ── generate_pipeline_bibliography ───────────────────────────────
|
|
|
|
|
|
class TestGeneratePipelineBibliography:
|
|
def test_returns_refs_for_tagged_exprs(self):
|
|
import narwhals as nw
|
|
|
|
from bib.item import Source
|
|
from bib.meta import generate_pipeline_bibliography
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
item = Source(
|
|
title="Test Spec",
|
|
tags=["fn:test.passthrough"],
|
|
institution="CMS",
|
|
date_published="2025-01-01",
|
|
)
|
|
store.create(item)
|
|
|
|
@nw.narwhalify
|
|
def fn(df):
|
|
return df
|
|
|
|
from aco.express.base import Expr
|
|
from aco.pipe.base import Pipeline
|
|
|
|
expr = Expr(name="test._passthrough", fn=fn)
|
|
pipeline = Pipeline(exprs=[expr])
|
|
|
|
result = generate_pipeline_bibliography(pipeline, store)
|
|
assert "test._passthrough" in result
|
|
assert "Test Spec" in result["test._passthrough"]
|
|
|
|
def test_empty_pipeline(self):
|
|
from bib.meta import generate_pipeline_bibliography
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
|
|
from aco.pipe.base import Pipeline
|
|
|
|
pipeline = Pipeline(exprs=[])
|
|
result = generate_pipeline_bibliography(pipeline, store)
|
|
assert result == {}
|
|
|
|
|
|
# ── collect_column_comments ──────────────────────────────────────
|
|
|
|
|
|
class TestCollectColumnComments:
|
|
def test_collects_col_tags(self):
|
|
from bib.item import Source
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
item = Source(title="CCLF Spec")
|
|
key = store.create(item)
|
|
store.add_tag(
|
|
key,
|
|
Tag.col("core.encounter.encounter_id", "Unique encounter ID").label,
|
|
)
|
|
store.add_tag(
|
|
key,
|
|
Tag.col("core.encounter.admit_date", "Admission date").label,
|
|
)
|
|
|
|
comments = collect_column_comments(store)
|
|
assert comments["core.encounter.encounter_id"] == "Unique encounter ID"
|
|
assert comments["core.encounter.admit_date"] == "Admission date"
|
|
|
|
def test_empty_store_returns_empty(self):
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
assert collect_column_comments(store) == {}
|
|
|
|
def test_ignores_malformed_col_tags(self):
|
|
from bib.item import Source
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
item = Source(title="Bad tags")
|
|
key = store.create(item)
|
|
# Tag without = separator
|
|
store.add_tag(key, "col:no-equals-sign")
|
|
|
|
comments = collect_column_comments(store)
|
|
assert comments == {}
|
|
|
|
|
|
# ── apply_column_comments ────────────────────────────────────────
|
|
|
|
|
|
class TestApplyColumnComments:
|
|
def test_applies_comments_to_duckdb(self):
|
|
from bib.item import Source
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
item = Source(title="Spec")
|
|
key = store.create(item)
|
|
store.add_tag(
|
|
key,
|
|
Tag.col("s.t.col_a", "Column A description").label,
|
|
)
|
|
|
|
con = duckdb.connect(":memory:")
|
|
con.execute("CREATE SCHEMA s")
|
|
con.execute("CREATE TABLE s.t (col_a VARCHAR, col_b INTEGER)")
|
|
|
|
stmts = apply_column_comments(con, store)
|
|
assert len(stmts) == 1
|
|
assert "Column A description" in stmts[0]
|
|
|
|
# Verify comment was actually set
|
|
rows = con.execute(
|
|
"SELECT comment FROM duckdb_columns() "
|
|
"WHERE schema_name = 's' AND table_name = 't' "
|
|
"AND column_name = 'col_a'"
|
|
).fetchall()
|
|
assert rows[0][0] == "Column A description"
|
|
con.close()
|
|
|
|
def test_skips_invalid_column_refs(self):
|
|
from bib.item import Source
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
item = Source(title="Bad ref")
|
|
key = store.create(item)
|
|
# Only two parts instead of three
|
|
store.add_tag(key, "col:schema.table=desc")
|
|
|
|
con = duckdb.connect(":memory:")
|
|
stmts = apply_column_comments(con, store)
|
|
assert stmts == []
|
|
con.close()
|
|
|
|
def test_skips_nonexistent_columns(self):
|
|
from bib.item import Source
|
|
from bib.store import Store
|
|
|
|
store = Store(":memory:")
|
|
item = Source(title="Spec")
|
|
key = store.create(item)
|
|
store.add_tag(
|
|
key,
|
|
Tag.col("s.t.nonexistent", "desc").label,
|
|
)
|
|
|
|
con = duckdb.connect(":memory:")
|
|
# Table doesn't exist — should not raise
|
|
stmts = apply_column_comments(con, store)
|
|
assert stmts == []
|
|
con.close()
|