Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 20 additions & 3 deletions pyx12/map_if/_composite.py
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,12 @@ def is_valid_errors(
self.refdes,
)
return False, [
EleError(err_cde=COMP_1_MANDATORY_MISSING, err_str=err_str, refdes=self.refdes)
EleError(
err_cde=COMP_1_MANDATORY_MISSING,
err_str=err_str,
refdes=self.refdes,
map_node=self,
)
]

# Past here, comp_data is non-None: the (None or empty)+(N/S) branch
Expand All @@ -143,12 +148,24 @@ def is_valid_errors(

if self.usage == "N" and not comp_data.is_empty():
err_str = 'Composite "%s" (%s) is marked as Not Used' % (self.name, self.refdes)
return False, [EleError(err_cde=COMP_5_NOT_USED, err_str=err_str, refdes=self.refdes)]
return False, [
EleError(
err_cde=COMP_5_NOT_USED,
err_str=err_str,
refdes=self.refdes,
map_node=self,
)
]

if len(comp_data) > self.get_child_count():
err_str = 'Too many sub-elements in composite "%s" (%s)' % (self.name, self.refdes)
errors.append(
EleError(err_cde=COMP_3_TOO_MANY_SUBELEMENTS, err_str=err_str, refdes=self.refdes)
EleError(
err_cde=COMP_3_TOO_MANY_SUBELEMENTS,
err_str=err_str,
refdes=self.refdes,
map_node=self,
)
)
valid = False
for i in range(min(len(comp_data), self.get_child_count())):
Expand Down
49 changes: 32 additions & 17 deletions pyx12/map_if/_segment.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,39 +40,54 @@

def apply_segment_errors(node: segment_if, seg_data: pyx12.segment.Segment, errh: Any) -> bool:
"""Drive a segment validation: run is_valid_errors and forward errors
with cursor maintenance. Per-element errors carry map_node from the
leaf and trigger add_ele(map_node) when the cursor changes. Seg-level
errors (too many elements, too many sub-elements, syntax violations,
mandatory composite missing, composite-not-used) leave map_node=None;
they have no element to attach to, so they route through seg_error
with X12 IK3 code "8" ("segment has data element errors") — the only
IK3 code that semantically fits these data-element-level issues.
Their original IK4 codes ("3", "10", "1", "5", "2") collide with
different IK3 semantics and would produce wrong or dropped 999
output.
with cursor maintenance.

Routing rule: any element- or composite-level error (``err_cde`` with
prefix ``ELE_`` or ``COMP_``, identified by a non-None ``map_node``)
is preserved in ``err_ele.errors`` with its specific pyx12 code, AND
triggers a single ``SEG_8_HAS_DATA_ELEMENT_ERRORS`` in
``err_seg.errors``. SEG-level validator errors (``map_node`` is
``None``) collapse to that same single ``SEG_8`` (their specific
``SEG_3_*`` / ``SEG_2_*`` / ``SEG_10_*`` codes are dropped per
PR #161 spec correctness — emitting them as IK3-04 would map to
wrong X12 semantics).

Exactly one ``SEG_8`` per segment, no matter how many element /
composite / seg-validator errors fired. The first SEG-level
validator error's err_str / err_val is preserved on the SEG_8
emission for diagnostic detail; if only element/composite errors
fired, a generic err_str is used.

Errors whose pyx12 code is in ``errh.suppress_error_codes`` are
filtered out before they reach the err_handler tree. ``ok`` still
filtered out before reaching the err_handler tree. ``ok`` still
reflects whether the validator found any errors (suppression does
not flip a bad segment to good)."""
ok, errors = node.is_valid_errors(seg_data)
suppressed: set[str] = getattr(errh, "suppress_error_codes", set()) or set()
if suppressed:
errors = [e for e in errors if e.err_cde not in suppressed]
prev_cursor = None
needs_seg_8 = False
seg_8_err_str = "Segment has data element errors"
seg_8_err_val: str | None = None
for e in errors:
if e.map_node is None:
# Element-level validator errors collapsed to a seg-level code
# per PR #161: use SEG_8_HAS_DATA_ELEMENT_ERRORS so the
# err_handler tree carries a SEG_-prefixed code consistent
# with where it lands (err_seg.errors). The original
# element-level pyx12 code is preserved in e.err_str.
errh.seg_error(SEG_8_HAS_DATA_ELEMENT_ERRORS, e.err_str, e.err_val)
# SEG-level validator error: preserve the first one's err_str
# and err_val on the SEG_8 emission for diagnostic detail.
if not needs_seg_8:
seg_8_err_str = e.err_str
seg_8_err_val = e.err_val
needs_seg_8 = True
continue
# Element- or composite-level error: keep it in err_ele.errors
# with its specific pyx12 code.
if e.map_node is not prev_cursor:
errh.add_ele(e.map_node)
prev_cursor = e.map_node
errh.ele_error(e.err_cde, e.err_str, e.err_val, e.refdes)
needs_seg_8 = True
if needs_seg_8:
errh.seg_error(SEG_8_HAS_DATA_ELEMENT_ERRORS, seg_8_err_str, seg_8_err_val)
return ok


Expand Down
86 changes: 73 additions & 13 deletions pyx12/test/test_map_if.py
Original file line number Diff line number Diff line change
Expand Up @@ -772,10 +772,17 @@ def seg_error(self, err_cde, err_str, err_val=None, src_line=None):


class ApplySegmentErrorsRouting(unittest.TestCase):
"""Regression for the 4.0.0rc2 AttributeError: seg-level errors with
map_node=None must route through errh.seg_error (IK3) rather than
errh.ele_error (IK4), since the latter dereferences cur_ele_node which
is None on the first error of a segment."""
"""Routing rule for element/composite/seg-validator errors:

* Element- and composite-level errors are preserved in
``err_ele.errors`` with their specific pyx12 code AND trigger a
single ``SEG_8_HAS_DATA_ELEMENT_ERRORS`` in ``err_seg.errors``.
* SEG-validator errors (``map_node`` is ``None``) collapse to that
same single ``SEG_8`` (their specific codes are dropped per
PR #161 spec correctness).
* Exactly one ``SEG_8`` per segment regardless of how many child
errors fired.
"""

def setUp(self):
from pyx12.map_if._segment import apply_segment_errors
Expand All @@ -784,18 +791,71 @@ def setUp(self):
param = pyx12.params.params()
self.map = pyx12.map_if.load_map_file("837.4010.X098.A1.xml", param)

def test_composite_not_used_routes_to_seg_error(self):
def test_composite_not_used_routes_to_both_ele_and_seg(self):
# REF segment with REF04 (a composite marked Not Used) supplied:
# _composite emits err_cde="5" with map_node=None. Without the
# upstream remap this would call errh.ele_error against a None
# cur_ele_node and crash with AttributeError (the production
# 4.0.0rc2 traceback). With the remap it must route to seg_error
# with code "8" — the only valid IK3 code that fits.
# _composite emits COMP_5_NOT_USED with map_node=composite_if.
# The composite error must land in err_ele (via ele_error), AND
# a SEG_8_HAS_DATA_ELEMENT_ERRORS must land in err_seg.
ref_node = self.map.getnodebypath("/ISA_LOOP/GS_LOOP/ST_LOOP/HEADER/REF")
seg_data = pyx12.segment.Segment("REF*87*004010X098A1**:1~", "~", "*", ":")
errh = _RecordingErrh()
ok = self.apply(ref_node, seg_data, errh)
self.assertFalse(ok)
self.assertIn(("seg_error", "SEG_8_has_data_element_errors", None), errh.calls)
self.assertFalse(any(c[0] == "ele_error" for c in errh.calls))
self.assertFalse(any(c[0] == "add_ele" for c in errh.calls))
seg_calls = [c for c in errh.calls if c[0] == "seg_error"]
ele_calls = [c for c in errh.calls if c[0] == "ele_error"]
add_ele_calls = [c for c in errh.calls if c[0] == "add_ele"]
self.assertEqual(
seg_calls,
[("seg_error", "SEG_8_has_data_element_errors", None)],
)
self.assertEqual(len(ele_calls), 1)
self.assertEqual(ele_calls[0][1], "COMP_5_not_used")
self.assertEqual(len(add_ele_calls), 1)

def test_single_element_error_triggers_one_seg_8(self):
# BHT segment with BHT04 = invalid date fires ELE_8_invalid_date
# at the element. Routing rule: the element error is preserved
# in err_ele AND a single SEG_8 lands in err_seg.
bht_node = self.map.getnodebypath("/ISA_LOOP/GS_LOOP/ST_LOOP/HEADER/BHT")
seg_data = pyx12.segment.Segment("BHT*0019*00*0123*99999999*1432*CH~", "~", "*", ":")
errh = _RecordingErrh()
ok = self.apply(bht_node, seg_data, errh)
self.assertFalse(ok)
seg_calls = [c for c in errh.calls if c[0] == "seg_error"]
ele_calls = [c for c in errh.calls if c[0] == "ele_error"]
self.assertEqual(
seg_calls,
[("seg_error", "SEG_8_has_data_element_errors", None)],
)
self.assertEqual(len(ele_calls), 1)
self.assertEqual(ele_calls[0][1], "ELE_8_invalid_date")

def test_multiple_element_errors_emit_one_seg_8(self):
# NM1 segment with NM108 = "MIM" fires both ELE_5_too_long
# (max_len 2) and ELE_7_invalid_code on the same element.
# Both must be preserved in err_ele, and exactly one SEG_8
# lands in err_seg.
nm1_node = self.map.getnodebypath("/ISA_LOOP/GS_LOOP/ST_LOOP/DETAIL/2000A/2010AA/NM1")
seg_data = pyx12.segment.Segment("NM1*85*1*PROVIDER**A***MIM*123~", "~", "*", ":")
errh = _RecordingErrh()
ok = self.apply(nm1_node, seg_data, errh)
self.assertFalse(ok)
seg_calls = [c for c in errh.calls if c[0] == "seg_error"]
ele_calls = [c for c in errh.calls if c[0] == "ele_error"]
self.assertEqual(
seg_calls,
[("seg_error", "SEG_8_has_data_element_errors", None)],
)
self.assertGreaterEqual(len(ele_calls), 2)
for c in ele_calls:
self.assertTrue(c[1].startswith("ELE_"))

def test_no_errors_emits_no_seg_8(self):
# Valid NM1 segment fires zero errors: neither ele_error nor
# seg_error should be called.
nm1_node = self.map.getnodebypath("/ISA_LOOP/GS_LOOP/ST_LOOP/DETAIL/2000A/2010AA/NM1")
seg_data = pyx12.segment.Segment("NM1*85*1*PROVIDER**A***24*123456789~", "~", "*", ":")
errh = _RecordingErrh()
ok = self.apply(nm1_node, seg_data, errh)
self.assertTrue(ok)
self.assertEqual(errh.calls, [])
Loading