Nora
f10ab83628
- Adds `cyclone resubmit-rejected-claims` to push corrected single-claim 837 files to the Gainwell ToHPE SFTP dir. Idempotent (stat-then-skip by byte size). One persistent paramiko session per batch with reconnect-every=50 to dodge MOVEit's silent per-session file cap (~200 puts/session, no exception). - Validates each file via `parse_837` before upload and rejects any whose payer_id is not `CO_TXIX` (catches a bad byte-fix early). - Refreshes test fixtures (`minimal_837p.txt`, `co_medicaid_837p.txt`, `co_medicaid_837p_with_renderer.txt`) and the corresponding test assertions (`test_payer.py`, `test_payer_summary.py`, `test_co_medicaid_fixture.py`, `test_parse_837.py`) from the old `SKCO0`/`COHCPF` payer IDs to `CO_TXIX`, matching PayerConfig.co_medicaid() and the HCPF 837P Companion Guide. - Adds `ingest/` to .gitignore — local scratch / production-data staging only.
84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from cyclone.parsers.parse_837 import parse
|
|
from cyclone.parsers.payer import PayerConfig
|
|
|
|
|
|
FIXTURE = Path(__file__).parent / "fixtures" / "minimal_837p.txt"
|
|
|
|
|
|
def test_parse_minimal_fixture_returns_one_claim():
|
|
text = FIXTURE.read_text()
|
|
result = parse(text, PayerConfig.co_medicaid())
|
|
assert len(result.claims) == 1
|
|
claim = result.claims[0]
|
|
assert claim.claim_id == "CLM001"
|
|
assert claim.billing_provider.npi == "1993999998"
|
|
assert claim.subscriber.last_name == "Doe"
|
|
assert claim.subscriber.first_name == "John"
|
|
assert claim.subscriber.member_id == "ABC123"
|
|
assert claim.payer.id == "CO_TXIX"
|
|
assert claim.claim.frequency_code == "1"
|
|
assert claim.claim.place_of_service == "12"
|
|
assert claim.claim.prior_auth == "PA123"
|
|
assert claim.diagnoses[0].code == "Z00"
|
|
assert claim.diagnoses[0].qualifier == "ABK"
|
|
assert len(claim.service_lines) == 1
|
|
sl = claim.service_lines[0]
|
|
assert sl.procedure.code == "99213"
|
|
assert sl.procedure.qualifier == "HC"
|
|
assert sl.charge == 100.00
|
|
|
|
|
|
def test_parse_envelope_captured():
|
|
result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid())
|
|
assert result.envelope is not None
|
|
assert result.envelope.control_number == "991102977"
|
|
assert result.envelope.transaction_date == date(2026, 6, 11)
|
|
assert result.envelope.sender_id == "11525703"
|
|
|
|
|
|
def test_parse_summary_counts():
|
|
result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid())
|
|
assert result.summary.total_claims == 1
|
|
assert result.summary.passed == 1
|
|
assert result.summary.failed == 0
|
|
|
|
|
|
def test_parse_continues_on_malformed_subscriber():
|
|
"""A bad subscriber record must not abort the whole batch."""
|
|
text = FIXTURE.read_text().replace("MI*ABC123", "MI*") # truncate member id segment
|
|
result = parse(text, PayerConfig.co_medicaid())
|
|
# Should still find the claim, but with errors
|
|
assert len(result.claims) == 1
|
|
assert result.claims[0].validation.passed is False
|
|
assert result.summary.failed == 1
|
|
|
|
|
|
def test_parse_preserves_raw_segments():
|
|
result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid())
|
|
claim = result.claims[0]
|
|
assert any(s[0] == "CLM" for s in claim.raw_segments)
|
|
assert any(s[0] == "SV1" for s in claim.raw_segments)
|
|
|
|
|
|
def test_parse_uses_generic_config_when_requested():
|
|
result = parse(FIXTURE.read_text(), PayerConfig.generic_837p())
|
|
assert len(result.claims) == 1
|
|
# No patient-loop rule on generic config
|
|
assert result.claims[0].validation.passed is True
|
|
|
|
|
|
def test_parse_837_extracts_rendering_provider_npi_from_nm1_82():
|
|
"""SP32: NM1*82 (rendering provider) populates ClaimOutput.rendering_provider_npi."""
|
|
from pathlib import Path
|
|
|
|
text = Path("tests/fixtures/co_medicaid_837p_with_renderer.txt").read_text()
|
|
result = parse(text, PayerConfig.co_medicaid())
|
|
assert len(result.claims) >= 1
|
|
# Find the claim that has the rendering NPI (others may not).
|
|
matches = [c for c in result.claims if c.rendering_provider_npi is not None]
|
|
assert len(matches) == 1
|
|
assert matches[0].rendering_provider_npi == "1234567893"
|