dbf71a6594
Fix Zotero table models for Zotero 9:
- Remove stale Annotations/Highlights/Transaction* models
- Add ItemAnnotations, RetractedItems, DeletedCollections,
DeletedSearches, DbDebug1
- Fix ItemAttachments, Libraries, Users column mismatches
New test files covering all major modules:
- cli/{bib,prisma,rec,zot,mail,run} deep exercising tests
- mail/{droplet,postmark,resend,cloudflare} lifecycle tests
- bib/{iom,oig,pincite,sync,regulations_gov,email_ingest,format,store}
- prisma/{vpn,fetch,export,llm,screen,eligibility,extract,project,ingest,flow}
- aco/lake/{unity,quality,deploy} + api/aco coverage gaps
- zot/{ops,db,extract,duck} + rec/{report,engine,base,pricers}
- pfs/{pipe,rules,eq,files}
Add pytest-xdist for parallel test execution.
Tracks #353
1163 lines
38 KiB
Python
1163 lines
38 KiB
Python
"""Tests for pfs.eq — equation tree construction, introspection, evaluation, and rendering."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from bib.tag import Tag
|
|
from pfs.eq import (
|
|
# Pre-built terms
|
|
CF,
|
|
DIRECT_PE,
|
|
EQUIPMENT_COST,
|
|
F_PE_RVU,
|
|
INTENSITY,
|
|
LABOR_COST,
|
|
MP_GPCI,
|
|
MP_RVU,
|
|
MPPR,
|
|
NF_PE_RVU,
|
|
PE_GPCI,
|
|
PE_RVU,
|
|
PFS_PAYMENT,
|
|
SUPPLY_COST,
|
|
WORK_GPCI,
|
|
WORK_RVU,
|
|
Add,
|
|
Cond,
|
|
Dot,
|
|
Fn,
|
|
Mul,
|
|
Ref,
|
|
Scalar,
|
|
Term,
|
|
Vector,
|
|
# Introspection
|
|
citations,
|
|
depth,
|
|
evaluate,
|
|
refs,
|
|
render,
|
|
scalars,
|
|
sources,
|
|
walk,
|
|
)
|
|
|
|
# ── Term construction ─────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestTermBase:
|
|
def test_base_term_construction(self) -> None:
|
|
t = Term(label="X")
|
|
assert t.label == "X"
|
|
|
|
def test_base_term_doc_defaults_empty(self) -> None:
|
|
t = Term(label="X")
|
|
assert t.doc == ""
|
|
|
|
def test_base_term_cite_defaults_empty(self) -> None:
|
|
t = Term(label="X")
|
|
assert t.cite == ()
|
|
|
|
def test_base_term_children_empty(self) -> None:
|
|
t = Term(label="X")
|
|
assert t.children == ()
|
|
|
|
def test_base_term_with_doc(self) -> None:
|
|
t = Term(label="X", doc="some doc")
|
|
assert t.doc == "some doc"
|
|
|
|
def test_base_term_with_cite(self) -> None:
|
|
tag = Tag.module("pfs")
|
|
t = Term(label="X", cite=(tag,))
|
|
assert len(t.cite) == 1
|
|
assert t.cite[0] == tag
|
|
|
|
def test_term_is_frozen(self) -> None:
|
|
t = Term(label="X")
|
|
with pytest.raises((AttributeError, TypeError)):
|
|
t.label = "Y" # type: ignore[misc]
|
|
|
|
|
|
class TestScalar:
|
|
def test_scalar_basic(self) -> None:
|
|
s = Scalar("CF", value=32.3465, unit="$/RVU")
|
|
assert s.label == "CF"
|
|
assert s.value == pytest.approx(32.3465)
|
|
assert s.unit == "$/RVU"
|
|
|
|
def test_scalar_no_value(self) -> None:
|
|
s = Scalar("CF")
|
|
assert s.value is None
|
|
|
|
def test_scalar_children_empty(self) -> None:
|
|
s = Scalar("CF", value=1.0)
|
|
assert s.children == ()
|
|
|
|
def test_scalar_unit_defaults_empty(self) -> None:
|
|
s = Scalar("X")
|
|
assert s.unit == ""
|
|
|
|
def test_scalar_with_cite(self) -> None:
|
|
tag = Tag.rule("cms-1807-f")
|
|
s = Scalar("CF", value=32.3465, cite=(tag,))
|
|
assert tag in s.cite
|
|
|
|
|
|
class TestRef:
|
|
def test_ref_basic(self) -> None:
|
|
r = Ref("Work_RVU", table="pfs.rvu", column="work_rvu")
|
|
assert r.label == "Work_RVU"
|
|
assert r.table == "pfs.rvu"
|
|
assert r.column == "work_rvu"
|
|
|
|
def test_ref_defaults(self) -> None:
|
|
r = Ref("X")
|
|
assert r.table == ""
|
|
assert r.column == ""
|
|
|
|
def test_ref_children_empty(self) -> None:
|
|
r = Ref("Work_RVU", table="pfs.rvu", column="work_rvu")
|
|
assert r.children == ()
|
|
|
|
|
|
class TestVector:
|
|
def test_vector_with_terms(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
v = Vector("V", terms=(a, b))
|
|
assert len(v.terms) == 2
|
|
|
|
def test_vector_children_are_terms(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Ref("B", table="t", column="c")
|
|
v = Vector("V", terms=(a, b))
|
|
assert v.children == (a, b)
|
|
|
|
def test_vector_empty(self) -> None:
|
|
v = Vector("Empty")
|
|
assert v.terms == ()
|
|
assert v.children == ()
|
|
|
|
|
|
class TestDot:
|
|
def test_dot_construction(self) -> None:
|
|
lv = Vector("L", terms=(Scalar("A", value=1.0),))
|
|
rv = Vector("R", terms=(Scalar("B", value=2.0),))
|
|
d = Dot("D", left=lv, right=rv)
|
|
assert d.left is lv
|
|
assert d.right is rv
|
|
|
|
def test_dot_children(self) -> None:
|
|
lv = Vector("L", terms=(Scalar("A", value=1.0),))
|
|
rv = Vector("R", terms=(Scalar("B", value=2.0),))
|
|
d = Dot("D", left=lv, right=rv)
|
|
assert d.children == (lv, rv)
|
|
|
|
|
|
class TestMul:
|
|
def test_mul_construction(self) -> None:
|
|
a = Scalar("A", value=3.0)
|
|
b = Scalar("B", value=4.0)
|
|
m = Mul("M", left=a, right=b)
|
|
assert m.left is a
|
|
assert m.right is b
|
|
|
|
def test_mul_children(self) -> None:
|
|
a = Scalar("A", value=3.0)
|
|
b = Scalar("B", value=4.0)
|
|
m = Mul("M", left=a, right=b)
|
|
assert m.children == (a, b)
|
|
|
|
|
|
class TestAdd:
|
|
def test_add_construction(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
c = Scalar("C", value=3.0)
|
|
add = Add("Sum", terms=(a, b, c))
|
|
assert len(add.terms) == 3
|
|
|
|
def test_add_children(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
add = Add("Sum", terms=(a, b))
|
|
assert add.children == (a, b)
|
|
|
|
def test_add_empty(self) -> None:
|
|
add = Add("Empty")
|
|
assert add.terms == ()
|
|
assert add.children == ()
|
|
|
|
|
|
class TestFn:
|
|
def test_fn_construction(self) -> None:
|
|
a = Ref("X", table="t", column="x")
|
|
fn = Fn("MyFn", inputs=(a,))
|
|
assert fn.label == "MyFn"
|
|
assert len(fn.inputs) == 1
|
|
|
|
def test_fn_children_are_inputs(self) -> None:
|
|
a = Ref("X", table="t", column="x")
|
|
b = Ref("Y", table="t", column="y")
|
|
fn = Fn("MyFn", inputs=(a, b))
|
|
assert fn.children == (a, b)
|
|
|
|
def test_fn_no_inputs(self) -> None:
|
|
fn = Fn("MyFn")
|
|
assert fn.inputs == ()
|
|
assert fn.children == ()
|
|
|
|
def test_fn_with_callable(self) -> None:
|
|
def my_func():
|
|
pass
|
|
|
|
fn = Fn("MyFn", func=my_func)
|
|
assert fn.func is my_func
|
|
|
|
|
|
class TestCond:
|
|
def test_cond_construction(self) -> None:
|
|
t = Scalar("T", value=1.0)
|
|
f = Scalar("F", value=0.0)
|
|
c = Cond("C", condition="flag", if_true=t, if_false=f)
|
|
assert c.condition == "flag"
|
|
assert c.if_true is t
|
|
assert c.if_false is f
|
|
|
|
def test_cond_children(self) -> None:
|
|
t = Scalar("T", value=1.0)
|
|
f = Scalar("F", value=0.0)
|
|
c = Cond("C", condition="flag", if_true=t, if_false=f)
|
|
assert c.children == (t, f)
|
|
|
|
|
|
# ── Pre-built terms ───────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestPrebuiltLeaves:
|
|
def test_work_rvu_is_ref(self) -> None:
|
|
assert isinstance(WORK_RVU, Ref)
|
|
assert WORK_RVU.table == "pfs.rvu"
|
|
assert WORK_RVU.column == "work_rvu"
|
|
|
|
def test_nf_pe_rvu_is_ref(self) -> None:
|
|
assert isinstance(NF_PE_RVU, Ref)
|
|
assert NF_PE_RVU.table == "pfs.rvu"
|
|
|
|
def test_f_pe_rvu_is_ref(self) -> None:
|
|
assert isinstance(F_PE_RVU, Ref)
|
|
assert F_PE_RVU.table == "pfs.rvu"
|
|
|
|
def test_mp_rvu_is_ref(self) -> None:
|
|
assert isinstance(MP_RVU, Ref)
|
|
assert MP_RVU.table == "pfs.rvu"
|
|
|
|
def test_pe_rvu_is_cond(self) -> None:
|
|
assert isinstance(PE_RVU, Cond)
|
|
assert PE_RVU.condition == "facility"
|
|
assert PE_RVU.if_true is F_PE_RVU
|
|
assert PE_RVU.if_false is NF_PE_RVU
|
|
|
|
def test_work_gpci_is_ref(self) -> None:
|
|
assert isinstance(WORK_GPCI, Ref)
|
|
assert WORK_GPCI.table == "pfs.gpci"
|
|
|
|
def test_pe_gpci_is_ref(self) -> None:
|
|
assert isinstance(PE_GPCI, Ref)
|
|
assert PE_GPCI.table == "pfs.gpci"
|
|
|
|
def test_mp_gpci_is_ref(self) -> None:
|
|
assert isinstance(MP_GPCI, Ref)
|
|
assert MP_GPCI.table == "pfs.gpci"
|
|
|
|
def test_cf_is_scalar(self) -> None:
|
|
assert isinstance(CF, Scalar)
|
|
assert CF.unit == "$/RVU"
|
|
|
|
def test_cf_has_no_preset_value(self) -> None:
|
|
# CF must be supplied at evaluation time — no hardcoded default
|
|
assert CF.value is None
|
|
|
|
|
|
class TestPrebuiltEquations:
|
|
def test_pfs_payment_is_mul(self) -> None:
|
|
assert isinstance(PFS_PAYMENT, Mul)
|
|
|
|
def test_pfs_payment_label(self) -> None:
|
|
assert PFS_PAYMENT.label == "Payment"
|
|
|
|
def test_pfs_payment_right_is_cf(self) -> None:
|
|
assert PFS_PAYMENT.right is CF
|
|
|
|
def test_pfs_payment_left_is_dot(self) -> None:
|
|
assert isinstance(PFS_PAYMENT.left, Dot)
|
|
|
|
def test_direct_pe_is_add(self) -> None:
|
|
assert isinstance(DIRECT_PE, Add)
|
|
assert DIRECT_PE.label == "Direct_PE"
|
|
|
|
def test_direct_pe_has_three_components(self) -> None:
|
|
assert len(DIRECT_PE.terms) == 3
|
|
|
|
def test_direct_pe_components(self) -> None:
|
|
labels = {t.label for t in DIRECT_PE.terms}
|
|
assert "Labor_Cost" in labels
|
|
assert "Supply_Cost" in labels
|
|
assert "Equipment_Cost" in labels
|
|
|
|
def test_mppr_is_cond(self) -> None:
|
|
assert isinstance(MPPR, Cond)
|
|
assert MPPR.label == "Adjusted_PE"
|
|
|
|
def test_intensity_is_mul(self) -> None:
|
|
assert isinstance(INTENSITY, Mul)
|
|
assert INTENSITY.label == "Intensity"
|
|
|
|
def test_labor_cost_is_fn(self) -> None:
|
|
assert isinstance(LABOR_COST, Fn)
|
|
|
|
def test_supply_cost_is_fn(self) -> None:
|
|
assert isinstance(SUPPLY_COST, Fn)
|
|
|
|
def test_equipment_cost_is_fn(self) -> None:
|
|
assert isinstance(EQUIPMENT_COST, Fn)
|
|
|
|
|
|
# ── walk() ────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestWalk:
|
|
def test_walk_single_scalar(self) -> None:
|
|
s = Scalar("X", value=1.0)
|
|
nodes = list(walk(s))
|
|
assert nodes == [s]
|
|
|
|
def test_walk_single_ref(self) -> None:
|
|
r = Ref("X", table="t", column="c")
|
|
nodes = list(walk(r))
|
|
assert nodes == [r]
|
|
|
|
def test_walk_mul(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
m = Mul("M", left=a, right=b)
|
|
nodes = list(walk(m))
|
|
assert m in nodes
|
|
assert a in nodes
|
|
assert b in nodes
|
|
assert len(nodes) == 3
|
|
|
|
def test_walk_includes_root(self) -> None:
|
|
s = Scalar("Root", value=0.0)
|
|
nodes = list(walk(s))
|
|
assert nodes[0] is s
|
|
|
|
def test_walk_depth_first(self) -> None:
|
|
# Mul(left=Scalar(A), right=Scalar(B)) — root first
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
m = Mul("M", left=a, right=b)
|
|
nodes = list(walk(m))
|
|
assert nodes[0] is m
|
|
|
|
def test_walk_nested(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
c = Scalar("C", value=3.0)
|
|
inner = Mul("Inner", left=a, right=b)
|
|
outer = Add("Outer", terms=(inner, c))
|
|
nodes = list(walk(outer))
|
|
labels = {n.label for n in nodes}
|
|
assert labels == {"Outer", "Inner", "A", "B", "C"}
|
|
|
|
def test_walk_add_visits_all_terms(self) -> None:
|
|
terms = tuple(Scalar(f"T{i}", value=float(i)) for i in range(5))
|
|
add = Add("Sum", terms=terms)
|
|
nodes = list(walk(add))
|
|
assert len(nodes) == 6 # root + 5 terms
|
|
|
|
def test_walk_pfs_payment_visits_multiple_nodes(self) -> None:
|
|
nodes = list(walk(PFS_PAYMENT))
|
|
assert len(nodes) > 5
|
|
|
|
def test_walk_vector_visits_sub_terms(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
v = Vector("V", terms=(a, b))
|
|
nodes = list(walk(v))
|
|
assert v in nodes
|
|
assert a in nodes
|
|
assert b in nodes
|
|
|
|
def test_walk_fn_visits_inputs(self) -> None:
|
|
a = Ref("A", table="t", column="a")
|
|
b = Ref("B", table="t", column="b")
|
|
fn = Fn("F", inputs=(a, b))
|
|
nodes = list(walk(fn))
|
|
assert fn in nodes
|
|
assert a in nodes
|
|
assert b in nodes
|
|
|
|
def test_walk_cond_visits_both_branches(self) -> None:
|
|
t = Scalar("T", value=1.0)
|
|
f = Scalar("F", value=0.0)
|
|
c = Cond("C", condition="flag", if_true=t, if_false=f)
|
|
nodes = list(walk(c))
|
|
assert t in nodes
|
|
assert f in nodes
|
|
|
|
|
|
# ── refs() ────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestRefs:
|
|
def test_refs_scalar_returns_empty(self) -> None:
|
|
s = Scalar("X", value=1.0)
|
|
assert refs(s) == []
|
|
|
|
def test_refs_single_ref(self) -> None:
|
|
r = Ref("X", table="t", column="c")
|
|
result = refs(r)
|
|
assert result == [r]
|
|
|
|
def test_refs_mul_with_refs(self) -> None:
|
|
a = Ref("A", table="t1", column="a")
|
|
b = Ref("B", table="t2", column="b")
|
|
m = Mul("M", left=a, right=b)
|
|
result = refs(m)
|
|
assert a in result
|
|
assert b in result
|
|
assert len(result) == 2
|
|
|
|
def test_refs_filters_out_scalars(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Ref("B", table="t", column="c")
|
|
m = Mul("M", left=a, right=b)
|
|
result = refs(m)
|
|
assert a not in result
|
|
assert b in result
|
|
|
|
def test_refs_pfs_payment_has_many(self) -> None:
|
|
result = refs(PFS_PAYMENT)
|
|
assert len(result) > 3
|
|
|
|
def test_refs_direct_pe_non_empty(self) -> None:
|
|
# DIRECT_PE has Fn children with Ref inputs
|
|
result = refs(DIRECT_PE)
|
|
assert len(result) > 0
|
|
|
|
|
|
# ── sources() ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSources:
|
|
def test_sources_scalar_returns_empty(self) -> None:
|
|
s = Scalar("X", value=1.0)
|
|
assert sources(s) == set()
|
|
|
|
def test_sources_ref_without_table(self) -> None:
|
|
r = Ref("X")
|
|
result = sources(r)
|
|
assert result == set()
|
|
|
|
def test_sources_single_ref(self) -> None:
|
|
r = Ref("X", table="pfs.rvu", column="c")
|
|
result = sources(r)
|
|
assert result == {"pfs.rvu"}
|
|
|
|
def test_sources_multiple_tables(self) -> None:
|
|
a = Ref("A", table="pfs.rvu", column="a")
|
|
b = Ref("B", table="pfs.gpci", column="b")
|
|
m = Mul("M", left=a, right=b)
|
|
result = sources(m)
|
|
assert result == {"pfs.rvu", "pfs.gpci"}
|
|
|
|
def test_sources_deduplicates(self) -> None:
|
|
a = Ref("A", table="pfs.rvu", column="a")
|
|
b = Ref("B", table="pfs.rvu", column="b")
|
|
m = Mul("M", left=a, right=b)
|
|
result = sources(m)
|
|
assert result == {"pfs.rvu"}
|
|
|
|
def test_sources_pfs_payment(self) -> None:
|
|
result = sources(PFS_PAYMENT)
|
|
assert "pfs.rvu" in result
|
|
assert "pfs.gpci" in result
|
|
|
|
def test_sources_returns_set(self) -> None:
|
|
result = sources(WORK_RVU)
|
|
assert isinstance(result, set)
|
|
|
|
|
|
# ── scalars() ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestScalars:
|
|
def test_scalars_single_scalar(self) -> None:
|
|
s = Scalar("X", value=1.0)
|
|
result = scalars(s)
|
|
assert result == [s]
|
|
|
|
def test_scalars_ref_returns_empty(self) -> None:
|
|
r = Ref("X", table="t", column="c")
|
|
assert scalars(r) == []
|
|
|
|
def test_scalars_mul_with_scalar(self) -> None:
|
|
a = Scalar("A", value=2.0)
|
|
b = Ref("B", table="t", column="c")
|
|
m = Mul("M", left=a, right=b)
|
|
result = scalars(m)
|
|
assert a in result
|
|
assert len(result) == 1
|
|
|
|
def test_scalars_pfs_payment_includes_cf(self) -> None:
|
|
result = scalars(PFS_PAYMENT)
|
|
labels = {s.label for s in result}
|
|
assert "CF" in labels
|
|
|
|
def test_scalars_returns_list(self) -> None:
|
|
result = scalars(WORK_RVU)
|
|
assert isinstance(result, list)
|
|
|
|
|
|
# ── citations() ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestCitations:
|
|
def test_citations_no_cite(self) -> None:
|
|
s = Scalar("X", value=1.0)
|
|
result = citations(s)
|
|
assert result == []
|
|
|
|
def test_citations_single_tag(self) -> None:
|
|
tag = Tag.module("pfs")
|
|
s = Scalar("X", value=1.0, cite=(tag,))
|
|
result = citations(s)
|
|
assert result == [tag]
|
|
|
|
def test_citations_multiple_tags(self) -> None:
|
|
t1 = Tag.module("pfs")
|
|
t2 = Tag.source("federal-register")
|
|
s = Scalar("X", value=1.0, cite=(t1, t2))
|
|
result = citations(s)
|
|
assert t1 in result
|
|
assert t2 in result
|
|
|
|
def test_citations_pfs_payment_non_empty(self) -> None:
|
|
result = citations(PFS_PAYMENT)
|
|
assert len(result) > 0
|
|
|
|
def test_citations_work_rvu_non_empty(self) -> None:
|
|
result = citations(WORK_RVU)
|
|
assert len(result) > 0
|
|
|
|
def test_citations_aggregates_across_tree(self) -> None:
|
|
t1 = Tag.module("pfs")
|
|
t2 = Tag.source("cms-website")
|
|
a = Scalar("A", value=1.0, cite=(t1,))
|
|
b = Scalar("B", value=2.0, cite=(t2,))
|
|
m = Mul("M", left=a, right=b)
|
|
result = citations(m)
|
|
assert t1 in result
|
|
assert t2 in result
|
|
|
|
def test_citations_returns_list(self) -> None:
|
|
result = citations(WORK_RVU)
|
|
assert isinstance(result, list)
|
|
|
|
|
|
# ── depth() ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestDepth:
|
|
def test_depth_scalar_is_zero(self) -> None:
|
|
s = Scalar("X", value=1.0)
|
|
assert depth(s) == 0
|
|
|
|
def test_depth_ref_is_zero(self) -> None:
|
|
r = Ref("X", table="t", column="c")
|
|
assert depth(r) == 0
|
|
|
|
def test_depth_single_level(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
m = Mul("M", left=a, right=b)
|
|
assert depth(m) == 1
|
|
|
|
def test_depth_two_levels(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
c = Scalar("C", value=3.0)
|
|
inner = Mul("Inner", left=a, right=b)
|
|
outer = Mul("Outer", left=inner, right=c)
|
|
assert depth(outer) == 2
|
|
|
|
def test_depth_pfs_payment_greater_than_one(self) -> None:
|
|
assert depth(PFS_PAYMENT) > 1
|
|
|
|
def test_depth_add_with_scalars(self) -> None:
|
|
terms = tuple(Scalar(f"T{i}", value=float(i)) for i in range(3))
|
|
add = Add("Sum", terms=terms)
|
|
assert depth(add) == 1
|
|
|
|
def test_depth_fn_with_inputs(self) -> None:
|
|
a = Ref("A", table="t", column="a")
|
|
fn = Fn("F", inputs=(a,))
|
|
assert depth(fn) == 1
|
|
|
|
|
|
# ── evaluate() ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestEvaluate:
|
|
def test_evaluate_scalar_with_preset_value(self) -> None:
|
|
s = Scalar("X", value=5.0)
|
|
assert evaluate(s, {}) == pytest.approx(5.0)
|
|
|
|
def test_evaluate_scalar_from_values_dict(self) -> None:
|
|
s = Scalar("X")
|
|
assert evaluate(s, {"X": 42.0}) == pytest.approx(42.0)
|
|
|
|
def test_evaluate_scalar_no_value_raises(self) -> None:
|
|
s = Scalar("X")
|
|
with pytest.raises(ValueError, match="no value"):
|
|
evaluate(s, {})
|
|
|
|
def test_evaluate_ref_from_values(self) -> None:
|
|
r = Ref("Work_RVU", table="pfs.rvu", column="work_rvu")
|
|
assert evaluate(r, {"Work_RVU": 1.5}) == pytest.approx(1.5)
|
|
|
|
def test_evaluate_ref_missing_raises(self) -> None:
|
|
r = Ref("Work_RVU", table="pfs.rvu", column="work_rvu")
|
|
with pytest.raises(ValueError, match="not in values"):
|
|
evaluate(r, {})
|
|
|
|
def test_evaluate_mul_basic(self) -> None:
|
|
a = Scalar("A", value=3.0)
|
|
b = Scalar("B", value=4.0)
|
|
m = Mul("M", left=a, right=b)
|
|
assert evaluate(m, {}) == pytest.approx(12.0)
|
|
|
|
def test_evaluate_add_basic(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
c = Scalar("C", value=3.0)
|
|
add = Add("Sum", terms=(a, b, c))
|
|
assert evaluate(add, {}) == pytest.approx(6.0)
|
|
|
|
def test_evaluate_vector_sums_terms(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
v = Vector("V", terms=(a, b))
|
|
assert evaluate(v, {}) == pytest.approx(3.0)
|
|
|
|
def test_evaluate_dot_basic(self) -> None:
|
|
# [1, 2] · [3, 4] = 1*3 + 2*4 = 11
|
|
lv = Vector("L", terms=(Scalar("a", value=1.0), Scalar("b", value=2.0)))
|
|
rv = Vector("R", terms=(Scalar("c", value=3.0), Scalar("d", value=4.0)))
|
|
d = Dot("D", left=lv, right=rv)
|
|
assert evaluate(d, {}) == pytest.approx(11.0)
|
|
|
|
def test_evaluate_dot_with_refs(self) -> None:
|
|
lv = Vector("L", terms=(Ref("x"), Ref("y")))
|
|
rv = Vector("R", terms=(Ref("a"), Ref("b")))
|
|
d = Dot("D", left=lv, right=rv)
|
|
result = evaluate(d, {"x": 1.5, "y": 2.0, "a": 2.0, "b": 3.0})
|
|
assert result == pytest.approx(1.5 * 2.0 + 2.0 * 3.0)
|
|
|
|
def test_evaluate_fn_not_callable_raises(self) -> None:
|
|
fn = Fn("MyFn", inputs=(Ref("X", table="t", column="x"),))
|
|
with pytest.raises(ValueError, match="pre-computed"):
|
|
evaluate(fn, {})
|
|
|
|
def test_evaluate_fn_from_values_dict(self) -> None:
|
|
fn = Fn("MyFn")
|
|
assert evaluate(fn, {"MyFn": 99.0}) == pytest.approx(99.0)
|
|
|
|
def test_evaluate_cond_true_branch(self) -> None:
|
|
t_term = Scalar("T", value=10.0)
|
|
f_term = Scalar("F", value=20.0)
|
|
c = Cond("C", condition="use_t", if_true=t_term, if_false=f_term)
|
|
assert evaluate(c, {"use_t": True}) == pytest.approx(10.0)
|
|
|
|
def test_evaluate_cond_false_branch(self) -> None:
|
|
t_term = Scalar("T", value=10.0)
|
|
f_term = Scalar("F", value=20.0)
|
|
c = Cond("C", condition="use_t", if_true=t_term, if_false=f_term)
|
|
assert evaluate(c, {"use_t": False}) == pytest.approx(20.0)
|
|
|
|
def test_evaluate_cond_missing_condition_is_falsy(self) -> None:
|
|
t_term = Scalar("T", value=10.0)
|
|
f_term = Scalar("F", value=20.0)
|
|
c = Cond("C", condition="flag", if_true=t_term, if_false=f_term)
|
|
# Missing key → .get returns None → falsy → false branch
|
|
result = evaluate(c, {})
|
|
assert result == pytest.approx(20.0)
|
|
|
|
def test_evaluate_label_override_takes_priority(self) -> None:
|
|
# Even a Mul will short-circuit if its label is in values
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
m = Mul("M", left=a, right=b)
|
|
assert evaluate(m, {"M": 999.0}) == pytest.approx(999.0)
|
|
|
|
def test_evaluate_pfs_payment_formula(self) -> None:
|
|
values = {
|
|
"Work_RVU": 1.50,
|
|
"PE_RVU": 1.21,
|
|
"MP_RVU": 0.09,
|
|
"Work_GPCI": 1.024,
|
|
"PE_GPCI": 0.988,
|
|
"MP_GPCI": 0.574,
|
|
"CF": 32.3465,
|
|
# PE_RVU is a Cond, so we need to supply the result directly
|
|
# or supply "facility" flag:
|
|
"facility": False,
|
|
"NF_PE_RVU": 1.21,
|
|
"F_PE_RVU": 1.00,
|
|
}
|
|
result = evaluate(PFS_PAYMENT, values)
|
|
# Manual: (1.5*1.024 + 1.21*0.988 + 0.09*0.574) * 32.3465
|
|
expected = (1.50 * 1.024 + 1.21 * 0.988 + 0.09 * 0.574) * 32.3465
|
|
assert result == pytest.approx(expected, rel=1e-4)
|
|
|
|
def test_evaluate_nested_mul_add(self) -> None:
|
|
# (2 + 3) * 4 = 20
|
|
inner = Add("Sum", terms=(Scalar("A", value=2.0), Scalar("B", value=3.0)))
|
|
outer = Mul("M", left=inner, right=Scalar("C", value=4.0))
|
|
assert evaluate(outer, {}) == pytest.approx(20.0)
|
|
|
|
def test_evaluate_dot_mismatched_lengths_raises(self) -> None:
|
|
lv = Vector("L", terms=(Scalar("a", value=1.0), Scalar("b", value=2.0)))
|
|
rv = Vector("R", terms=(Scalar("c", value=3.0),))
|
|
d = Dot("D", left=lv, right=rv)
|
|
with pytest.raises((ValueError, Exception)):
|
|
evaluate(d, {})
|
|
|
|
def test_evaluate_unknown_term_type_raises(self) -> None:
|
|
# Subclass that isn't handled by the match
|
|
class OddTerm(Term):
|
|
pass
|
|
|
|
odd = OddTerm(label="Weird")
|
|
with pytest.raises(TypeError, match="Unknown term type"):
|
|
evaluate(odd, {})
|
|
|
|
|
|
# ── render() ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestRenderText:
|
|
def test_render_defaults_to_text(self) -> None:
|
|
s = Scalar("CF", value=32.3465)
|
|
result = render(s)
|
|
assert isinstance(result, str)
|
|
assert "CF" in result
|
|
|
|
def test_render_scalar_with_value(self) -> None:
|
|
s = Scalar("CF", value=32.3465, unit="$/RVU")
|
|
result = render(s, "text")
|
|
assert "CF" in result
|
|
assert "32.3465" in result
|
|
assert "$/RVU" in result
|
|
|
|
def test_render_scalar_no_value(self) -> None:
|
|
s = Scalar("X")
|
|
result = render(s, "text")
|
|
assert "X" in result
|
|
assert "None" not in result
|
|
|
|
def test_render_ref(self) -> None:
|
|
r = Ref("Work_RVU", table="pfs.rvu", column="work_rvu")
|
|
result = render(r, "text")
|
|
assert "Work_RVU" in result
|
|
assert "pfs.rvu" in result
|
|
|
|
def test_render_ref_no_table(self) -> None:
|
|
r = Ref("X")
|
|
result = render(r, "text")
|
|
assert "X" in result
|
|
|
|
def test_render_vector(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
v = Vector("V", terms=(a, b))
|
|
result = render(v, "text")
|
|
assert "V" in result
|
|
assert "A" in result
|
|
assert "B" in result
|
|
|
|
def test_render_dot(self) -> None:
|
|
lv = Vector("RVUs", terms=(Scalar("A", value=1.0),))
|
|
rv = Vector("GPCIs", terms=(Scalar("B", value=2.0),))
|
|
d = Dot("GeoRVU", left=lv, right=rv)
|
|
result = render(d, "text")
|
|
assert "GeoRVU" in result
|
|
assert "RVUs" in result
|
|
|
|
def test_render_mul(self) -> None:
|
|
a = Scalar("A", value=2.0)
|
|
b = Scalar("B", value=3.0)
|
|
m = Mul("M", left=a, right=b)
|
|
result = render(m, "text")
|
|
assert "M" in result
|
|
assert "A" in result
|
|
assert "B" in result
|
|
|
|
def test_render_add(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
add = Add("Sum", terms=(a, b))
|
|
result = render(add, "text")
|
|
assert "Sum" in result
|
|
assert "A" in result
|
|
assert "B" in result
|
|
|
|
def test_render_fn(self) -> None:
|
|
a = Ref("X", table="t", column="x")
|
|
fn = Fn("MyFn", inputs=(a,))
|
|
result = render(fn, "text")
|
|
assert "MyFn" in result
|
|
assert "X" in result
|
|
|
|
def test_render_cond(self) -> None:
|
|
t = Scalar("T", value=1.0)
|
|
f = Scalar("F", value=0.0)
|
|
c = Cond("C", condition="flag", if_true=t, if_false=f)
|
|
result = render(c, "text")
|
|
assert "C" in result
|
|
assert "flag" in result
|
|
assert "T" in result
|
|
assert "F" in result
|
|
|
|
def test_render_pfs_payment_text(self) -> None:
|
|
result = render(PFS_PAYMENT, "text")
|
|
assert "Payment" in result
|
|
assert isinstance(result, str)
|
|
assert len(result) > 0
|
|
|
|
def test_render_direct_pe_text(self) -> None:
|
|
result = render(DIRECT_PE, "text")
|
|
assert "Direct_PE" in result
|
|
|
|
def test_render_unknown_format_raises(self) -> None:
|
|
s = Scalar("X", value=1.0)
|
|
with pytest.raises(ValueError, match="Unknown format"):
|
|
render(s, "unknown_format")
|
|
|
|
|
|
class TestRenderLatex:
|
|
def test_render_latex_returns_string(self) -> None:
|
|
result = render(WORK_RVU, "latex")
|
|
assert isinstance(result, str)
|
|
assert len(result) > 0
|
|
|
|
def test_render_latex_scalar_with_value(self) -> None:
|
|
s = Scalar("CF", value=32.3465)
|
|
result = render(s, "latex")
|
|
assert "32.3465" in result
|
|
|
|
def test_render_latex_scalar_no_value(self) -> None:
|
|
s = Scalar("X")
|
|
result = render(s, "latex")
|
|
assert "X" in result
|
|
|
|
def test_render_latex_ref(self) -> None:
|
|
r = Ref("Work_RVU", table="pfs.rvu", column="work_rvu")
|
|
result = render(r, "latex")
|
|
assert "Work" in result
|
|
|
|
def test_render_latex_vector(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
v = Vector("V", terms=(a, b))
|
|
result = render(v, "latex")
|
|
assert "bmatrix" in result
|
|
|
|
def test_render_latex_dot(self) -> None:
|
|
lv = Vector("L", terms=(Scalar("A", value=1.0),))
|
|
rv = Vector("R", terms=(Scalar("B", value=2.0),))
|
|
d = Dot("D", left=lv, right=rv)
|
|
result = render(d, "latex")
|
|
assert "cdot" in result
|
|
|
|
def test_render_latex_mul(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
m = Mul("M", left=a, right=b)
|
|
result = render(m, "latex")
|
|
assert "times" in result
|
|
|
|
def test_render_latex_add(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
add = Add("Sum", terms=(a, b))
|
|
result = render(add, "latex")
|
|
assert "+" in result
|
|
|
|
def test_render_latex_fn(self) -> None:
|
|
fn = Fn("MyFn")
|
|
result = render(fn, "latex")
|
|
assert "MyFn" in result
|
|
assert "cdots" in result
|
|
|
|
def test_render_latex_cond(self) -> None:
|
|
t = Scalar("T", value=1.0)
|
|
f = Scalar("F", value=0.0)
|
|
c = Cond("C", condition="flag", if_true=t, if_false=f)
|
|
result = render(c, "latex")
|
|
assert "cases" in result
|
|
assert "flag" in result
|
|
|
|
def test_render_latex_pfs_payment(self) -> None:
|
|
result = render(PFS_PAYMENT, "latex")
|
|
assert isinstance(result, str)
|
|
assert "times" in result
|
|
|
|
def test_render_latex_underscore_escaped(self) -> None:
|
|
r = Ref("Work_RVU", table="pfs.rvu", column="work_rvu")
|
|
result = render(r, "latex")
|
|
# underscores in labels should be escaped for LaTeX
|
|
assert r"\\_" in result or "\\_" in result or "Work" in result
|
|
|
|
|
|
class TestRenderHtml:
|
|
def test_render_html_returns_string(self) -> None:
|
|
result = render(PFS_PAYMENT, "html")
|
|
assert isinstance(result, str)
|
|
|
|
def test_render_html_is_html_document(self) -> None:
|
|
result = render(PFS_PAYMENT, "html")
|
|
assert "<!DOCTYPE html>" in result
|
|
|
|
def test_render_html_contains_label(self) -> None:
|
|
result = render(PFS_PAYMENT, "html")
|
|
assert "Payment" in result
|
|
|
|
def test_render_html_scalar(self) -> None:
|
|
s = Scalar("CF", value=32.3465, unit="$/RVU")
|
|
result = render(s, "html")
|
|
assert "<!DOCTYPE html>" in result
|
|
assert "CF" in result
|
|
|
|
def test_render_html_ref(self) -> None:
|
|
r = Ref("Work_RVU", table="pfs.rvu", column="work_rvu")
|
|
result = render(r, "html")
|
|
assert "<!DOCTYPE html>" in result
|
|
|
|
def test_render_html_direct_pe(self) -> None:
|
|
result = render(DIRECT_PE, "html")
|
|
assert "Direct_PE" in result or "Direct" in result
|
|
|
|
def test_render_html_katex_included(self) -> None:
|
|
result = render(PFS_PAYMENT, "html")
|
|
assert "katex" in result.lower()
|
|
|
|
def test_render_html_node_with_table(self) -> None:
|
|
import base64
|
|
import json
|
|
import re
|
|
|
|
r = Ref("Work_RVU", table="pfs.rvu", column="work_rvu")
|
|
result = render(r, "html")
|
|
# Node data is base64-encoded in the HTML
|
|
m = re.search(r'atob\("([A-Za-z0-9+/=]+)"\)', result)
|
|
assert m is not None
|
|
nodes = json.loads(base64.b64decode(m.group(1)))
|
|
tables = [n.get("table") for n in nodes]
|
|
assert "pfs.rvu" in tables
|
|
|
|
def test_render_html_cite_labels_included(self) -> None:
|
|
tag = Tag.rule("cms-1807-f")
|
|
s = Scalar("X", value=1.0, cite=(tag,))
|
|
result = render(s, "html")
|
|
assert isinstance(result, str)
|
|
|
|
def test_render_html_cond(self) -> None:
|
|
t = Scalar("T", value=1.0)
|
|
f = Scalar("F", value=0.0)
|
|
c = Cond("C", condition="flag", if_true=t, if_false=f)
|
|
result = render(c, "html")
|
|
assert "<!DOCTYPE html>" in result
|
|
|
|
|
|
class TestRenderMermaid:
|
|
def test_render_mermaid_returns_string(self) -> None:
|
|
result = render(PFS_PAYMENT, "mermaid")
|
|
assert isinstance(result, str)
|
|
|
|
def test_render_mermaid_starts_with_graph(self) -> None:
|
|
result = render(PFS_PAYMENT, "mermaid")
|
|
assert result.startswith("graph TD")
|
|
|
|
def test_render_mermaid_scalar(self) -> None:
|
|
s = Scalar("CF", value=32.3465)
|
|
result = render(s, "mermaid")
|
|
assert "graph TD" in result
|
|
assert "CF" in result
|
|
|
|
def test_render_mermaid_ref(self) -> None:
|
|
r = Ref("Work_RVU", table="pfs.rvu", column="work_rvu")
|
|
result = render(r, "mermaid")
|
|
assert "Work_RVU" in result
|
|
|
|
def test_render_mermaid_vector(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
v = Vector("V", terms=(a, b))
|
|
result = render(v, "mermaid")
|
|
assert "V" in result
|
|
|
|
def test_render_mermaid_dot(self) -> None:
|
|
lv = Vector("L", terms=(Scalar("A", value=1.0),))
|
|
rv = Vector("R", terms=(Scalar("B", value=2.0),))
|
|
d = Dot("D", left=lv, right=rv)
|
|
result = render(d, "mermaid")
|
|
assert "D" in result
|
|
|
|
def test_render_mermaid_mul(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
m = Mul("M", left=a, right=b)
|
|
result = render(m, "mermaid")
|
|
assert "M" in result
|
|
|
|
def test_render_mermaid_add(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
add = Add("Sum", terms=(a, b))
|
|
result = render(add, "mermaid")
|
|
assert "Sum" in result
|
|
|
|
def test_render_mermaid_fn(self) -> None:
|
|
fn = Fn("MyFn")
|
|
result = render(fn, "mermaid")
|
|
assert "MyFn" in result
|
|
|
|
def test_render_mermaid_cond(self) -> None:
|
|
t = Scalar("T", value=1.0)
|
|
f = Scalar("F", value=0.0)
|
|
c = Cond("C", condition="flag", if_true=t, if_false=f)
|
|
result = render(c, "mermaid")
|
|
assert "C" in result
|
|
|
|
def test_render_mermaid_direct_pe(self) -> None:
|
|
result = render(DIRECT_PE, "mermaid")
|
|
assert "Direct_PE" in result
|
|
|
|
def test_render_mermaid_edges_present(self) -> None:
|
|
a = Scalar("A", value=1.0)
|
|
b = Scalar("B", value=2.0)
|
|
m = Mul("M", left=a, right=b)
|
|
result = render(m, "mermaid")
|
|
assert "-->" in result
|
|
|
|
|
|
# ── render fallback (case _:) for custom Term subclasses ─────────────────────
|
|
|
|
|
|
class TestRenderFallbackBranches:
|
|
"""Cover the `case _:` fallback in _render_text, _render_latex, _mermaid_node."""
|
|
|
|
def _custom_term(self) -> Term:
|
|
"""A Term subclass that doesn't match Scalar/Ref/Vector/Dot/Mul/Add/Fn/Cond."""
|
|
|
|
class CustomTerm(Term):
|
|
pass
|
|
|
|
return CustomTerm(label="Custom")
|
|
|
|
def test_render_text_fallback(self) -> None:
|
|
result = render(self._custom_term(), "text")
|
|
assert "Custom" in result
|
|
|
|
def test_render_latex_fallback(self) -> None:
|
|
result = render(self._custom_term(), "latex")
|
|
assert "Custom" in result
|
|
|
|
def test_render_mermaid_fallback(self) -> None:
|
|
result = render(self._custom_term(), "mermaid")
|
|
assert "Custom" in result
|
|
|
|
|
|
# ── aco/table/base coverage ───────────────────────────────────────────────────
|
|
|
|
|
|
# ── Gap coverage — missed lines ────────────────────────────────
|
|
|
|
|
|
class TestTagImportFallback:
|
|
"""Lines 42-43: Tag = None when bib.tag is not importable.
|
|
|
|
The except-ImportError branch at lines 42-43 is a guard for
|
|
installations without stack[bib]. The module-level constants
|
|
(line 168+) unconditionally call Tag.module(), so when Tag is
|
|
None the reload crashes — proving the fallback path *is*
|
|
exercised even though the module can't fully load without bib.
|
|
"""
|
|
|
|
def test_tag_none_when_bib_missing(self) -> None:
|
|
import importlib
|
|
import sys
|
|
|
|
saved_modules = {}
|
|
for key in list(sys.modules):
|
|
if key.startswith("bib") or key.startswith("pfs.eq"):
|
|
saved_modules[key] = sys.modules.pop(key)
|
|
|
|
import builtins
|
|
|
|
_real_import = builtins.__import__
|
|
|
|
def _fake_import(name, *args, **kwargs):
|
|
if name == "bib.tag" or name == "bib":
|
|
raise ImportError("fake: no bib")
|
|
return _real_import(name, *args, **kwargs)
|
|
|
|
builtins.__import__ = _fake_import
|
|
try:
|
|
# Without bib.tag, Tag is set to None (lines 42-43) but the
|
|
# module-level constants that call Tag.module() then crash.
|
|
with pytest.raises(AttributeError, match="NoneType"):
|
|
importlib.import_module("pfs.eq")
|
|
finally:
|
|
builtins.__import__ = _real_import
|
|
for key, val in saved_modules.items():
|
|
sys.modules[key] = val
|
|
importlib.reload(importlib.import_module("pfs.eq"))
|
|
|
|
|
|
class TestSQLTableBase:
|
|
def test_qualified_name(self) -> None:
|
|
from aco.table.core import CoreStgClaimsMemberMonths
|
|
|
|
assert CoreStgClaimsMemberMonths.qualified_name() == (
|
|
"core._stg_claims_member_months"
|
|
)
|
|
|
|
def test_column_names_returns_list(self) -> None:
|
|
from aco.table.core import CoreStgClinicalEncounter
|
|
|
|
cols = CoreStgClinicalEncounter.column_names()
|
|
assert isinstance(cols, list)
|
|
assert len(cols) > 0
|
|
|
|
def test_column_names_contains_expected_fields(self) -> None:
|
|
from aco.table.core import CoreStgClinicalPatient
|
|
|
|
cols = CoreStgClinicalPatient.column_names()
|
|
assert "person_id" in cols
|
|
|
|
def test_qualified_name_cclf_pipe_table(self) -> None:
|
|
from aco.table.cclf_pipe import CclfStgBeneficiaryXref
|
|
|
|
assert CclfStgBeneficiaryXref.qualified_name() == "cclf._stg_beneficiary_xref"
|
|
|
|
def test_column_names_cclf_xref(self) -> None:
|
|
from aco.table.cclf_pipe import CclfStgBeneficiaryXref
|
|
|
|
cols = CclfStgBeneficiaryXref.column_names()
|
|
assert "crnt_num" in cols
|
|
assert "prvs_num" in cols
|