tyler
c3a6c53096
Three changes that unblock the daily inbound pull from Gainwell's FromHPE MFT path: 1. INBOUND_RE now accepts both 'TP' and 'tp' prefixes via inline (?i:TP) scoping — case-folding the whole pattern would let lowercase '837p' / tracking IDs through, which is invalid HCPF. The rest of the pattern is still case-sensitive. 2. _list_inbound_paramiko skips *_warn.txt entries. Gainwell's MFT drops ~583 advisory text-format notes in the same inbound dir; they come first alphabetically and were padding every poll with ~80 min of pointless downloads. 3. New SftpClient.list_inbound_names() + download_inbound() pair gives a metadata-only listing and on-demand fetch. The scheduler's existing full-listing path still works (now without the warn padding); the new path is what the new /api/admin/scheduler/pull-inbound endpoint and the 'cyclone pull-inbound' CLI use to fast-target a date range without paying the cost of a full ~6000-file download. Scheduler.process_inbound_files() runs the same per-file pipeline as a regular tick on the pre-fetched list, so dedup via processed_inbound_files still applies. Tests added in test_filenames.py (lowercase + mixed-case cases), test_sftp_paramiko.py (warn skip + no-download listing), and test_scheduler.py (process_inbound_files idempotency). With this, the daily 385-file pull for 20260624 completes in seconds via 'docker exec cyclone-backend-1 python -m cyclone pull-inbound --date 20260624 --block dzinesco' (or the equivalent POST to /api/admin/scheduler/pull-inbound?date=20260624).
214 lines
7.9 KiB
Python
214 lines
7.9 KiB
Python
"""SP9 — HCPF X12 File Naming Standards helper tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
|
|
from cyclone.edi.filenames import (
|
|
ALLOWED_FILE_TYPES,
|
|
OUTBOUND_RE,
|
|
INBOUND_RE,
|
|
build_outbound_filename,
|
|
is_inbound_filename,
|
|
is_outbound_filename,
|
|
parse_inbound_filename,
|
|
)
|
|
from cyclone.providers import InboundFilename
|
|
|
|
MT = ZoneInfo("America/Denver")
|
|
|
|
|
|
# ----- build_outbound_filename --------------------------------------------
|
|
|
|
|
|
def test_build_outbound_with_explicit_mt():
|
|
now = datetime(2026, 6, 20, 13, 22, 43, 505_000, tzinfo=MT)
|
|
name = build_outbound_filename("11525703", "837P", now_mt=now)
|
|
# HCPF outbound format: tp prefix on the tpid
|
|
assert name == "tp11525703-837P-20260620132243505-1of1.x12"
|
|
|
|
|
|
def test_build_outbound_default_extension():
|
|
now = datetime(2026, 6, 20, 13, 22, 43, 505_000, tzinfo=MT)
|
|
name = build_outbound_filename("11525703", "837P", now_mt=now)
|
|
assert name.endswith(".x12")
|
|
|
|
|
|
def test_build_outbound_custom_extension():
|
|
now = datetime(2026, 6, 20, 13, 22, 43, 505_000, tzinfo=MT)
|
|
name = build_outbound_filename("11525703", "837P", ext="txt", now_mt=now)
|
|
assert name == "tp11525703-837P-20260620132243505-1of1.txt"
|
|
|
|
|
|
def test_build_outbound_uses_mt_when_no_arg():
|
|
# Snapshot test — the timestamp will be very recent; check format only
|
|
name = build_outbound_filename("11525703", "837P")
|
|
assert OUTBOUND_RE.match(name), name
|
|
# tp11525703-837P-YYYYMMDDhhmmssSSS-1of1.x12 — 4 dash-separated parts
|
|
parts = name.split("-")
|
|
assert len(parts) == 4
|
|
assert parts[0] == "tp11525703"
|
|
assert len(parts[2]) == 17 # yyyymmddhhmmssSSS
|
|
|
|
|
|
def test_build_outbound_rejects_non_numeric_tpid():
|
|
now = datetime(2026, 6, 20, tzinfo=MT)
|
|
with pytest.raises(ValueError, match="tpid must be digits"):
|
|
build_outbound_filename("abc123", "837P", now_mt=now)
|
|
|
|
|
|
def test_build_outbound_rejects_invalid_tx():
|
|
now = datetime(2026, 6, 20, tzinfo=MT)
|
|
with pytest.raises(ValueError, match="tx must be uppercase alnum"):
|
|
build_outbound_filename("11525703", "837-lower", now_mt=now)
|
|
|
|
|
|
def test_build_outbound_rejects_naive_dt():
|
|
with pytest.raises(ValueError, match="timezone-aware"):
|
|
build_outbound_filename("11525703", "837P", now_mt=datetime(2026, 6, 20))
|
|
|
|
|
|
def test_build_outbound_converts_other_tz_to_mt():
|
|
# 2026-06-20 19:22:43 UTC = 2026-06-20 13:22:43 MT (during DST)
|
|
from datetime import timezone
|
|
now_utc = datetime(2026, 6, 20, 19, 22, 43, 505_000, tzinfo=timezone.utc)
|
|
name = build_outbound_filename("11525703", "837P", now_mt=now_utc)
|
|
assert "20260620132243505" in name
|
|
|
|
|
|
# ----- parse_inbound_filename ---------------------------------------------
|
|
|
|
|
|
def test_parse_inbound_999_real_prodfile():
|
|
# Real production 999 filename from docs/prodfiles/FromHPE/
|
|
name = "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12"
|
|
parsed = parse_inbound_filename(name)
|
|
assert parsed.tpid == "11525703"
|
|
assert parsed.orig_tx == "837P"
|
|
assert parsed.tracking == "M019048402"
|
|
assert parsed.ts == "20260520231513488"
|
|
assert parsed.file_type == "999"
|
|
assert parsed.ext == "x12"
|
|
|
|
|
|
def test_parse_inbound_ta1_real_prodfile():
|
|
name = "TP11525703-837P_M019044969-20260520180505477-1of1_TA1.x12"
|
|
parsed = parse_inbound_filename(name)
|
|
assert parsed.file_type == "TA1"
|
|
assert parsed.tracking == "M019044969"
|
|
|
|
|
|
def test_parse_inbound_277():
|
|
name = "TP11525703-837P_M019110219-20260601003507042-1of1_277.x12"
|
|
parsed = parse_inbound_filename(name)
|
|
assert parsed.file_type == "277"
|
|
|
|
|
|
def test_parse_inbound_lowercase_tp_prefix_999():
|
|
# Gainwell's production filer uses lowercase `tp` for inbound 999/TA1.
|
|
# The inbound regex must accept both casings on the TP prefix.
|
|
name = "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12"
|
|
parsed = parse_inbound_filename(name)
|
|
assert parsed.tpid == "11525703"
|
|
assert parsed.orig_tx == "837P"
|
|
assert parsed.tracking == "M019048402"
|
|
assert parsed.file_type == "999"
|
|
assert parsed.ext == "x12"
|
|
|
|
|
|
def test_parse_inbound_lowercase_tp_prefix_ta1():
|
|
name = "tp11525703-837P_M019044969-20260520180505477-1of1_TA1.x12"
|
|
parsed = parse_inbound_filename(name)
|
|
assert parsed.file_type == "TA1"
|
|
|
|
|
|
def test_is_inbound_filename_accepts_both_cases():
|
|
# is_inbound_filename() is the fast path used by the scheduler to
|
|
# filter the listing. It must accept both casings.
|
|
upper = "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12"
|
|
lower = "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12"
|
|
assert is_inbound_filename(upper)
|
|
assert is_inbound_filename(lower)
|
|
|
|
|
|
def test_parse_inbound_rejects_mixed_case_tracking():
|
|
# Tracking value must stay uppercase alnum; the case-insensitive
|
|
# flag is intentionally scoped to the TP prefix by the file_type
|
|
# and timestamp constraints, so a mixed-case tracking should still
|
|
# be rejected (it'd be invalid HCPF).
|
|
# We exercise the obvious "totally lowercase" rejection to confirm
|
|
# the rest of the pattern is still strict.
|
|
with pytest.raises(ValueError, match="Not a valid HCPF inbound"):
|
|
# Lowercase orig_tx; the orig_tx class is [A-Z0-9]+ so it
|
|
# must be uppercase.
|
|
parse_inbound_filename(
|
|
"tp11525703-837p_M019048402-20260520231513488-1of1_999.x12"
|
|
)
|
|
|
|
|
|
def test_parse_inbound_rejects_missing_tp_prefix():
|
|
with pytest.raises(ValueError, match="Not a valid HCPF inbound"):
|
|
parse_inbound_filename("11525703-837P_M019048402-20260520231513488-1of1_999.x12")
|
|
|
|
|
|
def test_parse_inbound_rejects_wrong_segment_count():
|
|
with pytest.raises(ValueError):
|
|
parse_inbound_filename("TP11525703_837P_M019048402-1of1_999.x12")
|
|
|
|
|
|
def test_parse_inbound_rejects_unknown_file_type():
|
|
with pytest.raises(ValueError, match="not in allowed HCPF set"):
|
|
parse_inbound_filename("TP11525703-837P_M019048402-20260520231513488-1of1_XXX.x12")
|
|
|
|
|
|
def test_parse_inbound_rejects_non_x12_ext():
|
|
with pytest.raises(ValueError):
|
|
parse_inbound_filename("TP11525703-837P_M019048402-20260520231513488-1of1_999.txt")
|
|
|
|
|
|
# ----- round-trip ---------------------------------------------------------
|
|
|
|
|
|
def test_roundtrip_outbound_to_inbound():
|
|
# Outbound uses tp{...}, inbound uses TP{...} (case differs but both
|
|
# prefixes are required). The two regexes use different shapes —
|
|
# round-trip via tpid only.
|
|
now = datetime(2026, 6, 20, 13, 22, 43, 505_000, tzinfo=MT)
|
|
out = build_outbound_filename("11525703", "837P", now_mt=now)
|
|
assert OUTBOUND_RE.match(out)
|
|
assert "11525703" in out
|
|
assert "837P" in out
|
|
assert out.endswith("1of1.x12")
|
|
|
|
|
|
# ----- validators ---------------------------------------------------------
|
|
|
|
|
|
def test_is_outbound_filename():
|
|
# HCPF outbound always has the lowercase "tp" prefix
|
|
assert is_outbound_filename("tp11525703-837P-20260620132243505-1of1.x12")
|
|
# Bare tpid (no tp prefix) is no longer a valid outbound filename
|
|
assert not is_outbound_filename("11525703-837P-20260620132243505-1of1.x12")
|
|
# Uppercase TP prefix is the inbound shape, not outbound
|
|
assert not is_outbound_filename("TP11525703-837P-20260620132243505-1of1.x12")
|
|
assert not is_outbound_filename("not-a-filename")
|
|
|
|
|
|
def test_is_inbound_filename():
|
|
assert is_inbound_filename("TP11525703-837P_M019048402-20260520231513488-1of1_999.x12")
|
|
# Lowercase tp prefix is now accepted too — Gainwell's filer has
|
|
# used both casings on inbound 999/TA1 files.
|
|
assert is_inbound_filename("tp11525703-837P_M019048402-20260520231513488-1of1_999.x12")
|
|
# Outbound shape is still rejected (no tracking/ts/file_type).
|
|
assert not is_inbound_filename("11525703-837P-20260620132243505-1of1.x12")
|
|
|
|
|
|
def test_allowed_file_types_includes_277ca():
|
|
assert "277CA" in ALLOWED_FILE_TYPES
|
|
assert "TA1" in ALLOWED_FILE_TYPES
|
|
assert "999" in ALLOWED_FILE_TYPES
|
|
assert "835" in ALLOWED_FILE_TYPES
|