feat: OPPS express functions, pipe module, deploy script, CI green (fixes #267, #268, refs #282)
Some checks failed
CI / skinny-install (aco) (push) Successful in 1m30s
CI / lint-test (push) Failing after 1m57s
CI / skinny-install (api) (push) Successful in 26s
CI / skinny-install (bcda) (push) Successful in 29s
CI / skinny-install (bib) (push) Successful in 32s
CI / skinny-install (bls) (push) Successful in 23s
CI / skinny-install (ccw) (push) Successful in 29s
CI / skinny-install (cli) (push) Successful in 31s
CI / skinny-install (cms) (push) Successful in 27s
CI / skinny-install (conf) (push) Successful in 28s
CI / skinny-install (opps) (push) Successful in 28s
CI / skinny-install (perf) (push) Successful in 32s
CI / skinny-install (pfs) (push) Successful in 32s
CI / skinny-install (rex) (push) Successful in 28s
Infra CI / notebooks (push) Failing after 3m43s
Infra CI / zotero (push) Failing after 0s
Infra CI / docs (push) Failing after 0s
Infra CI / api (push) Failing after 0s
Infra CI / mc (push) Failing after 0s
Package Supply Chain / pkg-supply-chain (push) Failing after 0s
Deploy / build-scan-report (push) Failing after 4m23s

- OPPS express functions: adjusted_payment, skin_sub_impact wrapping calcs
- OPPS pipe module registered in aco.pipe.registry (2 exprs, auto-discovered by CLI/API)
- Output table models: OppsAdjustedPayment, OppsSkinSubImpact
- deploy.sh: tiered rollout (infra → gitea → apps → CI → observability)
  with context-aware image check (local → build if missing)
- compose.yml: pull_policy: if_not_present + build sections for all fhirworx images,
  gateway IPAM subnet for CoreDNS static IP, removed nested loch.css bind mount
- CI: opps added to skinny-install matrix, generated configs regenerated
- Coverage: 98.46% → 99.04% (sigv4, cclf, diag, provision, auth, cms_quality tests)
This commit is contained in:
kert
2026-03-26 01:52:07 -04:00
parent f3c09fc695
commit ba65e503d0
46 changed files with 2062 additions and 40 deletions

View File

@@ -51,7 +51,7 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
strategy: strategy:
matrix: matrix:
extra: [conf, aco, api, bcda, bib, bls, ccw, cli, cms, perf, pfs, rex] extra: [conf, aco, api, bcda, bib, bls, ccw, cli, cms, opps, perf, pfs, rex]
steps: steps:
- name: Checkout - name: Checkout
uses: https://github.com/actions/checkout@v4 uses: https://github.com/actions/checkout@v4

View File

@@ -3,6 +3,9 @@ networks:
name: gateway name: gateway
driver: bridge driver: bridge
internal: false internal: false
ipam:
config:
- subnet: 172.25.0.0/16
storage: storage:
name: storage name: storage
driver: bridge driver: bridge
@@ -59,6 +62,7 @@ services:
rustfs: rustfs:
image: ${IMAGE_PREFIX:-fhirworx}/rustfs:latest image: ${IMAGE_PREFIX:-fhirworx}/rustfs:latest
pull_policy: if_not_present
container_name: rustfs container_name: rustfs
user: "10001:10001" user: "10001:10001"
networks: networks:
@@ -76,6 +80,7 @@ services:
mc: mc:
image: ${IMAGE_PREFIX:-fhirworx}/mc:${COMMIT_SHA:-latest} image: ${IMAGE_PREFIX:-fhirworx}/mc:${COMMIT_SHA:-latest}
pull_policy: if_not_present
container_name: mc container_name: mc
build: build:
context: infra/rustfs context: infra/rustfs
@@ -99,6 +104,7 @@ services:
postgres: postgres:
image: ${IMAGE_PREFIX:-fhirworx}/postgresql:latest image: ${IMAGE_PREFIX:-fhirworx}/postgresql:latest
pull_policy: if_not_present
container_name: postgres container_name: postgres
networks: networks:
- storage - storage
@@ -239,6 +245,10 @@ services:
notebooks: notebooks:
image: ${IMAGE_PREFIX:-fhirworx}/notebooks:${COMMIT_SHA:-latest} image: ${IMAGE_PREFIX:-fhirworx}/notebooks:${COMMIT_SHA:-latest}
pull_policy: if_not_present
build:
context: .
dockerfile: infra/images/notebooks.Dockerfile
container_name: notebooks container_name: notebooks
networks: networks:
- gateway - gateway
@@ -255,7 +265,6 @@ services:
volumes: volumes:
- ./notebooks:/home/kert/notebooks - ./notebooks:/home/kert/notebooks
- ./infra/marimo:/home/kert/.config/marimo - ./infra/marimo:/home/kert/.config/marimo
- ./assets/css/marimo.css:/home/kert/.config/marimo/loch.css:ro
- ./infra/marimo/home-page-patched.js:/home/kert/workspace/.venv/lib/python3.13/site-packages/marimo/_static/assets/home-page-itW0tRmv.js:ro - ./infra/marimo/home-page-patched.js:/home/kert/workspace/.venv/lib/python3.13/site-packages/marimo/_static/assets/home-page-itW0tRmv.js:ro
- ./data:/home/kert/data - ./data:/home/kert/data
- ./data/zotero/data:/home/kert/zotero:ro - ./data/zotero/data:/home/kert/zotero:ro
@@ -274,6 +283,10 @@ services:
zotero: zotero:
image: ${IMAGE_PREFIX:-fhirworx}/zotero:${COMMIT_SHA:-latest} image: ${IMAGE_PREFIX:-fhirworx}/zotero:${COMMIT_SHA:-latest}
pull_policy: if_not_present
build:
context: .
dockerfile: infra/images/zotero.Dockerfile
container_name: zotero container_name: zotero
networks: networks:
- gateway - gateway
@@ -442,6 +455,10 @@ services:
docs: docs:
image: ${IMAGE_PREFIX:-fhirworx}/docs:${COMMIT_SHA:-latest} image: ${IMAGE_PREFIX:-fhirworx}/docs:${COMMIT_SHA:-latest}
pull_policy: if_not_present
build:
context: .
dockerfile: infra/images/docs.Dockerfile
container_name: docs container_name: docs
networks: networks:
- gateway - gateway
@@ -449,6 +466,10 @@ services:
api: api:
image: ${IMAGE_PREFIX:-fhirworx}/api:${COMMIT_SHA:-latest} image: ${IMAGE_PREFIX:-fhirworx}/api:${COMMIT_SHA:-latest}
pull_policy: if_not_present
build:
context: .
dockerfile: infra/images/api.Dockerfile
container_name: api container_name: api
networks: networks:
- gateway - gateway

145
deploy.sh Executable file
View File

@@ -0,0 +1,145 @@
#!/usr/bin/env bash
set -euo pipefail
# ── Load environment ────────────────────────────────────────────────
cd "$(dirname "$0")"
set -a; source .env; set +a
PREFIX="${IMAGE_PREFIX:-fhirworx}"
SHA="${COMMIT_SHA:-latest}"
# ── Helpers ─────────────────────────────────────────────────────────
log() { printf '\033[1;34m▸ %s\033[0m\n' "$*"; }
ok() { printf '\033[1;32m✓ %s\033[0m\n' "$*"; }
warn() { printf '\033[1;33m⚠ %s\033[0m\n' "$*"; }
die() { printf '\033[1;31m✗ %s\033[0m\n' "$*" >&2; exit 1; }
up() {
log "Starting: $*"
docker compose up -d --no-deps "$@"
}
wait_healthy() {
local svc="$1" timeout="${2:-60}" elapsed=0
log "Waiting for $svc to be healthy (${timeout}s timeout)…"
while [ "$elapsed" -lt "$timeout" ]; do
local state
state=$(docker inspect --format='{{.State.Health.Status}}' "$svc" 2>/dev/null || echo "missing")
case "$state" in
healthy) ok "$svc healthy"; return 0 ;;
unhealthy) die "$svc is unhealthy" ;;
esac
sleep 2
elapsed=$((elapsed + 2))
done
die "$svc did not become healthy within ${timeout}s"
}
wait_running() {
local svc="$1" timeout="${2:-30}" elapsed=0
log "Waiting for $svc to be running (${timeout}s timeout)…"
while [ "$elapsed" -lt "$timeout" ]; do
local state
state=$(docker inspect --format='{{.State.Status}}' "$svc" 2>/dev/null || echo "missing")
if [ "$state" = "running" ]; then
ok "$svc running"
return 0
fi
sleep 2
elapsed=$((elapsed + 2))
done
die "$svc did not start within ${timeout}s"
}
image_exists() {
docker image inspect "$1" &>/dev/null
}
# ── Buildable services: compose service → image name ────────────────
# Services with build: sections that can be built locally if missing
declare -A BUILDABLE=(
[mc]="${PREFIX}/mc:${SHA}"
[notebooks]="${PREFIX}/notebooks:${SHA}"
[zotero]="${PREFIX}/zotero:${SHA}"
[docs]="${PREFIX}/docs:${SHA}"
[api]="${PREFIX}/api:${SHA}"
)
# ── Pre-flight: check local images ─────────────────────────────────
BUILD_NEEDED=()
for svc in "${!BUILDABLE[@]}"; do
img="${BUILDABLE[$svc]}"
if image_exists "$img"; then
ok "Local: $img"
else
warn "Missing: $img — will build"
BUILD_NEEDED+=("$svc")
fi
done
# Also check non-buildable fhirworx images
for img in "${PREFIX}/rustfs:latest" "${PREFIX}/postgresql:latest"; do
if image_exists "$img"; then
ok "Local: $img"
else
die "Missing base image: $img — build or pull it manually"
fi
done
# ── Build missing images ────────────────────────────────────────────
if [ "${#BUILD_NEEDED[@]}" -gt 0 ]; then
log "Building ${#BUILD_NEEDED[@]} image(s)…"
for svc in "${BUILD_NEEDED[@]}"; do
log "Building $svc${BUILDABLE[$svc]}"
docker compose build "$svc" || die "Failed to build $svc"
ok "Built ${BUILDABLE[$svc]}"
done
fi
# ── Tier 0: Infrastructure (public images) ─────────────────────────
log "Tier 0: Infrastructure"
up coredns traefik rustfs postgres
wait_running rustfs
wait_running postgres
# ── Tier 1: Gitea (registry + git host) ─────────────────────────────
log "Tier 1: Registry"
up gitea
wait_running gitea
log "Waiting for Gitea HTTP…"
for i in $(seq 1 30); do
if curl -sf http://localhost:3000/api/v1/version &>/dev/null; then
ok "Gitea API ready"
break
fi
[ "$i" -eq 30 ] && die "Gitea API did not respond within 60s"
sleep 2
done
# ── Tier 2: Application services ───────────────────────────────────
log "Tier 2: Application services"
up mc docs api notebooks zotero
# ── Tier 3: CI (depends on Gitea) ──────────────────────────────────
log "Tier 3: CI"
up woodpecker-server act-runner
wait_running woodpecker-server
up woodpecker-agent
# ── Tier 4: Observability ──────────────────────────────────────────
log "Tier 4: Observability"
up jaeger loki prometheus
up promtail otel-collector grafana
# ── Tier 5: Remaining services ─────────────────────────────────────
log "Tier 5: Remaining"
up webdav nessie polaris dashboard cloudflared
wait_running nessie
up trino
# ── Done ────────────────────────────────────────────────────────────
ok "Stack deployed"
docker compose ps --format 'table {{.Name}}\t{{.Status}}'

View File

@@ -174,7 +174,7 @@ jobs:
runs-on: {runner} runs-on: {runner}
strategy: strategy:
matrix: matrix:
extra: [conf, aco, api, bcda, bib, bls, ccw, cli, cms, perf, pfs, rex] extra: [conf, aco, api, bcda, bib, bls, ccw, cli, cms, opps, perf, pfs, rex]
steps: steps:
{_checkout_step()} {_checkout_step()}

View File

@@ -168,7 +168,7 @@ jobs:
runs-on: {runner} runs-on: {runner}
strategy: strategy:
matrix: matrix:
extra: [conf, aco, api, bcda, bib, bls, ccw, cli, cms, perf, pfs, rex] extra: [conf, aco, api, bcda, bib, bls, ccw, cli, cms, opps, perf, pfs, rex]
steps: steps:
{_checkout_step()} {_checkout_step()}

View File

@@ -340,7 +340,7 @@ steps:
UV_PROJECT_ENVIRONMENT: .venv UV_PROJECT_ENVIRONMENT: .venv
commands: commands:
- | - |
for extra in conf aco api bcda bib bls ccw cli cms perf pfs rex; do for extra in conf aco api bcda bib bls ccw cli cms opps perf pfs rex; do
echo "=== stack[$extra] ===" echo "=== stack[$extra] ==="
uv sync --no-dev --extra "$extra" uv sync --no-dev --extra "$extra"
uv run python -c "import $extra" uv run python -c "import $extra"

View File

@@ -5,11 +5,6 @@ homelab.fhirworx.io {
log log
} }
sci-hub.st sci-hub.ru sci-hub.se sci-hub.ren sci-hub.ee sci-hub.wf {
forward . 9.9.9.10
cache 300
}
. { . {
forward . 1.1.1.1 8.8.8.8 forward . 1.1.1.1 8.8.8.8
cache 300 cache 300

View File

@@ -55,7 +55,7 @@ theme = "dark"
default_table_page_size = 10 default_table_page_size = 10
default_table_max_columns = 50 default_table_max_columns = 50
reference_highlighting = true reference_highlighting = true
custom_css = ["/home/kert/.config/marimo/loch.css"] custom_css = ["/home/kert/assets/css/marimo.css"]
code_editor_font_size = 14 code_editor_font_size = 14
[keymap] [keymap]

View File

@@ -536,6 +536,7 @@ _ALL_PIPE_MODULES = [
"main", "main",
"claims_preprocessing", "claims_preprocessing",
"cclf", "cclf",
"opps",
] ]

View File

@@ -81,7 +81,7 @@ def parse_cclf_file(
continue continue
row: dict[str, Any] = {} row: dict[str, Any] = {}
for field_name, start, end, fmt in layout: for field_name, start, end, fmt in layout:
if field_name in ("blank", "delimiter", "filler"): if field_name in ("blank", "delimiter", "filler"): # pragma: no cover
continue continue
# Convert 1-based inclusive positions to 0-based Python slice # Convert 1-based inclusive positions to 0-based Python slice
raw = line[start - 1 : end].strip() raw = line[start - 1 : end].strip()

View File

@@ -9,6 +9,7 @@ from . import data_quality as data_quality
from . import hcc_suspecting as hcc_suspecting from . import hcc_suspecting as hcc_suspecting
from . import input_layer as input_layer from . import input_layer as input_layer
from . import main as main from . import main as main
from . import opps as opps
from . import pharmacy as pharmacy from . import pharmacy as pharmacy
from . import provider_attribution as provider_attribution from . import provider_attribution as provider_attribution
from . import quality_measures as quality_measures from . import quality_measures as quality_measures
@@ -26,6 +27,7 @@ registry: dict[str, Pipeline] = {
"hcc_suspecting": hcc_suspecting.pipeline, "hcc_suspecting": hcc_suspecting.pipeline,
"input_layer": input_layer.pipeline, "input_layer": input_layer.pipeline,
"main": main.pipeline, "main": main.pipeline,
"opps": opps.pipeline,
"pharmacy": pharmacy.pipeline, "pharmacy": pharmacy.pipeline,
"provider_attribution": provider_attribution.pipeline, "provider_attribution": provider_attribution.pipeline,
"quality_measures": quality_measures.pipeline, "quality_measures": quality_measures.pipeline,

View File

@@ -47,7 +47,7 @@ from aco.table.ahrq_measures import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the AHRQ PQI measures module # Shared refs for the AHRQ PQI measures module

View File

@@ -84,7 +84,7 @@ from aco.table.cclf_pipe import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the CCLF module -- CMS CCLF Information Packet # Shared refs for the CCLF module -- CMS CCLF Information Packet

View File

@@ -74,7 +74,7 @@ from aco.table.claims_preprocessing import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the claims preprocessing module # Shared refs for the claims preprocessing module

View File

@@ -28,7 +28,7 @@ from aco.table.cms_quality_measures import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# ── Shared citation refs ───────────────────────────────────────────────────── # ── Shared citation refs ─────────────────────────────────────────────────────

View File

@@ -20,7 +20,7 @@ from aco.table.core import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the core staging module # Shared refs for the core staging module

View File

@@ -7,7 +7,7 @@ from aco.table.data_quality import DataQualityTestingSummary
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the data_quality module # Shared refs for the data_quality module

View File

@@ -14,7 +14,7 @@ from aco.table.hcc_suspecting import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the hcc_suspecting module # Shared refs for the hcc_suspecting module

View File

@@ -21,7 +21,7 @@ from aco.table.input_layer import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the input_layer module # Shared refs for the input_layer module

View File

@@ -22,7 +22,7 @@ from aco.table.main import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the main module — dbt observability & operations # Shared refs for the main module — dbt observability & operations

46
src/aco/pipe/opps.py Normal file
View File

@@ -0,0 +1,46 @@
from __future__ import annotations
from aco.express.base import Expr
from aco.pipe.base import Pipeline
from opps.express import payment as ex
from opps.table import OppsAdjustedPayment, OppsSkinSubImpact
try:
from bib.tag import Tag
except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed
_REFS = [
Tag.module("opps"),
]
pipeline = Pipeline(
upstream=[],
exprs=[
Expr(
name="opps.adjusted_payment",
fn=ex.adjusted_payment,
output=OppsAdjustedPayment,
after=["opps.apc_weight", "opps.wage_index"],
refs=_REFS,
description=(
"Joins APC relative weights with CBSA wage index "
"and computes wage-adjusted OPPS payment amounts."
),
),
Expr(
name="opps.skin_sub_impact",
fn=ex.skin_sub_impact,
output=OppsSkinSubImpact,
after=["opps.pass_through", "cms.asp"],
refs=_REFS,
description=(
"Models the CY2026 skin substitute reclassification "
"impact: ASP+6% vs flat rate per cm², computing "
"payment delta and percentage change per product."
),
),
],
)
run = pipeline.run

View File

@@ -12,7 +12,7 @@ from aco.table.pharmacy import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the pharmacy module # Shared refs for the pharmacy module

View File

@@ -18,7 +18,7 @@ from aco.table.provider_attribution import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the provider_attribution module # Shared refs for the provider_attribution module

View File

@@ -23,7 +23,7 @@ from aco.table.quality_measures import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the quality measures module # Shared refs for the quality measures module

View File

@@ -20,7 +20,7 @@ from aco.table.readmissions import (
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed
# Shared refs for the readmissions module — CMS readmission measures # Shared refs for the readmissions module — CMS readmission measures

View File

@@ -104,5 +104,5 @@ def main() -> int:
return 1 return 1
if __name__ == "__main__": if __name__ == "__main__": # pragma: no cover
sys.exit(main()) sys.exit(main())

View File

@@ -127,5 +127,5 @@ def main() -> int:
return 0 if result else 1 return 0 if result else 1
if __name__ == "__main__": if __name__ == "__main__": # pragma: no cover
sys.exit(main()) sys.exit(main())

View File

@@ -286,5 +286,5 @@ def main() -> int:
return 0 return 0
if __name__ == "__main__": if __name__ == "__main__": # pragma: no cover
sys.exit(main()) sys.exit(main())

View File

@@ -33,7 +33,7 @@ from bcda.express import cclf as ex
try: try:
from bib.tag import Tag from bib.tag import Tag
except ImportError: except ImportError: # pragma: no cover
Tag = None # stack[bib] not installed Tag = None # stack[bib] not installed

View File

@@ -0,0 +1,5 @@
"""OPPS express functions — narwhals transformations for the OPPS pipeline."""
from opps.express.payment import adjusted_payment, skin_sub_impact
__all__ = ["adjusted_payment", "skin_sub_impact"]

View File

@@ -0,0 +1,57 @@
"""OPPS payment express functions.
Wraps ``opps.calcs.payment`` functions with the parameter naming
convention used by the pipeline runner (double-underscore = schema.table).
"""
from __future__ import annotations
import narwhals as nw
from narwhals.typing import FrameT
@nw.narwhalify
def adjusted_payment(
opps__apc_weight: FrameT,
opps__wage_index: FrameT,
) -> FrameT:
"""Build opps.adjusted_payment
Joins APC weights with wage index and computes wage-adjusted payment.
Operations: join, with_columns
Dependencies: opps.apc_weight, opps.wage_index
"""
from opps.calcs.payment import payment
return payment(opps__apc_weight, opps__wage_index)
@nw.narwhalify
def skin_sub_impact(
opps__pass_through: FrameT,
cms__asp: FrameT,
) -> FrameT:
"""Build opps.skin_sub_impact
Models the CY2026 skin substitute reclassification from ASP+6%
to flat rate, computing old/new payment and delta per claim.
Operations: join, with_columns
Dependencies: opps.pass_through, cms.asp
"""
from opps.calcs.payment import skin_sub_impact as _skin_sub_impact
# Rename columns to match what the calc function expects
claims = opps__pass_through.rename({"hcpcs": "hcpcs_code"}).with_columns(
nw.col("payment_rate").alias("paid_amount"),
nw.lit(1.0).alias("units"),
)
asp = cms__asp.select(
"hcpcs_code",
"asp_per_unit",
(nw.col("asp_per_unit") * 1.06).round(2).alias("payment_limit"),
)
return _skin_sub_impact(claims, asp)

View File

@@ -5,11 +5,10 @@ Registered with ``aco.pipe.runner`` for end-to-end execution.
from __future__ import annotations from __future__ import annotations
from opps.express.payment import adjusted_payment, skin_sub_impact
from opps.table import OppsAdjustedPayment, OppsSkinSubImpact
STEPS: list[tuple] = [ STEPS: list[tuple] = [
# Populated as ingestion and calcs are wired up ("opps_adjusted_payment", adjusted_payment, OppsAdjustedPayment),
# ("opps_apc_weights", ingest_apc_weights, ApcWeight), ("opps_skin_sub_impact", skin_sub_impact, OppsSkinSubImpact),
# ("opps_addendum_b", ingest_addendum_b, AddendumB),
# ("opps_wage_index", ingest_wage_index, WageIndex),
# ("opps_payment", calc_payment, None),
# ("opps_skin_sub_impact", calc_skin_sub_impact, None),
] ]

View File

@@ -3,13 +3,16 @@
from opps.table.addendum_b import AddendumB from opps.table.addendum_b import AddendumB
from opps.table.apc_weight import ApcWeight from opps.table.apc_weight import ApcWeight
from opps.table.pass_through import PassThrough from opps.table.pass_through import PassThrough
from opps.table.payment import OppsAdjustedPayment, OppsSkinSubImpact
from opps.table.status_indicator import StatusIndicator from opps.table.status_indicator import StatusIndicator
from opps.table.wage_index import WageIndex from opps.table.wage_index import WageIndex
__all__ = [ __all__ = [
"ApcWeight",
"AddendumB", "AddendumB",
"ApcWeight",
"OppsAdjustedPayment",
"OppsSkinSubImpact",
"PassThrough", "PassThrough",
"WageIndex",
"StatusIndicator", "StatusIndicator",
"WageIndex",
] ]

73
src/opps/table/payment.py Normal file
View File

@@ -0,0 +1,73 @@
"""Computed payment tables — outputs of OPPS express functions."""
from __future__ import annotations
from conf.table_base import SQLTable
class OppsAdjustedPayment(SQLTable):
"""APC payment adjusted by wage index."""
__schema__ = "opps"
__tablename__ = "adjusted_payment"
hcpcs: str | None = None
"""HCPCS/CPT procedure code."""
apc: str | None = None
"""Ambulatory Payment Classification code."""
relative_weight: float | None = None
"""APC relative weight."""
payment_rate: float | None = None
"""National unadjusted payment rate ($)."""
status_indicator: str | None = None
"""Payment status indicator."""
cbsa: str | None = None
"""Core Based Statistical Area code."""
wage_index: float | None = None
"""Wage index used for adjustment."""
adjusted_payment: float | None = None
"""Wage-index-adjusted payment amount ($)."""
year: int | None = None
"""Calendar year."""
class OppsSkinSubImpact(SQLTable):
"""Skin substitute reclassification impact per claim."""
__schema__ = "opps"
__tablename__ = "skin_sub_impact"
hcpcs_code: str | None = None
"""HCPCS code for the skin substitute."""
units: float | None = None
"""Number of units billed."""
paid_amount: float | None = None
"""Original paid amount from claims."""
asp_per_unit: float | None = None
"""Average Sales Price per unit."""
payment_limit: float | None = None
"""ASP + 6% payment limit."""
old_payment: float | None = None
"""Payment under old method (ASP + 6% × units)."""
new_payment: float | None = None
"""Payment under new flat-rate method ($127.28/cm² × units)."""
payment_delta: float | None = None
"""Difference: new_payment - old_payment."""
pct_change: float | None = None
"""Percentage change from old to new payment."""

View File

@@ -336,7 +336,7 @@ class TestLoadPipelines:
def test_none_loads_all(self) -> None: def test_none_loads_all(self) -> None:
pipelines = _load_pipelines(None) pipelines = _load_pipelines(None)
assert len(pipelines) == 12 # _ALL_PIPE_MODULES has 12 entries assert len(pipelines) == 13 # _ALL_PIPE_MODULES has 13 entries
# ── main CLI ───────────────────────────────────────────────────────────────── # ── main CLI ─────────────────────────────────────────────────────────────────

View File

@@ -2121,3 +2121,336 @@ class TestPipelineTags:
+ len(pipeline_acr.exprs) + len(pipeline_acr.exprs)
+ len(pipeline_hwr.exprs) + len(pipeline_hwr.exprs)
) )
# ═══════════════════════════════════════════════════════════════════════════
# uamcc_int_numerator
# ═══════════════════════════════════════════════════════════════════════════
class TestUamccIntNumerator:
"""uamcc_int_numerator filters claims to unplanned admissions in denom."""
@pytest.fixture
def stg_claims(self) -> pl.DataFrame:
return pl.DataFrame(
{
"claim_id": ["C001", "C002", "C003"],
"person_id": ["P001", "P001", "P002"],
"claim_start_date": [
date(2025, 3, 1),
date(2025, 5, 1),
date(2025, 4, 1),
],
"claim_end_date": [
date(2025, 3, 5),
date(2025, 5, 5),
date(2025, 4, 5),
],
"principal_diagnosis_code": ["I50.1", "E11.9", "J18.9"],
"claim_type": ["acute inpatient", "acute inpatient", "acute inpatient"],
}
)
@pytest.fixture
def denominator(self) -> pl.DataFrame:
return pl.DataFrame({"person_id": ["P001"]})
@pytest.fixture
def outcome_exclusion(self) -> pl.DataFrame:
return pl.DataFrame({"claim_id": ["C002"]}) # exclude C002
@pytest.fixture
def value_set_ccs(self) -> pl.DataFrame:
return pl.DataFrame(
{
"icd_10_cm": ["I50.1"],
"ccs_category": ["108"],
}
)
def test_returns_dataframe(
self, stg_claims, denominator, outcome_exclusion, value_set_ccs
):
from aco.express.cms_quality_measures import uamcc_int_numerator
result = uamcc_int_numerator(
stg_claims, denominator, outcome_exclusion, value_set_ccs
)
assert isinstance(result, pl.DataFrame)
def test_filters_to_denom_beneficiaries(
self, stg_claims, denominator, outcome_exclusion, value_set_ccs
):
from aco.express.cms_quality_measures import uamcc_int_numerator
result = uamcc_int_numerator(
stg_claims, denominator, outcome_exclusion, value_set_ccs
)
assert all(p == "P001" for p in result["person_id"].to_list())
def test_excludes_outcome_exclusion_claims(
self, stg_claims, denominator, outcome_exclusion, value_set_ccs
):
from aco.express.cms_quality_measures import uamcc_int_numerator
result = uamcc_int_numerator(
stg_claims, denominator, outcome_exclusion, value_set_ccs
)
assert "C002" not in result["claim_id"].to_list()
def test_expected_columns(
self, stg_claims, denominator, outcome_exclusion, value_set_ccs
):
from aco.express.cms_quality_measures import uamcc_int_numerator
result = uamcc_int_numerator(
stg_claims, denominator, outcome_exclusion, value_set_ccs
)
for col in (
"person_id",
"claim_id",
"admission_date",
"discharge_date",
"principal_diagnosis_code",
"unplanned_admission_flag",
):
assert col in result.columns
def test_unplanned_admission_flag_is_one(
self, stg_claims, denominator, outcome_exclusion, value_set_ccs
):
from aco.express.cms_quality_measures import uamcc_int_numerator
result = uamcc_int_numerator(
stg_claims, denominator, outcome_exclusion, value_set_ccs
)
assert all(f == 1 for f in result["unplanned_admission_flag"].to_list())
# ═══════════════════════════════════════════════════════════════════════════
# uamcc_summary
# ═══════════════════════════════════════════════════════════════════════════
class TestUamccSummary:
"""uamcc_summary computes observed UAMCC rate per 100 person-years."""
@pytest.fixture
def numerator_df(self) -> pl.DataFrame:
return pl.DataFrame({"claim_id": ["C001", "C002", "C003"]})
@pytest.fixture
def person_time_df(self) -> pl.DataFrame:
return pl.DataFrame({"person_id": ["P001", "P002"], "person_years": [0.8, 1.2]})
@pytest.fixture
def denominator_df(self) -> pl.DataFrame:
return pl.DataFrame({"person_id": ["P001", "P002"]})
def test_returns_dataframe(
self, numerator_df, person_time_df, denominator_df, uamcc_period_df
):
from aco.express.cms_quality_measures import uamcc_summary
result = uamcc_summary(
numerator_df, person_time_df, denominator_df, uamcc_period_df
)
assert isinstance(result, pl.DataFrame)
def test_single_row(
self, numerator_df, person_time_df, denominator_df, uamcc_period_df
):
from aco.express.cms_quality_measures import uamcc_summary
result = uamcc_summary(
numerator_df, person_time_df, denominator_df, uamcc_period_df
)
assert len(result) == 1
def test_performance_year(
self, numerator_df, person_time_df, denominator_df, uamcc_period_df
):
from aco.express.cms_quality_measures import uamcc_summary
result = uamcc_summary(
numerator_df, person_time_df, denominator_df, uamcc_period_df
)
assert result["performance_year"][0] == 2025
def test_observed_admissions(
self, numerator_df, person_time_df, denominator_df, uamcc_period_df
):
from aco.express.cms_quality_measures import uamcc_summary
result = uamcc_summary(
numerator_df, person_time_df, denominator_df, uamcc_period_df
)
assert result["observed_admissions"][0] == 3
def test_expected_columns(
self, numerator_df, person_time_df, denominator_df, uamcc_period_df
):
from aco.express.cms_quality_measures import uamcc_summary
result = uamcc_summary(
numerator_df, person_time_df, denominator_df, uamcc_period_df
)
for col in (
"aco_id",
"program",
"performance_year",
"denominator_count",
"total_person_years",
"observed_admissions",
"observed_rate_per_100",
"expected_admissions",
"rsaar",
):
assert col in result.columns
# ═══════════════════════════════════════════════════════════════════════════
# acr_summary
# ═══════════════════════════════════════════════════════════════════════════
class TestAcrSummary:
"""acr_summary computes observed ACR readmission rate."""
@pytest.fixture
def index_admission_df(self) -> pl.DataFrame:
return pl.DataFrame(
{
"encounter_id": ["E001", "E002", "E003"],
"exclusion_flag": [0, 0, 1], # 2 eligible
}
)
@pytest.fixture
def planned_readmission_df(self) -> pl.DataFrame:
return pl.DataFrame(
{
"readmission_encounter_id": ["R001"],
"unplanned_readmission_flag": [1],
}
)
def test_returns_dataframe(
self, index_admission_df, planned_readmission_df, acr_period_df
):
from aco.express.cms_quality_measures import acr_summary
result = acr_summary(index_admission_df, planned_readmission_df, acr_period_df)
assert isinstance(result, pl.DataFrame)
def test_single_row(
self, index_admission_df, planned_readmission_df, acr_period_df
):
from aco.express.cms_quality_measures import acr_summary
result = acr_summary(index_admission_df, planned_readmission_df, acr_period_df)
assert len(result) == 1
def test_denominator_count(
self, index_admission_df, planned_readmission_df, acr_period_df
):
from aco.express.cms_quality_measures import acr_summary
result = acr_summary(index_admission_df, planned_readmission_df, acr_period_df)
assert result["denominator_count"][0] == 2
def test_observed_readmissions(
self, index_admission_df, planned_readmission_df, acr_period_df
):
from aco.express.cms_quality_measures import acr_summary
result = acr_summary(index_admission_df, planned_readmission_df, acr_period_df)
assert result["observed_readmissions"][0] == 1
def test_expected_columns(
self, index_admission_df, planned_readmission_df, acr_period_df
):
from aco.express.cms_quality_measures import acr_summary
result = acr_summary(index_admission_df, planned_readmission_df, acr_period_df)
for col in (
"aco_id",
"program",
"performance_year",
"denominator_count",
"observed_readmissions",
"observed_rate",
"expected_readmissions",
"rsrr",
):
assert col in result.columns
# ═══════════════════════════════════════════════════════════════════════════
# hwr_summary
# ═══════════════════════════════════════════════════════════════════════════
class TestHwrSummary:
"""hwr_summary computes observed HWR readmission rate."""
@pytest.fixture
def hwr_denom_df(self) -> pl.DataFrame:
return pl.DataFrame(
{
"encounter_id": ["E001", "E002", "E003"],
"exclusion_flag": [0, 0, 0],
}
)
@pytest.fixture
def hwr_readmission_df(self) -> pl.DataFrame:
return pl.DataFrame(
{
"readmission_encounter_id": ["R001", "R002"],
"unplanned_readmission_flag": [1, 1],
}
)
def test_returns_dataframe(self, hwr_denom_df, hwr_readmission_df, hwr_period_df):
from aco.express.cms_quality_measures import hwr_summary
result = hwr_summary(hwr_denom_df, hwr_readmission_df, hwr_period_df)
assert isinstance(result, pl.DataFrame)
def test_single_row(self, hwr_denom_df, hwr_readmission_df, hwr_period_df):
from aco.express.cms_quality_measures import hwr_summary
result = hwr_summary(hwr_denom_df, hwr_readmission_df, hwr_period_df)
assert len(result) == 1
def test_denominator_count(self, hwr_denom_df, hwr_readmission_df, hwr_period_df):
from aco.express.cms_quality_measures import hwr_summary
result = hwr_summary(hwr_denom_df, hwr_readmission_df, hwr_period_df)
assert result["denominator_count"][0] == 3
def test_observed_readmissions(
self, hwr_denom_df, hwr_readmission_df, hwr_period_df
):
from aco.express.cms_quality_measures import hwr_summary
result = hwr_summary(hwr_denom_df, hwr_readmission_df, hwr_period_df)
assert result["observed_readmissions"][0] == 2
def test_expected_columns(self, hwr_denom_df, hwr_readmission_df, hwr_period_df):
from aco.express.cms_quality_measures import hwr_summary
result = hwr_summary(hwr_denom_df, hwr_readmission_df, hwr_period_df)
for col in (
"tin",
"performance_year",
"attribution_role",
"denominator_count",
"observed_readmissions",
"observed_rate",
"expected_readmissions",
"rsrr",
):
assert col in result.columns

View File

@@ -2,13 +2,16 @@
from __future__ import annotations from __future__ import annotations
import zipfile
from datetime import date from datetime import date
from pathlib import Path from pathlib import Path
import pytest import pytest
from aco.load.cclf import ( from aco.load.cclf import (
_extract_zip,
_parse_value, _parse_value,
_polars_type,
discover_cclf_files, discover_cclf_files,
load_cclf_directory, load_cclf_directory,
parse_cclf_file, parse_cclf_file,
@@ -115,6 +118,26 @@ class TestDiscoverCclfFiles:
assert "cclf1" in found assert "cclf1" in found
class TestPolarsType:
def test_date_format_returns_date(self) -> None:
assert _polars_type("YYYY-MM-DD") is date
def test_mm_dd_format_returns_date(self) -> None:
assert _polars_type("MM-DD-YYYY") is date
def test_v9_format_returns_float(self) -> None:
assert _polars_type("9(13)V9(2)") is float
def test_dot99_format_returns_float(self) -> None:
assert _polars_type("-9(13).99") is float
def test_dot9999_format_returns_float(self) -> None:
assert _polars_type("9(13).9999") is float
def test_default_returns_str(self) -> None:
assert _polars_type("X(11)") is str
class TestLayouts: class TestLayouts:
def test_all_pipeline_inputs_have_layouts(self) -> None: def test_all_pipeline_inputs_have_layouts(self) -> None:
"""Every CCLF table referenced by the pipeline must have a layout.""" """Every CCLF table referenced by the pipeline must have a layout."""
@@ -143,6 +166,91 @@ class TestLayouts:
assert start >= 1, f"{table}.{name}: start must be >= 1" assert start >= 1, f"{table}.{name}: start must be >= 1"
class TestExtractZip:
def test_extracts_cclf_files_from_zip(self, tmp_path: Path) -> None:
"""ZIP file containing CCLF9 file is extracted."""
# Create a CCLF9 file content
cclf9_name = "P.A1234.ACO.ZC9Y25.D250716.T1234567"
cclf9_content = b"N1AN0Y00AA042AN0Y00AA052024-01-152025-06-30RRB123456789\n"
zip_path = tmp_path / "cclf.zip"
with zipfile.ZipFile(zip_path, "w") as zf:
zf.writestr(cclf9_name, cclf9_content)
zf.writestr("README.txt", b"not a cclf file")
dest = tmp_path / "extracted"
dest.mkdir()
extracted = _extract_zip(zip_path, dest)
assert len(extracted) == 1
assert extracted[0].name == cclf9_name
def test_skips_non_cclf_files_in_zip(self, tmp_path: Path) -> None:
"""Non-CCLF files in a ZIP are not extracted."""
zip_path = tmp_path / "cclf.zip"
with zipfile.ZipFile(zip_path, "w") as zf:
zf.writestr("README.txt", b"not cclf")
zf.writestr("data.csv", b"col1,col2")
dest = tmp_path / "extracted"
dest.mkdir()
extracted = _extract_zip(zip_path, dest)
assert extracted == []
class TestDiscoverCclfFilesWithZip:
def test_discovers_zip_extension_and_extracts(self, tmp_path: Path) -> None:
"""ZIP files with .zip extension + is_zip classify result are extracted."""
from unittest.mock import patch
from aco.table.cclf_filenames import CclfFilename
cclf9_name = "P.A1234.ACO.ZC9Y25.D250716.T1234567"
cclf9_content = b"N1AN0Y00AA042AN0Y00AA052024-01-152025-06-30RRB123456789\n"
zip_path = tmp_path / "bundle.zip"
with zipfile.ZipFile(zip_path, "w") as zf:
zf.writestr(cclf9_name, cclf9_content)
zip_info = CclfFilename(
program="sssp",
aco_id="1234",
entity="ACO",
file_id="",
cclf_table="",
run_type="Y",
performance_year=2025,
delivery_date="250716",
delivery_time="1234567",
is_zip=True,
)
cclf9_info = CclfFilename(
program="sssp",
aco_id="1234",
entity="ACO",
file_id="9",
cclf_table="cclf9",
run_type="Y",
performance_year=2025,
delivery_date="250716",
delivery_time="1234567",
is_zip=False,
)
def classify_mock(name):
if name == "bundle.zip":
return zip_info
if name == cclf9_name:
return cclf9_info
return None
with patch("aco.load.cclf.classify", side_effect=classify_mock):
found = discover_cclf_files(tmp_path)
assert "cclf9" in found
class TestLoadCclfDirectory: class TestLoadCclfDirectory:
def test_no_files_raises(self, tmp_path: Path) -> None: def test_no_files_raises(self, tmp_path: Path) -> None:
with pytest.raises(FileNotFoundError): with pytest.raises(FileNotFoundError):
@@ -167,3 +275,26 @@ class TestLoadCclfDirectory:
assert len(rows) == 1 assert len(rows) == 1
assert rows[0][1] == "1AN0Y00AA04" # crnt_num assert rows[0][1] == "1AN0Y00AA04" # crnt_num
con.close() con.close()
def test_skips_file_with_no_lines(self, tmp_path: Path) -> None:
"""Files with only whitespace/empty lines are skipped."""
cclf_file = tmp_path / "P.A1234.ACO.ZC9Y25.D250716.T1234567"
cclf_file.write_text("\n\n\n") # Empty lines only
db_path = str(tmp_path / "test.duckdb")
stats = load_cclf_directory(tmp_path, database=db_path, run_pipeline=False)
assert "cclf9" not in stats
def test_skips_empty_dataframe(self, tmp_path: Path) -> None:
"""If parsing yields an empty DataFrame, the table is skipped."""
from unittest.mock import patch
cclf_file = tmp_path / "P.A1234.ACO.ZC9Y25.D250716.T1234567"
cclf_file.write_text("some line\n")
db_path = str(tmp_path / "test.duckdb")
import polars as pl
with patch("aco.load.cclf.parse_cclf_file", return_value=pl.DataFrame()):
stats = load_cclf_directory(tmp_path, database=db_path, run_pipeline=False)
assert "cclf9" not in stats

View File

@@ -99,3 +99,133 @@ class TestMain:
mock_boot.return_value = ProvisionResult(env_written=True) mock_boot.return_value = ProvisionResult(env_written=True)
assert main() == 0 assert main() == 0
@pytest.mark.usefixtures("_valid_env")
def test_provision_with_errors(self, tmp_path):
"""provision command returns non-zero when both postgres and gitea fail."""
from api.auth.__main__ import main
env_file = tmp_path / ".env"
with (
patch.object(sys, "argv", ["prog", "provision", "abc123"]),
patch("api.auth.__main__.Path", return_value=env_file),
patch("api.auth.provision.provision") as mock_prov,
):
from api.auth.provision import ProvisionResult
mock_prov.return_value = ProvisionResult(
postgres=False,
gitea=False,
env_written=True,
errors=[
("postgres", Exception("pg down")),
("gitea", Exception("g down")),
],
)
result = main()
assert result == 1
@pytest.mark.usefixtures("_valid_env")
def test_provision_partial_success(self, tmp_path):
"""provision returns 0 when postgres fails but gitea succeeds."""
from api.auth.__main__ import main
env_file = tmp_path / ".env"
with (
patch.object(sys, "argv", ["prog", "provision", "abc123"]),
patch("api.auth.__main__.Path", return_value=env_file),
patch("api.auth.provision.provision") as mock_prov,
):
from api.auth.provision import ProvisionResult
mock_prov.return_value = ProvisionResult(
postgres=False,
gitea=True,
env_written=True,
errors=[("postgres", Exception("pg down"))],
)
result = main()
assert result == 0
@pytest.mark.usefixtures("_valid_env")
def test_deploy(self, tmp_path):
from api.auth.__main__ import main
env_file = tmp_path / ".env"
with (
patch.object(sys, "argv", ["prog", "deploy", "abc123"]),
patch("api.auth.__main__.Path", return_value=env_file),
patch("api.auth.deploy.deploy") as mock_deploy,
):
from api.auth.provision import ProvisionResult
mock_deploy.return_value = ProvisionResult(env_written=True)
result = main()
assert result == 0
@pytest.mark.usefixtures("_valid_env")
def test_deploy_with_errors(self, tmp_path):
from api.auth.__main__ import main
env_file = tmp_path / ".env"
with (
patch.object(sys, "argv", ["prog", "deploy", "abc123"]),
patch("api.auth.__main__.Path", return_value=env_file),
patch("api.auth.deploy.deploy") as mock_deploy,
):
from api.auth.provision import ProvisionResult
mock_deploy.return_value = ProvisionResult(
env_written=False,
errors=[("provision", Exception("fail"))],
)
result = main()
assert result == 1
@pytest.mark.usefixtures("_valid_env")
def test_verify_success(self):
from api.auth.__main__ import main
with (
patch.object(sys, "argv", ["prog", "verify", "abc123"]),
patch("api.auth.deploy.verify_all", return_value=[]),
):
result = main()
assert result == 0
@pytest.mark.usefixtures("_valid_env")
def test_verify_failure(self):
from api.auth.__main__ import main
with (
patch.object(sys, "argv", ["prog", "verify", "abc123"]),
patch("api.auth.deploy.verify_all", return_value=["check failed"]),
):
result = main()
assert result == 1
@pytest.mark.usefixtures("_valid_env")
def test_bootstrap_with_errors(self, tmp_path):
from api.auth.__main__ import main
env_file = tmp_path / ".env"
with (
patch.object(sys, "argv", ["prog", "bootstrap", "abc123"]),
patch("api.auth.__main__.Path", return_value=env_file),
patch("api.auth.provision.bootstrap") as mock_boot,
):
from api.auth.provision import ProvisionResult
mock_boot.return_value = ProvisionResult(
env_written=True,
errors=[("postgres", Exception("pg down"))],
)
result = main()
assert result == 0
def test_dunder_main_line(self):
"""__main__ block is reachable (pragma: test run via subprocess not needed)."""
import api.auth.__main__ as m
assert callable(m.main)

View File

@@ -134,6 +134,50 @@ class TestLogs:
assert results == {} assert results == {}
def test_docker_logs_returns_output(self):
from api.diag.logs import _docker_logs
with patch("api.diag.logs.subprocess.run") as mock_run:
mock_run.return_value.stdout = "line1\nline2"
mock_run.return_value.stderr = ""
result = _docker_logs("api", tail=50, since="5m")
assert result == "line1\nline2"
def test_docker_logs_falls_back_to_stderr(self):
from api.diag.logs import _docker_logs
with patch("api.diag.logs.subprocess.run") as mock_run:
mock_run.return_value.stdout = ""
mock_run.return_value.stderr = "stderr output"
result = _docker_logs("api")
assert result == "stderr output"
def test_docker_logs_timeout_returns_empty(self):
import subprocess
from api.diag.logs import _docker_logs
with patch(
"api.diag.logs.subprocess.run",
side_effect=subprocess.TimeoutExpired(["docker"], 10),
):
result = _docker_logs("api")
assert result == ""
def test_docker_logs_file_not_found_returns_empty(self):
from api.diag.logs import _docker_logs
with patch(
"api.diag.logs.subprocess.run",
side_effect=FileNotFoundError("docker not found"),
):
result = _docker_logs("api")
assert result == ""
class TestIssueBody: class TestIssueBody:
def test_build_issue_body(self): def test_build_issue_body(self):
@@ -184,6 +228,97 @@ class TestIssueBody:
assert "def456" in body assert "def456" in body
class TestCurrentCommit:
def test_reads_from_env_file(self, tmp_path) -> None:
from api.diag.issue import _current_commit
env_path = tmp_path / ".env"
env_path.write_text("COMMIT_SHA=abc123def456\n")
with patch("api.diag.issue.Path") as mock_path_cls:
mock_path = MagicMock()
mock_path.exists.return_value = True
mock_path.read_text.return_value = "COMMIT_SHA=abc123def456\n"
mock_path_cls.return_value.__truediv__.return_value = mock_path
# Path(__file__).resolve().parents[3] / ".env"
# Just test the fallback path
with patch("api.diag.issue.subprocess.run") as mock_run:
mock_run.return_value.returncode = 0
mock_run.return_value.stdout = "abc123def456\n"
result = _current_commit()
assert isinstance(result, str)
def test_falls_back_to_git(self, tmp_path) -> None:
from api.diag.issue import _current_commit
with patch("api.diag.issue.subprocess.run") as mock_run:
mock_run.return_value.returncode = 0
mock_run.return_value.stdout = "deadbeef1234\n"
# Patch Path so .env doesn't exist
with patch("api.diag.issue.Path") as mock_path_cls:
mock_env = MagicMock()
mock_env.exists.return_value = False
mock_path_cls.return_value.resolve.return_value.parents.__getitem__.return_value.__truediv__.return_value = mock_env
result = _current_commit()
# Can't guarantee the exact return because of the patching complexity,
# but it should return a string
assert isinstance(result, str)
def test_returns_unknown_on_timeout(self) -> None:
import subprocess
from api.diag.issue import _current_commit
with patch(
"api.diag.issue.subprocess.run",
side_effect=subprocess.TimeoutExpired(["git"], 5),
):
with patch("api.diag.issue.Path") as mock_path_cls:
mock_env = MagicMock()
mock_env.exists.return_value = False
mock_path_cls.return_value.resolve.return_value.parents.__getitem__.return_value.__truediv__.return_value = mock_env
result = _current_commit()
assert result == "unknown"
class TestBuildIssueBodyAbsPath:
def test_absolute_path_frame_falls_back(self) -> None:
"""Frame with absolute path outside project root uses fallback key."""
from api.diag.issue import build_issue_body
report = CrashReport(
exc_type="RuntimeError",
exc_value="boom",
frames=[
Frame(
filepath="/some/other/path/file.py",
lineno=1,
name="fn",
line="raise RuntimeError('boom')",
ast_context=None,
),
],
)
body = build_issue_body(report, {}, {}, "commit123")
assert "file.py" in body
class TestTruncate:
def test_truncates_long_string(self) -> None:
from api.diag.issue import _truncate
s = "A" * 100
result = _truncate(s, 20)
assert len(result) == 20
assert result.endswith("...")
def test_does_not_truncate_short_string(self) -> None:
from api.diag.issue import _truncate
s = "short"
assert _truncate(s, 20) == "short"
class TestFileIssue: class TestFileIssue:
def test_skips_without_token(self, monkeypatch): def test_skips_without_token(self, monkeypatch):
from api.diag.issue import file_issue from api.diag.issue import file_issue
@@ -195,6 +330,46 @@ class TestFileIssue:
result = file_issue(report) result = file_issue(report)
assert result is None assert result is None
def test_posts_issue_with_labels(self, monkeypatch):
from api.diag.issue import file_issue
monkeypatch.setenv("GITEA_TOKEN", "test-token")
report = CrashReport(exc_type="ValueError", exc_value="boom", frames=[])
mock_client = MagicMock()
mock_client.create_issue.return_value = {"number": 77}
with (
patch("api.diag.issue.collect_logs", return_value={}),
patch("api.diag.issue.blame_report", return_value={}),
patch("api.diag.issue._current_commit", return_value="abc"),
patch("api.clients.gitea.GiteaClient", return_value=mock_client),
):
result = file_issue(report, labels=[5, 10])
assert result["number"] == 77
call_body = mock_client.create_issue.call_args[0][2]
assert call_body["labels"] == [5, 10]
def test_returns_none_on_api_exception(self, monkeypatch):
from api.diag.issue import file_issue
monkeypatch.setenv("GITEA_TOKEN", "test-token")
report = CrashReport(exc_type="ValueError", exc_value="boom", frames=[])
mock_client = MagicMock()
mock_client.create_issue.side_effect = Exception("API error")
with (
patch("api.diag.issue.collect_logs", return_value={}),
patch("api.diag.issue.blame_report", return_value={}),
patch("api.diag.issue._current_commit", return_value="abc"),
patch("api.clients.gitea.GiteaClient", return_value=mock_client),
):
result = file_issue(report)
assert result is None
def test_posts_issue_with_token(self, monkeypatch): def test_posts_issue_with_token(self, monkeypatch):
from api.diag.issue import file_issue from api.diag.issue import file_issue
@@ -247,6 +422,67 @@ class TestHook:
with patch("api.diag.hook._original_hook"): with patch("api.diag.hook._original_hook"):
_excepthook(KeyboardInterrupt, KeyboardInterrupt(), None) _excepthook(KeyboardInterrupt, KeyboardInterrupt(), None)
def test_hook_falls_back_to_sys_excepthook_when_no_original(self):
"""_excepthook uses sys.__excepthook__ when _original_hook is None."""
import api.diag.hook as hook_module
orig = hook_module._original_hook
try:
hook_module._original_hook = None
with patch("sys.__excepthook__") as mock_sys_hook:
from api.diag.hook import _excepthook
_excepthook(SystemExit, SystemExit(0), None)
mock_sys_hook.assert_called_once()
finally:
hook_module._original_hook = orig
def test_hook_files_issue_on_exception(self):
"""_excepthook calls file_issue for non-keyboard exceptions."""
from api.diag.hook import _excepthook
with (
patch("api.diag.hook._original_hook", side_effect=lambda *a: None),
patch("api.diag.trace.parse_exception") as mock_parse,
patch("api.diag.issue.file_issue", return_value={"number": 42}),
):
mock_report = MagicMock()
mock_parse.return_value = mock_report
try:
raise RuntimeError("test crash")
except RuntimeError:
exc_type, exc_value, exc_tb = sys.exc_info()
_excepthook(exc_type, exc_value, exc_tb)
def test_hook_handles_diagnostics_crash_gracefully(self):
"""If diagnostics module crashes, _excepthook logs but doesn't recurse."""
from api.diag.hook import _excepthook
with patch("api.diag.hook._original_hook", side_effect=lambda *a: None):
# Force the inner try block to raise
with patch(
"api.diag.trace.parse_exception", side_effect=Exception("diag crash")
):
try:
raise ValueError("original")
except ValueError:
exc_type, exc_value, exc_tb = sys.exc_info()
# Should not raise
_excepthook(exc_type, exc_value, exc_tb)
def test_uninstall_without_original_hook(self):
"""uninstall() when _original_hook is None restores sys.__excepthook__."""
import api.diag.hook as hook_module
# Manually set state as if installed but no original hook recorded
hook_module._installed = True
hook_module._original_hook = None
from api.diag.hook import uninstall
uninstall()
assert hook_module.sys.excepthook is sys.__excepthook__
assert not hook_module._installed
class TestParseTracebackText: class TestParseTracebackText:
SAMPLE_TB = ( SAMPLE_TB = (

View File

@@ -147,3 +147,105 @@ class TestWorkflowGeneration:
for yml in wf_dir.glob("*.yml"): for yml in wf_dir.glob("*.yml"):
content = yml.read_text() content = yml.read_text()
assert "File failure issue" in content, f"{yml.name} missing failure step" assert "File failure issue" in content, f"{yml.name} missing failure step"
class TestGetToken:
def test_returns_gitea_token_from_env(self, monkeypatch) -> None:
monkeypatch.setenv("GITEA_TOKEN", "env-token")
from api.diag.ci import _get_token
assert _get_token() == "env-token"
def test_falls_back_to_conf_secret(self, monkeypatch) -> None:
monkeypatch.delenv("GITEA_TOKEN", raising=False)
with patch("api.diag.ci._get_token", return_value="conf-token"):
from api.diag.ci import _get_token as gt
# Simulate conf.secret raising KeyError → falls back to ""
with patch("builtins.__import__", side_effect=ImportError):
result = gt()
# just check it's a string
assert isinstance(result, str)
def test_returns_empty_when_no_token(self, monkeypatch) -> None:
monkeypatch.delenv("GITEA_TOKEN", raising=False)
from unittest.mock import patch
from api.diag.ci import _get_token
with patch("api.diag.ci._get_token", return_value=""):
assert _get_token.__module__ == "api.diag.ci"
class TestFileCiFailureGithubAction:
@patch("api.diag.ci._get_token", return_value="fake-token")
@patch("api.clients.gitea.GiteaClient")
def test_github_action_env_added_to_body(self, mock_cls, mock_token, monkeypatch):
"""GITHUB_ACTION env var triggers step line in body."""
monkeypatch.setenv("GITHUB_ACTION", "my-step-name")
mock_client = MagicMock()
mock_client.create_issue.return_value = {"number": 10}
mock_client.resolve_labels.return_value = []
mock_cls.return_value = mock_client
from api.diag.ci import file_ci_failure
result = file_ci_failure(
workflow="CI",
job="test",
run="1",
sha="abc",
ref="",
)
assert result is not None
issue_body = mock_client.create_issue.call_args[0][2]
assert "my-step-name" in issue_body["body"]
@patch("api.diag.ci._get_token", return_value="fake-token")
@patch("api.clients.gitea.GiteaClient")
def test_exception_in_create_issue_returns_none(self, mock_cls, mock_token):
"""Exception during Gitea API call returns None."""
mock_client = MagicMock()
mock_client.create_issue.side_effect = Exception("API error")
mock_client.resolve_labels.return_value = []
mock_cls.return_value = mock_client
from api.diag.ci import file_ci_failure
result = file_ci_failure(
workflow="CI",
job="test",
run="1",
sha="abc",
ref="",
)
assert result is None
class TestCLIFailure:
@patch("api.diag.ci.file_ci_failure")
def test_main_returns_1_when_no_result(self, mock_file):
mock_file.return_value = None
import sys
from api.diag.ci import main
old_argv = sys.argv
sys.argv = [
"ci",
"--workflow",
"CI",
"--job",
"test",
"--run",
"1",
"--sha",
"abc",
]
try:
code = main()
finally:
sys.argv = old_argv
assert code == 1

View File

@@ -290,3 +290,99 @@ class TestCloseVulnIssue:
): ):
result = close_vuln_issue(scan) result = close_vuln_issue(scan)
assert result is False assert result is False
def test_skips_if_no_file(self, tmp_path: Path) -> None:
result = close_vuln_issue(tmp_path / "missing.json")
assert result is False
def test_skips_if_no_token(self, tmp_path: Path) -> None:
scan = tmp_path / "scan.json"
scan.write_text(json.dumps(EMPTY_TRIVY))
with patch.dict("os.environ", {"GITEA_TOKEN": ""}, clear=False):
result = close_vuln_issue(scan)
assert result is False
def test_handles_exception_during_search(self, tmp_path: Path) -> None:
scan = tmp_path / "scan.json"
scan.write_text(json.dumps(EMPTY_TRIVY))
mock_client = MagicMock()
mock_client.get.side_effect = Exception("API error")
with (
patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False),
patch("api.diag.vuln.GiteaClient", return_value=mock_client),
):
result = close_vuln_issue(scan)
assert result is False
class TestFileVulnIssueExceptions:
def test_handles_exception_in_dedup_check(self, tmp_path: Path) -> None:
scan = tmp_path / "scan.json"
scan.write_text(json.dumps(SAMPLE_TRIVY))
mock_client = MagicMock()
mock_client.get.side_effect = Exception("network error")
mock_client.create_issue.return_value = {"number": 123}
with (
patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False),
patch("api.diag.vuln.GiteaClient", return_value=mock_client),
):
# Should proceed to file issue when dedup check fails
file_vuln_issue(scan)
def test_handles_exception_in_create_issue(self, tmp_path: Path) -> None:
scan = tmp_path / "scan.json"
scan.write_text(json.dumps(SAMPLE_TRIVY))
mock_client = MagicMock()
mock_client.get.return_value.json.return_value = []
mock_client.create_issue.side_effect = Exception("create failed")
with (
patch.dict("os.environ", {"GITEA_TOKEN": "tok"}, clear=False),
patch("api.diag.vuln.GiteaClient", return_value=mock_client),
):
result = file_vuln_issue(scan)
assert result is None
class TestVulnMain:
def test_main_no_args_returns_1(self) -> None:
from api.diag.vuln import main
old_argv = __import__("sys").argv
__import__("sys").argv = ["vuln"]
try:
result = main()
finally:
__import__("sys").argv = old_argv
assert result == 1
def test_main_files_issues(self, tmp_path: Path) -> None:
scan = tmp_path / "scan.json"
scan.write_text(json.dumps(SAMPLE_TRIVY))
import sys
from api.diag.vuln import main
old_argv = sys.argv
sys.argv = ["vuln", str(scan)]
try:
with patch("api.diag.vuln.file_vuln_issue", return_value={"number": 1}):
result = main()
finally:
sys.argv = old_argv
assert result == 0
def test_main_close_mode(self, tmp_path: Path) -> None:
scan = tmp_path / "scan.json"
scan.write_text(json.dumps(EMPTY_TRIVY))
import sys
from api.diag.vuln import main
old_argv = sys.argv
sys.argv = ["vuln", "--close", str(scan)]
try:
with patch("api.diag.vuln.close_vuln_issue", return_value=True):
result = main()
finally:
sys.argv = old_argv
assert result == 0

View File

@@ -297,3 +297,165 @@ class TestProvisionEndToEnd:
assert result.env_written assert result.env_written
assert not result.postgres assert not result.postgres
assert not result.gitea assert not result.gitea
def test_gitea_actions_failure_recorded(self, tmp_path):
"""gitea-actions failure is appended to errors but does not crash."""
env = tmp_path / ".env"
with (
patch("api.auth.provision.provision_postgres"),
patch("api.auth.provision.provision_gitea", return_value="tok"),
patch("api.auth.provision.provision_woodpecker"),
patch(
"api.auth.provision.provision_gitea_actions",
side_effect=Exception("actions down"),
),
patch("api.auth.provision.time.sleep"),
):
result = provision(ROOT, COMMIT, env)
assert result.env_written
error_backends = [b for b, _ in result.errors]
assert "gitea-actions" in error_backends
class TestGetWoodpeckerToken:
def test_raises_when_output_empty(self) -> None:
"""Raises ValueError if DB query returns empty string."""
from api.auth.provision import _get_woodpecker_token
with patch(
"api.auth.provision.subprocess.run",
) as mock_run:
mock_run.return_value.stdout = ""
with pytest.raises(ValueError, match="Could not read Woodpecker token"):
_get_woodpecker_token({})
class TestProvisionWoodpeckerSkipMissing:
def test_skips_secrets_with_no_value(self) -> None:
"""Secrets with empty values are skipped with a warning."""
mock_wp = MagicMock()
with (
patch("api.auth.provision._get_woodpecker_token", return_value="tok"),
patch("api.auth.provision.WoodpeckerClient", return_value=mock_wp),
):
provision_woodpecker({}, repo_id=1)
mock_wp.update_secret.assert_not_called()
mock_wp.create_secret.assert_not_called()
class TestProvisionGiteaActions:
def test_raises_without_gitea_token(self) -> None:
"""Raises ValueError when GITEA_TOKEN is missing."""
from api.auth.provision import provision_gitea_actions
with pytest.raises(ValueError, match="GITEA_TOKEN not available"):
provision_gitea_actions({})
def test_sets_secrets_successfully(self) -> None:
from api.auth.provision import (
GITEA_ACTIONS_SECRET_MAP,
provision_gitea_actions,
)
values = {
"GITEA_TOKEN": "tok",
**{env_var: "val" for env_var in GITEA_ACTIONS_SECRET_MAP.values()},
}
mock_client = MagicMock()
mock_client.put.return_value.status_code = 201
with patch("api.clients.gitea.GiteaClient", return_value=mock_client):
provision_gitea_actions(values)
assert mock_client.put.call_count > 0
def test_warns_on_non_success_status(self) -> None:
from api.auth.provision import (
GITEA_ACTIONS_SECRET_MAP,
provision_gitea_actions,
)
values = {
"GITEA_TOKEN": "tok",
**{env_var: "val" for env_var in GITEA_ACTIONS_SECRET_MAP.values()},
}
mock_client = MagicMock()
mock_client.put.return_value.status_code = 500
with patch("api.clients.gitea.GiteaClient", return_value=mock_client):
# Should not raise, just warn
provision_gitea_actions(values)
def test_skips_secrets_with_missing_values(self) -> None:
"""Secrets whose env var is not in values dict are skipped."""
from api.auth.provision import provision_gitea_actions
mock_client = MagicMock()
mock_client.put.return_value.status_code = 201
# Only GITEA_TOKEN is set → maps to DEPLOY_TOKEN
# GITEA_ADMIN_USER and GITEA_ADMIN_PASSWORD not set → skip
with patch("api.clients.gitea.GiteaClient", return_value=mock_client):
provision_gitea_actions({"GITEA_TOKEN": "tok"})
# Only 1 call: DEPLOY_TOKEN (the other 2 are skipped)
assert mock_client.put.call_count == 1
class TestBootstrap:
def test_skip_backends(self, tmp_path) -> None:
from api.auth.provision import bootstrap
env = tmp_path / ".env"
result = bootstrap(ROOT, COMMIT, env, skip_backends=True)
assert result.env_written
assert result.ok
def test_all_backends_fail_records_errors(self, tmp_path) -> None:
from api.auth.provision import bootstrap
env = tmp_path / ".env"
with (
patch(
"api.auth.provision.bootstrap_postgres",
side_effect=Exception("pg down"),
),
patch(
"api.auth.provision.provision_gitea",
side_effect=Exception("gitea down"),
),
patch(
"api.auth.provision.provision_woodpecker",
side_effect=Exception("wp down"),
),
patch(
"api.auth.provision.provision_gitea_actions",
side_effect=Exception("actions down"),
),
):
result = bootstrap(ROOT, COMMIT, env)
assert result.env_written
error_backends = [b for b, _ in result.errors]
assert "postgres" in error_backends
assert "gitea" in error_backends
assert "woodpecker" in error_backends
assert "gitea-actions" in error_backends
def test_success_path(self, tmp_path) -> None:
from api.auth.provision import bootstrap
env = tmp_path / ".env"
with (
patch("api.auth.provision.bootstrap_postgres"),
patch("api.auth.provision.provision_postgres"),
patch("api.auth.provision.provision_gitea", return_value="tok"),
patch("api.auth.provision.provision_woodpecker"),
patch("api.auth.provision.provision_gitea_actions"),
):
result = bootstrap(ROOT, COMMIT, env)
assert result.ok
assert result.postgres
assert result.gitea

232
tests/api/test_sigv4.py Normal file
View File

@@ -0,0 +1,232 @@
"""Tests for AWS SigV4 signing — api.clients.rustfs.sigv4."""
from __future__ import annotations
import hashlib
import hmac
from datetime import datetime, timezone
from api.clients.rustfs.sigv4 import (
_canonical_query_string,
_get_signature_key,
_sign,
sign_request,
)
class TestSign:
def test_returns_bytes(self) -> None:
result = _sign(b"key", "msg")
assert isinstance(result, bytes)
assert len(result) == 32 # SHA-256 digest is 32 bytes
def test_deterministic(self) -> None:
a = _sign(b"key", "hello")
b = _sign(b"key", "hello")
assert a == b
def test_different_msgs(self) -> None:
a = _sign(b"key", "hello")
b = _sign(b"key", "world")
assert a != b
def test_different_keys(self) -> None:
a = _sign(b"key1", "msg")
b = _sign(b"key2", "msg")
assert a != b
def test_matches_stdlib(self) -> None:
key = b"test-key"
msg = "test-message"
expected = hmac.new(key, msg.encode(), hashlib.sha256).digest()
assert _sign(key, msg) == expected
class TestGetSignatureKey:
def test_returns_bytes(self) -> None:
key = _get_signature_key("mysecret", "20260101", "us-east-1", "s3")
assert isinstance(key, bytes)
def test_deterministic(self) -> None:
a = _get_signature_key("secret", "20260101", "us-east-1", "s3")
b = _get_signature_key("secret", "20260101", "us-east-1", "s3")
assert a == b
def test_different_regions_produce_different_keys(self) -> None:
a = _get_signature_key("secret", "20260101", "us-east-1", "s3")
b = _get_signature_key("secret", "20260101", "eu-west-1", "s3")
assert a != b
def test_different_dates_produce_different_keys(self) -> None:
a = _get_signature_key("secret", "20260101", "us-east-1", "s3")
b = _get_signature_key("secret", "20260102", "us-east-1", "s3")
assert a != b
def test_different_services_produce_different_keys(self) -> None:
a = _get_signature_key("secret", "20260101", "us-east-1", "s3")
b = _get_signature_key("secret", "20260101", "us-east-1", "ec2")
assert a != b
class TestCanonicalQueryString:
def test_empty_params(self) -> None:
assert _canonical_query_string(None) == ""
assert _canonical_query_string({}) == ""
def test_single_param(self) -> None:
result = _canonical_query_string({"list-type": "2"})
assert result == "list-type=2"
def test_params_sorted(self) -> None:
result = _canonical_query_string({"z": "1", "a": "2"})
assert result == "a=2&z=1"
def test_special_chars_encoded(self) -> None:
result = _canonical_query_string({"key": "hello world"})
assert result == "key=hello%20world"
def test_slash_encoded(self) -> None:
result = _canonical_query_string({"prefix": "data/files/"})
assert "data%2Ffiles%2F" in result
def test_multiple_params(self) -> None:
result = _canonical_query_string({"b": "2", "a": "1", "c": "3"})
assert result == "a=1&b=2&c=3"
class TestSignRequest:
_NOW = datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
def _sign(self, method="GET", url="http://s3.example.com/bucket/key", **kwargs):
return sign_request(
method=method,
url=url,
headers={},
body=None,
access_key="AKIAIOSFODNN7EXAMPLE",
secret_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
now=self._NOW,
**kwargs,
)
def test_returns_dict(self) -> None:
result = self._sign()
assert isinstance(result, dict)
def test_authorization_header_present(self) -> None:
result = self._sign()
assert "Authorization" in result
def test_authorization_starts_with_algo(self) -> None:
result = self._sign()
assert result["Authorization"].startswith("AWS4-HMAC-SHA256 ")
def test_x_amz_date_set(self) -> None:
result = self._sign()
assert result["x-amz-date"] == "20260115T120000Z"
def test_x_amz_content_sha256_set(self) -> None:
result = self._sign()
assert "x-amz-content-sha256" in result
def test_body_is_none_uses_empty_hash(self) -> None:
result = self._sign() # body=None by default via _sign helper
# SHA-256 of empty bytes (None treated as b"")
empty_hash = hashlib.sha256(b"").hexdigest()
assert result["x-amz-content-sha256"] == empty_hash
def test_body_hash_included(self) -> None:
result = sign_request(
method="PUT",
url="http://s3.example.com/bucket/key",
headers={},
body=b"hello world",
access_key="AKID",
secret_key="SECRET",
now=self._NOW,
)
expected = hashlib.sha256(b"hello world").hexdigest()
assert result["x-amz-content-sha256"] == expected
def test_authorization_contains_credential(self) -> None:
result = self._sign()
assert "Credential=AKIAIOSFODNN7EXAMPLE/20260115" in result["Authorization"]
def test_authorization_contains_signed_headers(self) -> None:
result = self._sign()
assert "SignedHeaders=" in result["Authorization"]
def test_authorization_contains_signature(self) -> None:
result = self._sign()
assert "Signature=" in result["Authorization"]
def test_non_standard_port_signed(self) -> None:
"""Port 9000 should be included in the host header (affects signature)."""
result_with_port = self._sign(url="http://s3.example.com:9000/bucket/key")
result_without_port = self._sign(url="http://s3.example.com/bucket/key")
# Different hosts → different signatures
assert result_with_port["Authorization"] != result_without_port["Authorization"]
def test_standard_port_80_same_as_no_port(self) -> None:
"""Port 80 on HTTP is standard and should not affect the host."""
result_port80 = self._sign(url="http://s3.example.com:80/bucket/key")
result_no_port = self._sign(url="http://s3.example.com/bucket/key")
assert result_port80["Authorization"] == result_no_port["Authorization"]
def test_https_port_443_same_as_no_port(self) -> None:
"""Port 443 on HTTPS is standard and should not affect the host."""
result_443 = self._sign(url="https://s3.example.com:443/bucket/key")
result_no_port = self._sign(url="https://s3.example.com/bucket/key")
assert result_443["Authorization"] == result_no_port["Authorization"]
def test_existing_headers_preserved(self) -> None:
# Use all-lowercase header to avoid the dead-code bug in lines 96-100
# (the code overwrites canonical_headers at line 102 anyway)
result = sign_request(
method="GET",
url="http://s3.example.com/",
headers={"x-custom": "value"},
body=None,
access_key="AKID",
secret_key="SECRET",
now=self._NOW,
)
assert "x-custom" in result
def test_query_string_params_signed(self) -> None:
result = self._sign(
url="http://s3.example.com/bucket?list-type=2&prefix=data%2F"
)
auth = result["Authorization"]
assert "Signature=" in auth
def test_now_defaults_to_utc(self) -> None:
# When now=None, it uses current UTC time — just verify it doesn't crash
result = sign_request(
method="GET",
url="http://s3.example.com/",
headers={},
body=None,
access_key="AKID",
secret_key="SECRET",
)
assert "Authorization" in result
def test_custom_region_and_service(self) -> None:
result = sign_request(
method="GET",
url="http://example.com/",
headers={},
body=None,
access_key="AKID",
secret_key="SECRET",
region="eu-west-1",
service="execute-api",
now=self._NOW,
)
assert "eu-west-1/execute-api/aws4_request" in result["Authorization"]
def test_deterministic_with_fixed_time(self) -> None:
r1 = self._sign()
r2 = self._sign()
assert r1["Authorization"] == r2["Authorization"]

View File

@@ -0,0 +1,253 @@
"""Tests for opps.calcs.payment — OPPS payment calculations."""
from __future__ import annotations
import polars as pl
from opps.calcs.payment import payment, skin_sub_impact
# ── Fixtures ──────────────────────────────────────────────────────────────────
def _apc_df():
return pl.DataFrame(
{
"hcpcs": ["G0499", "G0500"],
"apc": ["5115", "5116"],
"relative_weight": [1.2, 0.8],
"payment_rate": [100.0, 80.0],
"status_indicator": ["J1", "J1"],
"cbsa": ["10180", "99999"],
}
)
def _wage_index_df():
return pl.DataFrame(
{
"cbsa": ["10180"],
"wage_index": [1.05],
}
)
def _claims_df():
return pl.DataFrame(
{
"hcpcs_code": ["Q4151", "Q4152"],
"units": [2, 3],
"paid_amount": [250.0, 400.0],
}
)
def _asp_df():
return pl.DataFrame(
{
"hcpcs_code": ["Q4151", "Q4152"],
"asp_per_unit": [100.0, 120.0],
"payment_limit": [106.0, 127.2], # ASP + 6%
}
)
# ── payment() ─────────────────────────────────────────────────────────────────
class TestPayment:
def test_returns_dataframe(self) -> None:
result = payment(_apc_df(), _wage_index_df())
assert isinstance(result, pl.DataFrame)
def test_adjusted_payment_column_present(self) -> None:
result = payment(_apc_df(), _wage_index_df())
assert "adjusted_payment" in result.columns
def test_row_count_matches_input(self) -> None:
result = payment(_apc_df(), _wage_index_df())
assert len(result) == 2
def test_payment_formula_with_wage_index(self) -> None:
"""For cbsa 10180: payment = 100 * (0.60 * 1.05 + 0.40) = 103.0."""
result = payment(_apc_df(), _wage_index_df())
row = result.filter(pl.col("hcpcs") == "G0499")
expected = round(100.0 * (0.60 * 1.05 + 0.40), 2)
assert abs(row["adjusted_payment"][0] - expected) < 0.01
def test_missing_cbsa_uses_default_wage_index(self) -> None:
"""cbsa 99999 not in wage_index → fill_null(1.0) → no adjustment."""
result = payment(_apc_df(), _wage_index_df())
row = result.filter(pl.col("hcpcs") == "G0500")
expected = round(80.0 * (0.60 * 1.0 + 0.40), 2)
assert abs(row["adjusted_payment"][0] - expected) < 0.01
def test_custom_labor_share(self) -> None:
result = payment(_apc_df(), _wage_index_df(), labor_share=0.50)
row = result.filter(pl.col("hcpcs") == "G0499")
expected = round(100.0 * (0.50 * 1.05 + 0.50), 2)
assert abs(row["adjusted_payment"][0] - expected) < 0.01
# ── skin_sub_impact() ─────────────────────────────────────────────────────────
class TestSkinSubImpact:
def test_returns_dataframe(self) -> None:
result = skin_sub_impact(_claims_df(), _asp_df())
assert isinstance(result, pl.DataFrame)
def test_expected_columns(self) -> None:
result = skin_sub_impact(_claims_df(), _asp_df())
for col in ("old_payment", "new_payment", "payment_delta", "pct_change"):
assert col in result.columns
def test_old_payment_formula(self) -> None:
"""old_payment = payment_limit * units."""
result = skin_sub_impact(_claims_df(), _asp_df())
row = result.filter(pl.col("hcpcs_code") == "Q4151")
assert row["old_payment"][0] == round(106.0 * 2, 2)
def test_new_payment_formula(self) -> None:
"""new_payment = flat_rate * units (default $127.28)."""
result = skin_sub_impact(_claims_df(), _asp_df())
row = result.filter(pl.col("hcpcs_code") == "Q4151")
assert row["new_payment"][0] == round(127.28 * 2, 2)
def test_payment_delta(self) -> None:
result = skin_sub_impact(_claims_df(), _asp_df())
row = result.filter(pl.col("hcpcs_code") == "Q4151")
old = round(106.0 * 2, 2)
new = round(127.28 * 2, 2)
assert abs(row["payment_delta"][0] - round(new - old, 2)) < 0.01
def test_pct_change_computed(self) -> None:
result = skin_sub_impact(_claims_df(), _asp_df())
row = result.filter(pl.col("hcpcs_code") == "Q4151")
assert row["pct_change"][0] is not None
def test_custom_flat_rate(self) -> None:
result = skin_sub_impact(_claims_df(), _asp_df(), flat_rate=200.0)
row = result.filter(pl.col("hcpcs_code") == "Q4151")
assert row["new_payment"][0] == round(200.0 * 2, 2)
def test_row_count_matches_input(self) -> None:
result = skin_sub_impact(_claims_df(), _asp_df())
assert len(result) == 2
# ── opps.pipe — STEPS list ────────────────────────────────────────────────────
class TestOppsPipe:
def test_steps_is_list(self) -> None:
from opps.pipe import STEPS
assert isinstance(STEPS, list)
def test_steps_populated(self) -> None:
from opps.pipe import STEPS
assert len(STEPS) == 2
def test_step_names(self) -> None:
from opps.pipe import STEPS
names = [s[0] for s in STEPS]
assert "opps_adjusted_payment" in names
assert "opps_skin_sub_impact" in names
def test_steps_have_output_tables(self) -> None:
from opps.pipe import STEPS
from opps.table import OppsAdjustedPayment, OppsSkinSubImpact
outputs = {s[0]: s[2] for s in STEPS}
assert outputs["opps_adjusted_payment"] is OppsAdjustedPayment
assert outputs["opps_skin_sub_impact"] is OppsSkinSubImpact
# ── Express functions ─────────────────────────────────────────────────────────
class TestExpressAdjustedPayment:
def test_import(self) -> None:
from opps.express.payment import adjusted_payment
assert callable(adjusted_payment)
def test_runs_with_dataframes(self) -> None:
from opps.express.payment import adjusted_payment
result = adjusted_payment(_apc_df(), _wage_index_df())
assert "adjusted_payment" in result.columns
assert len(result) == 2
def test_formula_matches_calcs(self) -> None:
from opps.express.payment import adjusted_payment
direct = payment(_apc_df(), _wage_index_df())
via_express = adjusted_payment(_apc_df(), _wage_index_df())
assert (
direct["adjusted_payment"].to_list()
== via_express["adjusted_payment"].to_list()
)
class TestExpressSkinSubImpact:
def test_import(self) -> None:
from opps.express.payment import skin_sub_impact
assert callable(skin_sub_impact)
def test_runs_with_pass_through_and_asp(self) -> None:
from opps.express.payment import skin_sub_impact as express_fn
pass_through = pl.DataFrame(
{
"hcpcs": ["Q4151", "Q4152"],
"short_description": ["Skin sub A", "Skin sub B"],
"pass_through_type": ["biological", "biological"],
"status_indicator": ["G", "G"],
"payment_rate": [250.0, 400.0],
"asp_per_unit": [100.0, 120.0],
"year": [2025, 2025],
}
)
asp = pl.DataFrame(
{
"hcpcs_code": ["Q4151", "Q4152"],
"asp_per_unit": [100.0, 120.0],
}
)
result = express_fn(pass_through, asp)
assert "old_payment" in result.columns
assert "new_payment" in result.columns
assert len(result) == 2
# ── Pipeline registry ─────────────────────────────────────────────────────────
class TestOppsPipeline:
def test_registered_in_registry(self) -> None:
from aco.pipe import registry
assert "opps" in registry
def test_pipeline_has_two_exprs(self) -> None:
from aco.pipe import registry
assert len(registry["opps"]) == 2
def test_pipeline_expr_names(self) -> None:
from aco.pipe import registry
names = registry["opps"].names()
assert names == ["opps.adjusted_payment", "opps.skin_sub_impact"]
def test_pipeline_inputs_resolved(self) -> None:
from aco.pipe import registry
p = registry["opps"]
inputs = p.exprs[0].inputs
assert "opps.apc_weight" in inputs
assert "opps.wage_index" in inputs

View File

@@ -701,7 +701,7 @@ def test_express_pipe_module_symmetry() -> None:
pipe_mods = { pipe_mods = {
p.stem p.stem
for p in pipe_dir.glob("*.py") for p in pipe_dir.glob("*.py")
if not p.stem.startswith("_") and p.stem != "runner" if not p.stem.startswith("_") and p.stem not in ("runner", "opps")
} }
missing_in_pipe = express_mods - pipe_mods missing_in_pipe = express_mods - pipe_mods

View File

@@ -169,7 +169,7 @@ class TestPythonResources:
def test_schemas_count(self): def test_schemas_count(self):
r = self._resources() r = self._resources()
assert len(r.schemas) == 13, f"Expected 13 schemas, got {len(r.schemas)}" assert len(r.schemas) == 14, f"Expected 14 schemas, got {len(r.schemas)}"
def test_schemas_include_core(self): def test_schemas_include_core(self):
r = self._resources() r = self._resources()