8f99eeb154
- 5519 unit tests covering all modules (aco, bcda, bls, cms, pfs, rex, bib) - ruff lint + format enforcement across entire codebase (377 files reformatted) - pre-commit hook: ruff check, ruff format, pytest - Woodpecker CI split into ci.yml (quality gate) and deploy.yml (package + images) - ci.yml: lint → test → validate-compose, runs on every push/PR - deploy.yml: build + publish Python package to Gitea PyPI registry, then container image builds, Trivy scans, and registry push (main branch only) - Gitea branch protection on main: requires CI status checks to pass - .gitignore updated for .coverage, dist/, *.egg-info/ - grafana config moved to dev/grafana/ Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
431 lines
14 KiB
Python
431 lines
14 KiB
Python
"""Tests for aco.express.provider_attribution — staging and
|
|
int_person_years functions.
|
|
|
|
Tests verify column selection, renames, and the person-year
|
|
grouping/casting logic.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import polars as pl
|
|
import pytest
|
|
|
|
from aco.express.provider_attribution import (
|
|
int_person_years,
|
|
stg_core__claims_medical_claim,
|
|
stg_core__medical_claim,
|
|
stg_core__member_months,
|
|
stg_reference_data__calendar,
|
|
stg_terminology__provider,
|
|
)
|
|
|
|
# ── helpers ──────────────────────────────────────────────────────
|
|
|
|
|
|
def _cols(result: pl.DataFrame, *expected: str) -> None:
|
|
missing = [c for c in expected if c not in result.columns]
|
|
assert not missing, f"Missing columns: {missing}"
|
|
|
|
|
|
# ── stg_core__claims_medical_claim ───────────────────────────────
|
|
|
|
|
|
class TestStgCoreClaimsMedicalClaim:
|
|
"""Selects claim_id, claim_line_number, data_source,
|
|
encounter_id."""
|
|
|
|
@pytest.fixture
|
|
def input_df(self) -> pl.DataFrame:
|
|
return pl.DataFrame(
|
|
{
|
|
"claim_id": ["C1", "C2"],
|
|
"claim_line_number": [1, 2],
|
|
"data_source": ["src_a", "src_b"],
|
|
"encounter_id": ["E1", "E2"],
|
|
"extra_col": ["x", "y"],
|
|
}
|
|
)
|
|
|
|
def test_returns_dataframe(self, input_df) -> None:
|
|
result = stg_core__claims_medical_claim(input_df)
|
|
assert isinstance(result, pl.DataFrame)
|
|
|
|
def test_selects_four_columns(self, input_df) -> None:
|
|
result = stg_core__claims_medical_claim(input_df)
|
|
assert set(result.columns) == {
|
|
"claim_id",
|
|
"claim_line_number",
|
|
"data_source",
|
|
"encounter_id",
|
|
}
|
|
|
|
def test_excludes_extra_columns(self, input_df) -> None:
|
|
result = stg_core__claims_medical_claim(input_df)
|
|
assert "extra_col" not in result.columns
|
|
|
|
def test_preserves_row_count(self, input_df) -> None:
|
|
result = stg_core__claims_medical_claim(input_df)
|
|
assert len(result) == 2
|
|
|
|
def test_values_passed_through(self, input_df) -> None:
|
|
result = stg_core__claims_medical_claim(input_df)
|
|
assert result["claim_id"].to_list() == [
|
|
"C1",
|
|
"C2",
|
|
]
|
|
|
|
|
|
# ── stg_core__medical_claim ──────────────────────────────────────
|
|
|
|
|
|
class TestStgCoreMedicalClaim:
|
|
"""Selects 11 columns, renames rendering_id to
|
|
rendering_npi."""
|
|
|
|
@pytest.fixture
|
|
def input_df(self) -> pl.DataFrame:
|
|
return pl.DataFrame(
|
|
{
|
|
"claim_id": ["C1"],
|
|
"claim_line_number": [1],
|
|
"person_id": ["P1"],
|
|
"claim_start_date": ["2024-01-15"],
|
|
"claim_end_date": ["2024-01-20"],
|
|
"allowed_amount": [100.0],
|
|
"paid_amount": [80.0],
|
|
"rendering_id": ["NPI123"],
|
|
"hcpcs_code": ["99213"],
|
|
"data_source": ["test"],
|
|
"encounter_id": ["E1"],
|
|
"extra": ["drop_me"],
|
|
}
|
|
)
|
|
|
|
def test_returns_dataframe(self, input_df) -> None:
|
|
result = stg_core__medical_claim(input_df)
|
|
assert isinstance(result, pl.DataFrame)
|
|
|
|
def test_renames_rendering_id(self, input_df) -> None:
|
|
result = stg_core__medical_claim(input_df)
|
|
assert "rendering_npi" in result.columns
|
|
assert "rendering_id" not in result.columns
|
|
assert result["rendering_npi"][0] == "NPI123"
|
|
|
|
def test_selects_11_columns(self, input_df) -> None:
|
|
result = stg_core__medical_claim(input_df)
|
|
expected = {
|
|
"claim_id",
|
|
"claim_line_number",
|
|
"person_id",
|
|
"claim_start_date",
|
|
"claim_end_date",
|
|
"allowed_amount",
|
|
"paid_amount",
|
|
"rendering_npi",
|
|
"hcpcs_code",
|
|
"data_source",
|
|
"encounter_id",
|
|
}
|
|
assert set(result.columns) == expected
|
|
|
|
def test_excludes_extra(self, input_df) -> None:
|
|
result = stg_core__medical_claim(input_df)
|
|
assert "extra" not in result.columns
|
|
|
|
def test_preserves_values(self, input_df) -> None:
|
|
result = stg_core__medical_claim(input_df)
|
|
assert result["claim_id"][0] == "C1"
|
|
assert result["person_id"][0] == "P1"
|
|
assert result["allowed_amount"][0] == pytest.approx(100.0)
|
|
|
|
|
|
# ── stg_core__member_months ──────────────────────────────────────
|
|
|
|
|
|
class TestStgCoreMemberMonths:
|
|
"""Selects person_id and year_month."""
|
|
|
|
@pytest.fixture
|
|
def input_df(self) -> pl.DataFrame:
|
|
return pl.DataFrame(
|
|
{
|
|
"person_id": ["P1", "P2", "P3"],
|
|
"year_month": [
|
|
"202401",
|
|
"202402",
|
|
"202403",
|
|
],
|
|
"payer": ["MCR", "MCR", "MCR"],
|
|
}
|
|
)
|
|
|
|
def test_returns_dataframe(self, input_df) -> None:
|
|
result = stg_core__member_months(input_df)
|
|
assert isinstance(result, pl.DataFrame)
|
|
|
|
def test_selects_two_columns(self, input_df) -> None:
|
|
result = stg_core__member_months(input_df)
|
|
assert set(result.columns) == {
|
|
"person_id",
|
|
"year_month",
|
|
}
|
|
|
|
def test_excludes_payer(self, input_df) -> None:
|
|
result = stg_core__member_months(input_df)
|
|
assert "payer" not in result.columns
|
|
|
|
def test_preserves_row_count(self, input_df) -> None:
|
|
result = stg_core__member_months(input_df)
|
|
assert len(result) == 3
|
|
|
|
def test_values_passed_through(self, input_df) -> None:
|
|
result = stg_core__member_months(input_df)
|
|
assert result["person_id"].to_list() == [
|
|
"P1",
|
|
"P2",
|
|
"P3",
|
|
]
|
|
assert result["year_month"].to_list() == [
|
|
"202401",
|
|
"202402",
|
|
"202403",
|
|
]
|
|
|
|
|
|
# ── stg_reference_data__calendar ─────────────────────────────────
|
|
|
|
|
|
class TestStgReferenceDataCalendar:
|
|
"""Selects 7 date-related columns."""
|
|
|
|
@pytest.fixture
|
|
def input_df(self) -> pl.DataFrame:
|
|
from datetime import date
|
|
|
|
return pl.DataFrame(
|
|
{
|
|
"full_date": [date(2024, 1, 15)],
|
|
"year": [2024],
|
|
"month": [1],
|
|
"year_month": ["202401"],
|
|
"first_day_of_month": [date(2024, 1, 1)],
|
|
"last_day_of_month": [date(2024, 1, 31)],
|
|
"year_month_int": [202401],
|
|
"day_of_week": ["Monday"],
|
|
"quarter": [1],
|
|
}
|
|
)
|
|
|
|
def test_returns_dataframe(self, input_df) -> None:
|
|
result = stg_reference_data__calendar(input_df)
|
|
assert isinstance(result, pl.DataFrame)
|
|
|
|
def test_selects_seven_columns(self, input_df) -> None:
|
|
result = stg_reference_data__calendar(input_df)
|
|
expected = {
|
|
"full_date",
|
|
"year",
|
|
"month",
|
|
"year_month",
|
|
"first_day_of_month",
|
|
"last_day_of_month",
|
|
"year_month_int",
|
|
}
|
|
assert set(result.columns) == expected
|
|
|
|
def test_excludes_extra(self, input_df) -> None:
|
|
result = stg_reference_data__calendar(input_df)
|
|
assert "day_of_week" not in result.columns
|
|
assert "quarter" not in result.columns
|
|
|
|
def test_values_passed_through(self, input_df) -> None:
|
|
from datetime import date
|
|
|
|
result = stg_reference_data__calendar(input_df)
|
|
assert result["full_date"][0] == date(2024, 1, 15)
|
|
assert result["year"][0] == 2024
|
|
assert result["year_month_int"][0] == 202401
|
|
|
|
|
|
# ── stg_terminology__provider ────────────────────────────────────
|
|
|
|
|
|
class TestStgTerminologyProvider:
|
|
"""Selects npi, primary_taxonomy_code,
|
|
primary_specialty_description, entity_type_description.
|
|
"""
|
|
|
|
@pytest.fixture
|
|
def input_df(self) -> pl.DataFrame:
|
|
return pl.DataFrame(
|
|
{
|
|
"npi": ["1234567890", "9876543210"],
|
|
"primary_taxonomy_code": [
|
|
"207Q00000X",
|
|
"208D00000X",
|
|
],
|
|
"primary_specialty_description": [
|
|
"Family Medicine",
|
|
"General Practice",
|
|
],
|
|
"entity_type_description": [
|
|
"Individual",
|
|
"Organization",
|
|
],
|
|
"provider_first_name": [
|
|
"Alice",
|
|
"Bob",
|
|
],
|
|
"provider_last_name": [
|
|
"Smith",
|
|
"Jones",
|
|
],
|
|
}
|
|
)
|
|
|
|
def test_returns_dataframe(self, input_df) -> None:
|
|
result = stg_terminology__provider(input_df)
|
|
assert isinstance(result, pl.DataFrame)
|
|
|
|
def test_selects_four_columns(self, input_df) -> None:
|
|
result = stg_terminology__provider(input_df)
|
|
expected = {
|
|
"npi",
|
|
"primary_taxonomy_code",
|
|
"primary_specialty_description",
|
|
"entity_type_description",
|
|
}
|
|
assert set(result.columns) == expected
|
|
|
|
def test_excludes_name_columns(self, input_df) -> None:
|
|
result = stg_terminology__provider(input_df)
|
|
assert "provider_first_name" not in result.columns
|
|
assert "provider_last_name" not in result.columns
|
|
|
|
def test_preserves_row_count(self, input_df) -> None:
|
|
result = stg_terminology__provider(input_df)
|
|
assert len(result) == 2
|
|
|
|
def test_values_passed_through(self, input_df) -> None:
|
|
result = stg_terminology__provider(input_df)
|
|
assert result["npi"].to_list() == [
|
|
"1234567890",
|
|
"9876543210",
|
|
]
|
|
assert result["entity_type_description"].to_list() == [
|
|
"Individual",
|
|
"Organization",
|
|
]
|
|
|
|
|
|
# ── int_person_years ─────────────────────────────────────────────
|
|
|
|
|
|
class TestIntPersonYears:
|
|
"""int_person_years groups member_months by person_id +
|
|
first 4 chars of year_month (performance_year), casts
|
|
performance_year to Int32."""
|
|
|
|
@pytest.fixture
|
|
def member_months_df(self) -> pl.DataFrame:
|
|
return pl.DataFrame(
|
|
{
|
|
"person_id": [
|
|
"P1",
|
|
"P1",
|
|
"P1",
|
|
"P2",
|
|
"P2",
|
|
],
|
|
"year_month": [
|
|
"202401",
|
|
"202402",
|
|
"202403",
|
|
"202401",
|
|
"202501",
|
|
],
|
|
}
|
|
)
|
|
|
|
def test_returns_dataframe(self, member_months_df) -> None:
|
|
result = int_person_years(member_months_df)
|
|
assert isinstance(result, pl.DataFrame)
|
|
|
|
def test_output_columns(self, member_months_df) -> None:
|
|
result = int_person_years(member_months_df)
|
|
assert set(result.columns) == {
|
|
"person_id",
|
|
"performance_year",
|
|
}
|
|
|
|
def test_groups_by_person_and_year(self, member_months_df) -> None:
|
|
result = int_person_years(member_months_df)
|
|
# P1 has 3 months all in 2024 -> 1 row
|
|
# P2 has months in 2024 and 2025 -> 2 rows
|
|
assert len(result) == 3
|
|
|
|
def test_performance_year_is_int32(self, member_months_df) -> None:
|
|
result = int_person_years(member_months_df)
|
|
assert result["performance_year"].dtype == pl.Int32
|
|
|
|
def test_person_id_is_string(self, member_months_df) -> None:
|
|
result = int_person_years(member_months_df)
|
|
assert result["person_id"].dtype == pl.Utf8
|
|
|
|
def test_performance_year_values(self, member_months_df) -> None:
|
|
result = int_person_years(member_months_df)
|
|
years = sorted(result["performance_year"].to_list())
|
|
assert years == [2024, 2024, 2025]
|
|
|
|
def test_single_person_single_year(self) -> None:
|
|
df = pl.DataFrame(
|
|
{
|
|
"person_id": ["P1"],
|
|
"year_month": ["202406"],
|
|
}
|
|
)
|
|
result = int_person_years(df)
|
|
assert len(result) == 1
|
|
assert result["person_id"][0] == "P1"
|
|
assert result["performance_year"][0] == 2024
|
|
|
|
def test_multiple_months_same_year_one_row(
|
|
self,
|
|
) -> None:
|
|
df = pl.DataFrame(
|
|
{
|
|
"person_id": ["P1"] * 12,
|
|
"year_month": [f"2024{m:02d}" for m in range(1, 13)],
|
|
}
|
|
)
|
|
result = int_person_years(df)
|
|
assert len(result) == 1
|
|
|
|
def test_integer_year_month_input(self) -> None:
|
|
"""year_month is cast to String before slicing,
|
|
so integer inputs should also work."""
|
|
df = pl.DataFrame(
|
|
{
|
|
"person_id": ["P1", "P1"],
|
|
"year_month": [202401, 202502],
|
|
}
|
|
)
|
|
result = int_person_years(df)
|
|
assert len(result) == 2
|
|
years = sorted(result["performance_year"].to_list())
|
|
assert years == [2024, 2025]
|
|
|
|
def test_empty_input(self) -> None:
|
|
df = pl.DataFrame(
|
|
{
|
|
"person_id": pl.Series([], dtype=pl.Utf8),
|
|
"year_month": pl.Series([], dtype=pl.Utf8),
|
|
}
|
|
)
|
|
result = int_person_years(df)
|
|
assert len(result) == 0
|
|
assert set(result.columns) == {
|
|
"person_id",
|
|
"performance_year",
|
|
}
|