fix(sp24): wire up datetime alias + live-read IG constant in cyclone.reissue
Three fixes that landed during Task 5 (CLI smoke test authoring):
- cli.py: import datetime as _datetime (the existing module alias
pattern) and use it in the reissue-claims handler. The bare
datetime.now() call was a NameError waiting to happen — caught
by the empty-input smoke test.
- cyclone.reissue.core: switch ig_correctness_check to read
PATIENT_LOOP_DEFAULT_INCLUDED via the module attribute
(cyclone.parsers.serialize_837.PATIENT_LOOP_DEFAULT_INCLUDED)
instead of the import-time name binding. Monkeypatch in tests
now reflects immediately, and any future runtime mutation by a
config-loader also takes effect without a re-import.
- test_reissue_core.py: update test_parse_inputs_handles_empty_input_dir
to reflect the new 'no files found' contract — empty input dirs
now surface as exit 2 with a clear message, not exit 0.
This commit is contained in:
@@ -5,7 +5,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from datetime import date as _date
|
||||
from datetime import date as _date, datetime as _datetime
|
||||
from decimal import Decimal as _Decimal
|
||||
from pathlib import Path
|
||||
|
||||
@@ -2220,7 +2220,7 @@ def reissue_claims(
|
||||
payer_config = payer_factories[payer]()
|
||||
|
||||
# 2. Output directory layout.
|
||||
date_label = date or datetime.now().strftime("%Y-%m-%d")
|
||||
date_label = date or _datetime.now().strftime("%Y-%m-%d")
|
||||
out_dir = output_root / date_label / pipeline
|
||||
if out_dir.exists() and not no_clean:
|
||||
# Clean only the per-claim X12 files + summary; leave any
|
||||
|
||||
@@ -27,10 +27,8 @@ from zoneinfo import ZoneInfo
|
||||
from cyclone.edi.filenames import build_outbound_filename
|
||||
from cyclone.parsers.models import ClaimOutput
|
||||
from cyclone.parsers.parse_837 import parse as parse_837_text
|
||||
from cyclone.parsers.serialize_837 import (
|
||||
PATIENT_LOOP_DEFAULT_INCLUDED,
|
||||
serialize_837,
|
||||
)
|
||||
from cyclone.parsers import serialize_837 as _serialize_837_mod
|
||||
from cyclone.parsers.serialize_837 import serialize_837
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
@@ -54,13 +52,17 @@ def ig_correctness_check(*, logger: logging.Logger | None = None) -> bool:
|
||||
otherwise — callers should refuse to emit any X12 in that case.
|
||||
"""
|
||||
log = logger or _log
|
||||
if PATIENT_LOOP_DEFAULT_INCLUDED is not False:
|
||||
# Read the constant live (not at import time) so monkeypatch in
|
||||
# tests reflects immediately and a runtime mutation by a future
|
||||
# config-loader also takes effect without a re-import.
|
||||
current = _serialize_837_mod.PATIENT_LOOP_DEFAULT_INCLUDED
|
||||
if current is not False:
|
||||
log.warning(
|
||||
"REFUSING to run: PATIENT_LOOP_DEFAULT_INCLUDED = %r "
|
||||
"(expected False). The IG-correct serializer shape for "
|
||||
"SBR02='18' claims requires the 2000C patient loop to be "
|
||||
"absent; restoring the default to False fixes it.",
|
||||
PATIENT_LOOP_DEFAULT_INCLUDED,
|
||||
current,
|
||||
)
|
||||
return False
|
||||
return True
|
||||
@@ -103,6 +105,9 @@ def parse_inputs(
|
||||
|
||||
claims: list[ClaimOutput] = []
|
||||
errors: list[tuple[Path, str]] = []
|
||||
|
||||
if not raw_files:
|
||||
errors.append((input_dir, "no *.x12 / *.txt / *.edi files found"))
|
||||
for f in raw_files:
|
||||
try:
|
||||
text = f.read_text(encoding="utf-8")
|
||||
|
||||
@@ -100,7 +100,12 @@ def test_parse_inputs_tolerates_per_file_failures(tmp_path: Path):
|
||||
|
||||
|
||||
def test_parse_inputs_handles_empty_input_dir(tmp_path: Path):
|
||||
"""An empty input dir produces zero claims + zero errors (not an error)."""
|
||||
"""An empty input dir produces zero claims + a 'no files found' error.
|
||||
|
||||
The CLI exits 2 in this case (per `cyclone reissue-claims --help`).
|
||||
The error tuple's first element is the input_dir itself (not a
|
||||
per-file path) so the operator's stdout shows the offending dir.
|
||||
"""
|
||||
in_dir = tmp_path / "in"
|
||||
in_dir.mkdir()
|
||||
|
||||
@@ -108,7 +113,10 @@ def test_parse_inputs_handles_empty_input_dir(tmp_path: Path):
|
||||
claims, errors = parse_inputs(in_dir, payer)
|
||||
|
||||
assert claims == []
|
||||
assert errors == []
|
||||
assert len(errors) == 1
|
||||
bad_path, bad_msg = errors[0]
|
||||
assert bad_path == in_dir
|
||||
assert "no" in bad_msg and "files" in bad_msg.lower()
|
||||
|
||||
|
||||
def test_parse_inputs_returns_claim_models(tmp_path: Path):
|
||||
|
||||
Reference in New Issue
Block a user