1baf98eb40
- CCLF: fixed-width parser using IP-derived field positions (cclf_layout.py), ZIP extraction, file discovery by CMS naming convention, DuckDB loading, optional pipeline execution to produce input_layer tables - BCDA: flatten ndjson → Parquet, load into bcda schema in DuckDB - Seeds: auto-discover CSV/Excel/Parquet files, load into reference_data schema - Unified staging: orchestrate all three loaders with optional Iceberg promotion - CLI: stack load cclf/bcda/seed with full options (--path, --database, etc.) - 57 tests covering parsers, loaders, CLI commands, and staging pipeline fixes #5 fixes #6 fixes #7 fixes #8
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
"""Tests for seed data loading."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import duckdb
|
|
import pytest
|
|
|
|
from aco.load.seed import _table_name, load_seeds
|
|
|
|
|
|
class TestTableName:
|
|
def test_simple_csv(self) -> None:
|
|
assert _table_name(Path("fips_county.csv")) == "fips_county"
|
|
|
|
def test_xlsx_with_parens(self) -> None:
|
|
result = _table_name(Path("ACO_REACH_bulk_upload (1).xlsx"))
|
|
assert result == "aco_reach_bulk_upload"
|
|
|
|
def test_spaces_become_underscores(self) -> None:
|
|
result = _table_name(Path("Static Report Crosswalk.xlsx"))
|
|
assert result == "static_report_crosswalk"
|
|
|
|
|
|
class TestLoadSeeds:
|
|
def test_no_dir_raises(self, tmp_path: Path) -> None:
|
|
with pytest.raises(FileNotFoundError):
|
|
load_seeds(seed_dir=tmp_path / "nonexistent")
|
|
|
|
def test_load_csv_seed(self, tmp_path: Path) -> None:
|
|
# Create a CSV seed
|
|
csv = tmp_path / "test_codes.csv"
|
|
csv.write_text("code,description\nA01,Test Code A\nB02,Test Code B\n")
|
|
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
stats = load_seeds(seed_dir=tmp_path, database=db_path)
|
|
|
|
assert "reference_data.test_codes" in stats
|
|
assert stats["reference_data.test_codes"] == 2
|
|
|
|
# Verify data
|
|
con = duckdb.connect(db_path, read_only=True)
|
|
rows = con.execute(
|
|
"SELECT * FROM reference_data.test_codes ORDER BY code"
|
|
).fetchall()
|
|
assert len(rows) == 2
|
|
assert rows[0][0] == "A01"
|
|
con.close()
|
|
|
|
def test_skips_non_tabular(self, tmp_path: Path) -> None:
|
|
(tmp_path / "readme.txt").write_text("not a seed")
|
|
(tmp_path / "data.pdf").write_text("not tabular")
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
stats = load_seeds(seed_dir=tmp_path, database=db_path)
|
|
assert stats == {}
|
|
|
|
def test_custom_schema(self, tmp_path: Path) -> None:
|
|
csv = tmp_path / "test.csv"
|
|
csv.write_text("x\n1\n")
|
|
db_path = str(tmp_path / "test.duckdb")
|
|
stats = load_seeds(seed_dir=tmp_path, database=db_path, schema="terminology")
|
|
assert "terminology.test" in stats
|