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
556 lines
18 KiB
Python
556 lines
18 KiB
Python
"""Tests targeting exact missing coverage lines across aco/ modules."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import polars as pl
|
|
import pytest
|
|
|
|
# ── aco/load/cclf.py gaps (lines 201,212,228,235-256) ───────────
|
|
|
|
|
|
class TestLoadCclfRunPipeline:
|
|
"""Cover lines 228, 235-256 (_run_cclf_pipeline)."""
|
|
|
|
def test_run_pipeline_on_success(self, tmp_path):
|
|
"""When run_pipeline=True and stats non-empty, _run_cclf_pipeline is called."""
|
|
from aco.load.cclf import load_cclf_directory
|
|
|
|
# Create a valid CCLF9 file
|
|
line = "N1AN0Y00AA042AN0Y00AA052024-01-152025-06-30RRB123456789"
|
|
cclf_file = tmp_path / "P.A1234.ACO.ZC9Y25.D250716.T1234567"
|
|
cclf_file.write_text(line + "\n")
|
|
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
|
|
with patch("aco.load.cclf._run_cclf_pipeline", return_value={}) as mock_run:
|
|
stats = load_cclf_directory(tmp_path, database=db_path, run_pipeline=True)
|
|
|
|
assert stats["cclf9"] == 1
|
|
mock_run.assert_called_once()
|
|
|
|
def test_run_cclf_pipeline_saves_outputs(self, tmp_path):
|
|
"""Cover lines 235-255."""
|
|
import duckdb
|
|
|
|
from aco.load.cclf import _run_cclf_pipeline
|
|
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
con = duckdb.connect(db_path)
|
|
con.close()
|
|
|
|
mock_ctx = MagicMock()
|
|
mock_cache = {
|
|
"cclf.medical_claim": pl.DataFrame({"a": [1]}),
|
|
"cclf.pharmacy_claim": pl.DataFrame({"b": [2]}),
|
|
"cclf.eligibility": pl.DataFrame({"c": [3]}),
|
|
}
|
|
mock_pipeline = MagicMock()
|
|
mock_pipeline.run.return_value = mock_cache
|
|
|
|
with (
|
|
patch("aco.lake.context.DuckDBContext", return_value=mock_ctx),
|
|
patch("aco.pipe.cclf.pipeline", mock_pipeline),
|
|
):
|
|
result = _run_cclf_pipeline(db_path)
|
|
|
|
assert "input_layer.medical_claim" in result
|
|
assert "input_layer.pharmacy_claim" in result
|
|
assert "input_layer.eligibility" in result
|
|
|
|
|
|
class TestLoadCclfNoLines:
|
|
"""Cover line 212 (all_lines empty)."""
|
|
|
|
def test_skips_empty_file(self, tmp_path):
|
|
from aco.load.cclf import load_cclf_directory
|
|
|
|
cclf_file = tmp_path / "P.A1234.ACO.ZC9Y25.D250716.T1234567"
|
|
cclf_file.write_text("") # Empty
|
|
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
# The file exists but is empty so all_lines will be empty
|
|
# But discover will find it; the table won't be in stats
|
|
stats = load_cclf_directory(tmp_path, database=db_path, run_pipeline=False)
|
|
assert "cclf9" not in stats # line 212
|
|
|
|
|
|
class TestLoadCclfNotInLayouts:
|
|
"""Cover line 201 (cclf_table not in LAYOUTS)."""
|
|
|
|
def test_unknown_table_skipped(self, tmp_path):
|
|
# Create a file that classifies as an unknown cclf table
|
|
from unittest.mock import patch
|
|
|
|
from aco.load.cclf import load_cclf_directory
|
|
from aco.table.cclf_filenames import CclfFilename
|
|
|
|
fake_file = tmp_path / "P.A1234.ACO.ZC99Y25.D250716.T1234567"
|
|
fake_file.write_text("data\n")
|
|
|
|
def mock_classify(name):
|
|
if "ZC99" in name:
|
|
return CclfFilename(
|
|
program="sssp",
|
|
aco_id="1234",
|
|
entity="ACO",
|
|
file_id="99",
|
|
cclf_table="cclf99",
|
|
run_type="Y",
|
|
performance_year=2025,
|
|
delivery_date="250716",
|
|
delivery_time="1234567",
|
|
is_zip=False,
|
|
)
|
|
return None
|
|
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
with patch("aco.load.cclf.classify", side_effect=mock_classify):
|
|
stats = load_cclf_directory(tmp_path, database=db_path, run_pipeline=False)
|
|
assert "cclf99" not in stats # line 201
|
|
|
|
|
|
# ── aco/lake/jobs.py gaps (lines 152,154,160-162,164,171,176-177,183-184) ──
|
|
|
|
|
|
class TestJobManagerCreateAllJobs:
|
|
"""Cover create_all_jobs lines."""
|
|
|
|
def test_create_all_jobs(self):
|
|
from aco.lake.jobs import JobManager
|
|
|
|
client = MagicMock()
|
|
client._ws.jobs.create.return_value = MagicMock(job_id=55)
|
|
|
|
mgr = JobManager(client, catalog="aco")
|
|
result = mgr.create_all_jobs()
|
|
assert result == {"all": 55} # lines 152-184
|
|
|
|
def test_create_all_jobs_with_schedule(self):
|
|
from aco.lake.jobs import JobManager
|
|
|
|
client = MagicMock()
|
|
client._ws.jobs.create.return_value = MagicMock(job_id=66)
|
|
|
|
mgr = JobManager(client, catalog="aco")
|
|
result = mgr.create_all_jobs(schedule="0 0 6 * * ?")
|
|
assert result == {"all": 66}
|
|
|
|
call_kwargs = client._ws.jobs.create.call_args.kwargs
|
|
assert call_kwargs["schedule"] is not None # line 176-177
|
|
|
|
|
|
# ── aco/load/seed.py gaps (lines 27-31, 80, 94, 100-101, 104) ──
|
|
|
|
|
|
class TestSeedReadTabular:
|
|
"""Cover lines 27-31 (_read_tabular formats)."""
|
|
|
|
def test_xlsx_not_supported_extension(self, tmp_path):
|
|
from aco.load.seed import _read_tabular
|
|
|
|
f = tmp_path / "test.json"
|
|
f.write_text("{}")
|
|
with pytest.raises(ValueError, match="Unsupported"):
|
|
_read_tabular(f) # line 31
|
|
|
|
def test_csv(self, tmp_path):
|
|
from aco.load.seed import _read_tabular
|
|
|
|
f = tmp_path / "test.csv"
|
|
f.write_text("a,b\n1,2\n")
|
|
df = _read_tabular(f)
|
|
assert len(df) == 1 # line 26
|
|
|
|
def test_parquet(self, tmp_path):
|
|
from aco.load.seed import _read_tabular
|
|
|
|
f = tmp_path / "test.parquet"
|
|
pl.DataFrame({"x": [1]}).write_parquet(f)
|
|
df = _read_tabular(f)
|
|
assert len(df) == 1 # line 30
|
|
|
|
|
|
class TestLoadSeeds:
|
|
"""Cover lines 80, 94, 100-101, 104."""
|
|
|
|
def test_load_seeds_default_dir(self, tmp_path):
|
|
"""Cover line 80 (seed_dir is None, uses ROOT/dev/seeds)."""
|
|
from aco.load.seed import load_seeds
|
|
|
|
seed_dir = tmp_path / "dev" / "seeds"
|
|
seed_dir.mkdir(parents=True)
|
|
(seed_dir / "test.csv").write_text("a,b\n1,2\n")
|
|
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
with patch("conf.ROOT", tmp_path):
|
|
stats = load_seeds(database=db_path)
|
|
assert "reference_data.test" in stats
|
|
|
|
def test_skips_dirs_and_hidden(self, tmp_path):
|
|
"""Cover line 94 (skip dirs/hidden)."""
|
|
from aco.load.seed import load_seeds
|
|
|
|
(tmp_path / ".hidden.csv").write_text("a\n1\n")
|
|
(tmp_path / "subdir").mkdir()
|
|
(tmp_path / "good.csv").write_text("a\n1\n")
|
|
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
stats = load_seeds(tmp_path, database=db_path)
|
|
assert len(stats) == 1 # only good.csv
|
|
|
|
def test_skips_unreadable_file(self, tmp_path):
|
|
"""Cover lines 100-101 (exception reading file)."""
|
|
from aco.load.seed import load_seeds
|
|
|
|
bad = tmp_path / "bad.csv"
|
|
bad.write_text("this is not valid csv\x00\x01\x02")
|
|
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
with patch("aco.load.seed._read_tabular", side_effect=Exception("parse error")):
|
|
stats = load_seeds(tmp_path, database=db_path)
|
|
assert stats == {} # lines 100-101 continue
|
|
|
|
def test_skips_empty_dataframe(self, tmp_path):
|
|
"""Cover line 104 (empty DataFrame)."""
|
|
from aco.load.seed import load_seeds
|
|
|
|
f = tmp_path / "empty.csv"
|
|
f.write_text("a,b\n") # headers only
|
|
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
stats = load_seeds(tmp_path, database=db_path)
|
|
assert "reference_data.empty" not in stats # line 104
|
|
|
|
def test_seed_dir_not_found(self, tmp_path):
|
|
from aco.load.seed import load_seeds
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
load_seeds(tmp_path / "nonexistent")
|
|
|
|
|
|
# ── aco/lake/catalog.py gaps (lines 339, 341-347) ───────────────
|
|
|
|
|
|
class TestCatalogPipelineSchemas:
|
|
"""Cover lines 339, 341-347 (pipeline_schemas)."""
|
|
|
|
def test_pipeline_schemas(self):
|
|
from aco.lake.catalog import Catalog
|
|
|
|
schemas = Catalog.pipeline_schemas()
|
|
assert isinstance(schemas, list)
|
|
assert len(schemas) > 0
|
|
assert all("." not in s for s in schemas) # schema names only
|
|
assert schemas == sorted(schemas) # sorted
|
|
|
|
|
|
# ── aco/lake/lineage.py gaps (lines 45, 93-99) ──────────────────
|
|
|
|
|
|
class TestLineageColumnSources:
|
|
"""Cover lines 93-99 (column_sources from Field metadata)."""
|
|
|
|
def test_build_lineage_includes_columns(self):
|
|
from aco.lake.lineage import build_lineage
|
|
|
|
graph = build_lineage()
|
|
assert isinstance(graph.table_edges, dict)
|
|
assert len(graph.table_edges) > 0 # line 45 implicit
|
|
|
|
def test_column_lineage_returns_none(self):
|
|
from aco.lake.lineage import LineageGraph
|
|
|
|
g = LineageGraph()
|
|
assert g.column_lineage("no.table", "no_col") is None # line 45
|
|
|
|
|
|
# ── aco/pipe/runner.py gaps (lines 143-144,151-152,207,230) ─────
|
|
|
|
|
|
class TestRunnerPerfAndHooks:
|
|
"""Cover lines 143-144 (perf.collector ImportError) and 151-152 (perf.hooks ImportError)."""
|
|
|
|
def test_run_steps_no_perf(self):
|
|
"""Lines 143-144, 151-152, 207, 230 — run a trivial pipeline with perf mocked out."""
|
|
from aco.pipe.runner import run_pipeline
|
|
|
|
# A minimal expr: one step, no schema
|
|
def identity(core_encounter):
|
|
return core_encounter
|
|
|
|
mock_df = MagicMock(__len__=lambda s: 3, columns=["a", "b"])
|
|
mock_load = MagicMock(return_value=mock_df)
|
|
|
|
exprs = [MagicMock(name="test.output")]
|
|
exprs[0].__len__ = lambda s: 2
|
|
exprs[0].__iter__ = lambda s: iter(("test.output", identity))
|
|
exprs[0].__getitem__ = lambda s, i: ("test.output", identity)[i]
|
|
|
|
# Remove perf modules to trigger ImportError paths
|
|
import sys
|
|
|
|
saved_collector = sys.modules.pop("perf.collector", None)
|
|
saved_hooks = sys.modules.pop("perf.hooks", None)
|
|
sys.modules["perf.collector"] = None # will cause ImportError
|
|
sys.modules["perf.hooks"] = None
|
|
try:
|
|
cache = run_pipeline(exprs, mock_load)
|
|
assert "test.output" in cache # line 207, 230
|
|
finally:
|
|
if saved_collector is not None:
|
|
sys.modules["perf.collector"] = saved_collector
|
|
else:
|
|
sys.modules.pop("perf.collector", None)
|
|
if saved_hooks is not None:
|
|
sys.modules["perf.hooks"] = saved_hooks
|
|
else:
|
|
sys.modules.pop("perf.hooks", None)
|
|
|
|
|
|
# ── aco/lake/load.py gaps (lines 56-58, 60, 62, 87) ─────────────
|
|
|
|
|
|
class TestLoadToIceberg:
|
|
"""Cover lines 56-58, 60, 62, 87."""
|
|
|
|
def test_polaris_catalog_type(self):
|
|
"""Cover lines 56-58 (polaris config path) and 87 (empty tables list)."""
|
|
from aco.lake.load import load_to_iceberg
|
|
|
|
mock_cat_inst = MagicMock()
|
|
mock_cat_inst.schemas.return_value = ["core"]
|
|
# tables() returns empty list, triggering `if not tables: continue` (line 87)
|
|
mock_cat_inst.tables.return_value = []
|
|
|
|
mock_duckdb_ctx = MagicMock()
|
|
mock_duckdb_ctx.load.return_value = MagicMock(__len__=lambda s: 5)
|
|
|
|
import aco.lake.catalog as cm
|
|
import aco.lake.context as xm
|
|
|
|
orig = (cm.Catalog, xm.DuckDBContext, xm.IcebergContext)
|
|
|
|
def make_cat(**kw):
|
|
return mock_cat_inst
|
|
|
|
cm.Catalog = make_cat
|
|
xm.DuckDBContext = lambda **kw: mock_duckdb_ctx
|
|
xm.IcebergContext = lambda **kw: MagicMock()
|
|
try:
|
|
mock_cfg = MagicMock()
|
|
mock_cfg.lake.polaris.catalog_uri = "http://polaris"
|
|
mock_cfg.lake.warehouse = "wh"
|
|
mock_cfg.path.return_value = "/fake"
|
|
with patch("conf.cfg", mock_cfg):
|
|
result = load_to_iceberg(catalog_type="polaris")
|
|
finally:
|
|
cm.Catalog, xm.DuckDBContext, xm.IcebergContext = orig
|
|
# "core" schema had no tables so it was skipped entirely (line 87)
|
|
assert "core" not in result
|
|
|
|
def test_nessie_catalog_type(self):
|
|
"""Cover line 60 (nessie config path)."""
|
|
from aco.lake.load import load_to_iceberg
|
|
|
|
mock_cat_inst = MagicMock()
|
|
mock_cat_inst.schemas.return_value = []
|
|
|
|
import aco.lake.catalog as cm
|
|
import aco.lake.context as xm
|
|
|
|
orig = (cm.Catalog, xm.DuckDBContext, xm.IcebergContext)
|
|
cm.Catalog = lambda **kw: mock_cat_inst
|
|
xm.DuckDBContext = lambda **kw: MagicMock()
|
|
xm.IcebergContext = lambda **kw: MagicMock()
|
|
try:
|
|
mock_cfg = MagicMock()
|
|
mock_cfg.lake.nessie.catalog_uri = "http://nessie"
|
|
mock_cfg.lake.warehouse = "wh"
|
|
mock_cfg.path.return_value = "/fake"
|
|
with patch("conf.cfg", mock_cfg):
|
|
result = load_to_iceberg(catalog_type="nessie")
|
|
finally:
|
|
cm.Catalog, xm.DuckDBContext, xm.IcebergContext = orig
|
|
assert result == {}
|
|
|
|
|
|
# ── aco/lake/governance.py gaps (lines 150-151, 160) ────────────
|
|
|
|
|
|
class TestGovernanceExceptionInGrants:
|
|
"""Cover lines 150-151, 160."""
|
|
|
|
def test_audit_governance_exception(self):
|
|
from aco.lake.governance import GovernancePolicy, GrantRule, audit_governance
|
|
|
|
mock_client = MagicMock()
|
|
ws = mock_client._ws
|
|
|
|
schema_mock = MagicMock()
|
|
schema_mock.name = "core"
|
|
ws.schemas.list.return_value = [schema_mock]
|
|
ws.grants.get.side_effect = Exception("no perms")
|
|
|
|
policy = GovernancePolicy(
|
|
catalog="aco",
|
|
rules=[GrantRule(schemas=["*"], group="team", privileges=["SELECT"])],
|
|
)
|
|
|
|
drifts = audit_governance(mock_client, policy)
|
|
assert len(drifts) >= 1
|
|
assert drifts[0].action == "missing"
|
|
|
|
|
|
# ── aco/express/base.py gaps (lines 45-46) ──────────────────────
|
|
|
|
|
|
class TestExprTagImportError:
|
|
"""Cover lines 45-46 (Tag import failure)."""
|
|
|
|
def test_tag_import_optional(self):
|
|
"""The try/except for bib.tag.Tag at import time just sets Tag=None."""
|
|
from aco.express.base import Expr
|
|
|
|
assert Expr is not None
|
|
|
|
|
|
# ── aco/lake/snapshot.py gaps (lines 43-44) ─────────────────────
|
|
|
|
|
|
class TestSnapshotGitShaFallback:
|
|
"""Cover lines 43-44 (_git_sha exception)."""
|
|
|
|
def test_git_sha_returns_unknown(self, tmp_path):
|
|
from aco.lake.snapshot import SnapshotManager
|
|
|
|
mgr = SnapshotManager(str(tmp_path / "test.duckdb"))
|
|
with patch("aco.lake.snapshot.subprocess.run", side_effect=Exception("no git")):
|
|
sha = mgr._git_sha()
|
|
assert sha == "unknown" # lines 43-44
|
|
|
|
|
|
# ── aco/lake/quality.py gaps (lines 150-151) ────────────────────
|
|
|
|
|
|
class TestQualityListMonitorsException:
|
|
"""Cover lines 150-151."""
|
|
|
|
def test_list_monitors_table_exception(self):
|
|
from aco.lake.quality import list_monitors
|
|
|
|
mock_client = MagicMock()
|
|
ws = mock_client._ws
|
|
|
|
schema_info = MagicMock()
|
|
schema_info.name = "core"
|
|
ws.schemas.list.return_value = [schema_info]
|
|
|
|
table = MagicMock()
|
|
table.full_name = "aco.core.encounter"
|
|
table.name = "encounter"
|
|
ws.tables.list.return_value = [table]
|
|
ws.quality_monitors.get.side_effect = Exception("no monitor")
|
|
|
|
result = list_monitors(mock_client, catalog="aco")
|
|
assert isinstance(result, list)
|
|
assert len(result) == 0 # exception swallowed, no results
|
|
|
|
|
|
# ── aco/load/bcda.py gaps (lines 51, 61) ────────────────────────
|
|
|
|
|
|
class TestLoadBcdaFlatDirNotFound:
|
|
"""Cover line 61 (flat_dir not found)."""
|
|
|
|
def test_flat_dir_missing(self, tmp_path):
|
|
from aco.load.bcda import load_bcda
|
|
|
|
store = tmp_path / "bcda"
|
|
store.mkdir()
|
|
ndjson = tmp_path / "ndjson"
|
|
ndjson.mkdir()
|
|
|
|
with (
|
|
patch("conf.path", return_value=store),
|
|
pytest.raises(FileNotFoundError, match="No flattened"),
|
|
):
|
|
load_bcda(ndjson, skip_flatten=True)
|
|
|
|
|
|
class TestLoadBcdaFindLatest:
|
|
"""Cover line 51 (_find_latest_export)."""
|
|
|
|
def test_find_latest_export(self, tmp_path):
|
|
from aco.load.bcda import _find_latest_export
|
|
|
|
exports = tmp_path / "exports"
|
|
exports.mkdir()
|
|
(exports / "job1").mkdir()
|
|
|
|
result = _find_latest_export(tmp_path)
|
|
assert result.name == "job1"
|
|
|
|
def test_find_latest_no_exports_dir(self, tmp_path):
|
|
from aco.load.bcda import _find_latest_export
|
|
|
|
with pytest.raises(FileNotFoundError, match="No exports"):
|
|
_find_latest_export(tmp_path)
|
|
|
|
def test_find_latest_empty_exports(self, tmp_path):
|
|
from aco.load.bcda import _find_latest_export
|
|
|
|
(tmp_path / "exports").mkdir()
|
|
with pytest.raises(FileNotFoundError, match="No export directories"):
|
|
_find_latest_export(tmp_path)
|
|
|
|
|
|
# ── aco/load/stage.py gap (line 117) ────────────────────────────
|
|
|
|
|
|
class TestStageNoQualifiedRef:
|
|
"""Cover line 117 ('.' not in table_ref → skip)."""
|
|
|
|
def test_no_dot_skipped(self):
|
|
from aco.load.stage import _promote_to_iceberg
|
|
|
|
mock_cfg = MagicMock()
|
|
mock_cfg.lake.nessie.catalog_uri = "http://nessie"
|
|
mock_cfg.lake.warehouse = "wh"
|
|
mock_cfg.lake.get.return_value = "http://rustfs"
|
|
mock_cfg.services.rustfs = "http://rustfs"
|
|
|
|
mock_ice = MagicMock()
|
|
|
|
import aco.lake.context as xm
|
|
|
|
orig_duck, orig_ice = xm.DuckDBContext, xm.IcebergContext
|
|
xm.DuckDBContext = lambda **kw: MagicMock()
|
|
xm.IcebergContext = lambda **kw: mock_ice
|
|
try:
|
|
with patch("conf.cfg", mock_cfg):
|
|
_promote_to_iceberg(
|
|
{"bcda": ["plain_table"]},
|
|
database="fake.db",
|
|
)
|
|
finally:
|
|
xm.DuckDBContext, xm.IcebergContext = orig_duck, orig_ice
|
|
mock_ice.save.assert_not_called()
|
|
|
|
|
|
def _make_import_raiser(module_name):
|
|
"""Create __import__ replacement that raises ImportError for a module."""
|
|
real_import = (
|
|
__builtins__["__import__"]
|
|
if isinstance(__builtins__, dict)
|
|
else __builtins__.__import__
|
|
)
|
|
|
|
def _import(name, *args, **kwargs):
|
|
if name == module_name:
|
|
raise ImportError(f"mocked: {name}")
|
|
return real_import(name, *args, **kwargs)
|
|
|
|
return _import
|