Skip to content

feat: migrate map_walker.py + x12file.py producers to pyx12 codes (PR 4 of 6) #178

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
azoner merged 1 commit into from
May 6, 2026
Merged

feat: migrate map_walker.py + x12file.py producers to pyx12 codes (PR 4 of 6) #178

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
38 changes: 31 additions & 7 deletions pyx12/error_codes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,10 +89,13 @@ class ErrorCodeSpec:
SEG_2_LOOP_NOT_USED = "SEG_2_loop_not_used"
SEG_3_MANDATORY_SEGMENT_MISSING = "SEG_3_mandatory_segment_missing"
SEG_3_MANDATORY_LOOP_MISSING = "SEG_3_mandatory_loop_missing"
SEG_4_SEGMENT_REPEAT_EXCEEDED = "SEG_4_segment_repeat_exceeded"
SEG_5_SEGMENT_REPEAT_EXCEEDED = "SEG_5_segment_repeat_exceeded"
SEG_4_LOOP_REPEAT_EXCEEDED = "SEG_4_loop_repeat_exceeded"
# From x12file.py parser. SEG1 historically remapped inline to AK/IK "8";
# HL1/HL2/LX historically had no ack remap (filtered out).
SEG_1_INVALID_SEG_ID = "SEG_1_invalid_seg_id"
SEG_1_LEADING_SPACE = "SEG_1_leading_space"
SEG_8_SEGMENT_EMPTY = "SEG_8_segment_empty"
SEG_8_TRAILING_TERMINATORS = "SEG_8_trailing_terminators"
SEG_8_HL_COUNT_MISMATCH = "SEG_8_hl_count_mismatch"
SEG_8_HL_INVALID_PARENT = "SEG_8_hl_invalid_parent"
Expand Down Expand Up @@ -286,21 +289,42 @@ class ErrorCodeSpec:
ak_code="3",
ik_code="3",
),
SEG_4_SEGMENT_REPEAT_EXCEEDED: ErrorCodeSpec(
code=SEG_4_SEGMENT_REPEAT_EXCEEDED,
SEG_5_SEGMENT_REPEAT_EXCEEDED: ErrorCodeSpec(
code=SEG_5_SEGMENT_REPEAT_EXCEEDED,
level="SEG",
description="Segment repeat count exceeded",
ak_code="4",
ik_code="4",
description="Segment repeat count exceeded (segment exceeds maximum use)",
ak_code="5",
ik_code="5",
),
SEG_4_LOOP_REPEAT_EXCEEDED: ErrorCodeSpec(
code=SEG_4_LOOP_REPEAT_EXCEEDED,
level="SEG",
description="Loop repeat count exceeded",
description="Loop repeat count exceeded (loop occurs over maximum times)",
ak_code="4",
ik_code="4",
),
# --- Segment-level: parser ---
SEG_1_INVALID_SEG_ID: ErrorCodeSpec(
code=SEG_1_INVALID_SEG_ID,
level="SEG",
description="Segment identifier is malformed (parser-time invalid seg ID)",
ak_code="1",
ik_code="1",
),
SEG_1_LEADING_SPACE: ErrorCodeSpec(
code=SEG_1_LEADING_SPACE,
level="SEG",
description="Segment line started with leading whitespace (parser-time)",
ak_code="1",
ik_code="1",
),
SEG_8_SEGMENT_EMPTY: ErrorCodeSpec(
code=SEG_8_SEGMENT_EMPTY,
level="SEG",
description="Segment is empty (parser-time)",
ak_code="8",
ik_code="8",
),
# SEG_8_TRAILING_TERMINATORS captures the historical "SEG1" code which
# the visitors inline-remapped to AK/IK "8". This table entry replaces
# the inline remap.
Expand Down
41 changes: 34 additions & 7 deletions pyx12/map_walker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,15 @@
import pyx12.segment

# Intrapackage imports
from .error_codes import (
SEG_1_SEGMENT_NOT_FOUND,
SEG_2_LOOP_NOT_USED,
SEG_2_SEGMENT_NOT_USED,
SEG_3_MANDATORY_LOOP_MISSING,
SEG_3_MANDATORY_SEGMENT_MISSING,
SEG_4_LOOP_REPEAT_EXCEEDED,
SEG_5_SEGMENT_REPEAT_EXCEEDED,
)
from .error_item import SegError
from .errors import EngineError
from .nodeCounter import NodeCounter
Expand Down Expand Up @@ -250,7 +259,9 @@ def _check_seg_usage(
raise EngineError("Segment usage must be R, S, or N (got %r)" % (seg_node.usage,))
if seg_node.usage == "N":
err_str = "Segment %s found but marked as not used" % (seg_node.id)
self.errors_pending.append(SegError(err_cde="2", err_str=err_str, src_line=cur_line))
self.errors_pending.append(
SegError(err_cde=SEG_2_SEGMENT_NOT_USED, err_str=err_str, src_line=cur_line)
)
elif seg_node.usage == "R" or seg_node.usage == "S":
if (
self.counter.get_count(seg_node.x12path) > seg_node.get_max_repeat()
Expand All @@ -262,7 +273,7 @@ def _check_seg_usage(
)
self.errors_pending.append(
SegError(
err_cde="5",
err_cde=SEG_5_SEGMENT_REPEAT_EXCEEDED,
err_str=err_str,
src_line=cur_line,
map_node=seg_node,
Expand Down Expand Up @@ -319,7 +330,15 @@ def _try_match_segment_child(
fake_seg = pyx12.segment.Segment("%s" % (child.id), "~", "*", ":")
err_str = 'Mandatory segment "%s" (%s) missing' % (child.name, child.id)
self.mandatory_segs_missing.append(
MissingMandatorySeg(child, fake_seg, "3", err_str, seg_count, cur_line, ls_id)
MissingMandatorySeg(
child,
fake_seg,
SEG_3_MANDATORY_SEGMENT_MISSING,
err_str,
seg_count,
cur_line,
ls_id,
)
)
return None

Expand Down Expand Up @@ -462,7 +481,7 @@ def _seg_not_found_error(
err_str = "Segment %s not found. Started at %s" % (seg_str, orig_node.get_path())
self.errors_pending.append(
SegError(
err_cde="1",
err_cde=SEG_1_SEGMENT_NOT_FOUND,
err_str=err_str,
src_line=cur_line,
map_node=orig_node,
Expand Down Expand Up @@ -521,7 +540,13 @@ def _record_loop_missing_if_required(
err_str = 'Mandatory loop "%s" (%s) missing' % (loop_node.name, loop_node.id)
self.mandatory_segs_missing.append(
MissingMandatorySeg(
first_child_node, fake_seg, "3", err_str, seg_count, cur_line, ls_id
first_child_node,
fake_seg,
SEG_3_MANDATORY_LOOP_MISSING,
err_str,
seg_count,
cur_line,
ls_id,
)
)

Expand Down Expand Up @@ -672,7 +697,9 @@ def _check_loop_usage(
raise EngineError("Loop usage must be R, S, or N (got %r)" % (loop_node.usage,))
if loop_node.usage == "N":
err_str = "Loop %s found but marked as not used" % (loop_node.id)
self.errors_pending.append(SegError(err_cde="2", err_str=err_str, src_line=cur_line))
self.errors_pending.append(
SegError(err_cde=SEG_2_LOOP_NOT_USED, err_str=err_str, src_line=cur_line)
)
elif loop_node.usage in ("R", "S"):
self.counter.reset_to_node(loop_node.x12path)
self.counter.increment(loop_node.x12path)
Expand All @@ -684,7 +711,7 @@ def _check_loop_usage(
)
self.errors_pending.append(
SegError(
err_cde="4",
err_cde=SEG_4_LOOP_REPEAT_EXCEEDED,
err_str=err_str,
src_line=cur_line,
map_node=loop_node,
Expand Down
41 changes: 28 additions & 13 deletions pyx12/test/test_map_walker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,7 +234,7 @@ def test_loop_required_fail1(self):
apply_walk_errors(self.errh, walk_errors)
# result = node.is_valid(seg_data, self.errh)
# self.assertFalse(result)
self.assertEqual(self.errh.err_cde, "3", self.errh.err_str)
self.assertEqual(self.errh.err_cde, "SEG_3_mandatory_loop_missing", self.errh.err_str)
self.assertEqual(get_id_list(pop), [])
self.assertEqual(get_id_list(push), ["2000B"])

Expand Down Expand Up @@ -415,7 +415,7 @@ def test_segment_required_fail1(self):
seg_data = pyx12.segment.Segment("N4*NOWHERE*MA*30001~", "~", "*", ":")
(node, pop, push, walk_errors) = self.walker.walk_errors(node, seg_data, 5, 4, None)
apply_walk_errors(self.errh, walk_errors)
self.assertEqual(self.errh.err_cde, "3", self.errh.err_str)
self.assertEqual(self.errh.err_cde, "SEG_3_mandatory_loop_missing", self.errh.err_str)
self.assertEqual(get_id_list(pop), [])
self.assertEqual(get_id_list(push), [])

Expand All @@ -432,7 +432,7 @@ def test_found_unused_segment1(self):
apply_walk_errors(self.errh, walk_errors)
# result = node.is_valid(comp, self.errh)
# self.assertTrue(result)
self.assertEqual(self.errh.err_cde, "2", self.errh.err_str)
self.assertEqual(self.errh.err_cde, "SEG_2_segment_not_used", self.errh.err_str)
self.assertEqual(get_id_list(pop), [])
self.assertEqual(get_id_list(push), [])

Expand Down Expand Up @@ -484,7 +484,7 @@ def test_segment_id_short(self):
(node, pop, push, walk_errors) = self.walker.walk_errors(node, seg_data, 5, 4, None)
apply_walk_errors(self.errh, walk_errors)
self.assertEqual(node, None)
self.assertEqual(self.errh.err_cde, "1", self.errh.err_str)
self.assertEqual(self.errh.err_cde, "SEG_1_segment_not_found", self.errh.err_str)
self.assertEqual(get_id_list(pop), [])
self.assertEqual(get_id_list(push), [])

Expand All @@ -495,7 +495,7 @@ def test_segment_id_long(self):
(node, pop, push, walk_errors) = self.walker.walk_errors(node, seg_data, 5, 4, None)
apply_walk_errors(self.errh, walk_errors)
self.assertEqual(node, None)
self.assertEqual(self.errh.err_cde, "1", self.errh.err_str)
self.assertEqual(self.errh.err_cde, "SEG_1_segment_not_found", self.errh.err_str)
self.assertEqual(get_id_list(pop), [])
self.assertEqual(get_id_list(push), [])

Expand Down Expand Up @@ -558,7 +558,7 @@ def test_count_fail1(self):
self.assertNotEqual(node, None, "Node not found")
(node, pop, push, walk_errors) = self.walker.walk_errors(node, seg_data, 5, 4, None)
apply_walk_errors(self.errh, walk_errors)
self.assertEqual(self.errh.err_cde, "5", self.errh.err_str)
self.assertEqual(self.errh.err_cde, "SEG_5_segment_repeat_exceeded", self.errh.err_str)
self.assertEqual(get_id_list(pop), [])
self.assertEqual(get_id_list(push), [])

Expand Down Expand Up @@ -602,7 +602,7 @@ def test_max_loop_count_fail1(self):
self.errh.reset()
(node, pop, push, walk_errors) = self.walker.walk_errors(node, seg_data, 5, 4, None)
apply_walk_errors(self.errh, walk_errors)
self.assertEqual(self.errh.err_cde, "4", self.errh.err_str)
self.assertEqual(self.errh.err_cde, "SEG_4_loop_repeat_exceeded", self.errh.err_str)
self.assertEqual(get_id_list(pop), ["2400"])
self.assertEqual(get_id_list(push), ["2400"])

Expand Down Expand Up @@ -703,7 +703,7 @@ def test_ord_bad1(self):
seg_data = pyx12.segment.Segment("REF*17*A232~", "~", "*", ":")
(node, pop, push, walk_errors) = self.walker.walk_errors(node, seg_data, 5, 4, None)
apply_walk_errors(self.errh, walk_errors)
self.assertEqual(self.errh.err_cde, "1", self.errh.err_str)
self.assertEqual(self.errh.err_cde, "SEG_1_segment_not_found", self.errh.err_str)
self.assertEqual(get_id_list(pop), [])
self.assertEqual(get_id_list(push), [])

Expand Down Expand Up @@ -980,7 +980,10 @@ def test_no_match_required_uncounted_accumulates_missing(self):
self.assertIsNone(result)
self.assertEqual(len(self.walker.mandatory_segs_missing), 1)
self.assertEqual(self.walker.mandatory_segs_missing[0].seg_node, child)
self.assertEqual(self.walker.mandatory_segs_missing[0].err_cde, "3")
self.assertEqual(
self.walker.mandatory_segs_missing[0].err_cde,
"SEG_3_mandatory_segment_missing",
)

def test_no_match_required_already_counted_no_accumulation(self):
"""Required child with count >= 1 + non-matching seg -> None, no accumulation."""
Expand All @@ -1007,7 +1010,10 @@ def test_match_unused_segment_emits_error_2(self):
loop_node, child, seg_data, child, 5, 4, None, []
)
self.assertIsNotNone(result)
self.assertEqual([e.err_cde for e in self.walker.errors_pending], ["2"])
self.assertEqual(
[e.err_cde for e in self.walker.errors_pending],
["SEG_2_segment_not_used"],
)

def test_match_exceeds_max_count_emits_error_5(self):
"""Increment past max_repeat on R/S match -> err_cde '5' accumulated."""
Expand All @@ -1025,7 +1031,10 @@ def test_match_exceeds_max_count_emits_error_5(self):
loop_node, child, seg_data, child, 5, 4, None, []
)
self.assertIsNotNone(result)
self.assertEqual([e.err_cde for e in self.walker.errors_pending], ["5"])
self.assertEqual(
[e.err_cde for e in self.walker.errors_pending],
["SEG_5_segment_repeat_exceeded"],
)


class TryMatchLoopChild(unittest.TestCase):
Expand Down Expand Up @@ -1092,7 +1101,10 @@ def test_no_match_required_uncounted_accumulates_missing(self):
self.assertEqual(len(self.walker.mandatory_segs_missing), 1)
# The accumulated entry references the loop's first child, not the loop itself.
self.assertEqual(self.walker.mandatory_segs_missing[0].seg_node, first_child)
self.assertEqual(self.walker.mandatory_segs_missing[0].err_cde, "3")
self.assertEqual(
self.walker.mandatory_segs_missing[0].err_cde,
"SEG_3_mandatory_loop_missing",
)

def test_no_match_required_counted_no_accumulation(self):
"""Required loop with count >= 1 + non-matching seg -> None, no accumulation."""
Expand All @@ -1112,7 +1124,10 @@ def test_match_exceeds_max_loop_count_emits_error_4(self):
seg_data = pyx12.segment.Segment("LX*51", "~", "*", ":")
result = self.walker._try_match_loop_child(child, seg_data, 5, 4, None, [])
self.assertIsNotNone(result)
self.assertEqual([e.err_cde for e in self.walker.errors_pending], ["4"])
self.assertEqual(
[e.err_cde for e in self.walker.errors_pending],
["SEG_4_loop_repeat_exceeded"],
)

# ---- pop list pass-through ----

Expand Down
20 changes: 10 additions & 10 deletions pyx12/test/test_x12file.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ def test_trailing_ele_delim(self):
if len(errors) > 0:
err_cde = errors[0][1]
err_str = errors[0][2]
self.assertEqual(err_cde, "SEG1", err_str)
self.assertEqual(err_cde, "SEG_8_trailing_terminators", err_str)


class ISA_header(X12fileTestCase):
Expand Down Expand Up @@ -260,7 +260,7 @@ def test_HL_increment_bad(self):
str1 += "GE*1*17~\n"
str1 += "IEA*1*000010121~\n"
(err_cde, err_str) = self._get_first_error(str1)
self.assertEqual(err_cde, "HL1", err_str)
self.assertEqual(err_cde, "SEG_8_hl_count_mismatch", err_str)

def test_HL_parent_good(self):
seg = None
Expand Down Expand Up @@ -290,7 +290,7 @@ def test_HL_parent_bad_invalid(self):
str1 += "GE*1*17~\n"
str1 += "IEA*1*000010121~\n"
(err_cde, err_str) = self._get_first_error(str1)
self.assertEqual(err_cde, "HL2", err_str)
self.assertEqual(err_cde, "SEG_8_hl_invalid_parent", err_str)

def xtest_HL_parent_bad_blank(self):
seg = None
Expand All @@ -305,7 +305,7 @@ def xtest_HL_parent_bad_blank(self):
str1 += "GE*1*17~\n"
str1 += "IEA*1*000010121~\n"
(err_cde, err_str) = self._get_first_error(str1)
self.assertEqual(err_cde, "HL2", err_str)
self.assertEqual(err_cde, "SEG_8_hl_invalid_parent", err_str)


class Formatting(X12fileTestCase):
Expand Down Expand Up @@ -344,7 +344,7 @@ def test_segment_id_short(self):
str1 = "ISA*00* *00* *ZZ*ZZ000 *ZZ*ZZ001 *030828*1128*U*00401*000010121*0*T*:~\n"
str1 += "Z*0019~\n"
(err_cde, err_str) = self._get_first_error(str1)
self.assertEqual(err_cde, "1", err_str)
self.assertEqual(err_cde, "SEG_1_invalid_seg_id", err_str)

def test_segment_last_space(self):
str1 = "ISA*00* *00* *ZZ*ZZ000 *ZZ*ZZ001 *030828*1128*U*00401*000010121*0*T*:~\n"
Expand All @@ -361,25 +361,25 @@ def test_segment_id_long(self):
str1 = "ISA*00* *00* *ZZ*ZZ000 *ZZ*ZZ001 *030828*1128*U*00401*000010121*0*T*:~\n"
str1 += "ZZZZ*0019~\n"
(err_cde, err_str) = self._get_first_error(str1)
self.assertEqual(err_cde, "1", err_str)
self.assertEqual(err_cde, "SEG_1_invalid_seg_id", err_str)

def test_segment_id_empty(self):
str1 = "ISA*00* *00* *ZZ*ZZ000 *ZZ*ZZ001 *030828*1128*U*00401*000010121*0*T*:~\n"
str1 += "*1~\n"
(err_cde, err_str) = self._get_first_error(str1)
self.assertEqual(err_cde, "1", err_str)
self.assertEqual(err_cde, "SEG_1_invalid_seg_id", err_str)

def test_segment_empty(self):
str1 = "ISA*00* *00* *ZZ*ZZ000 *ZZ*ZZ001 *030828*1128*U*00401*000010121*0*T*:~\n"
str1 += "TST~\n"
(err_cde, err_str) = self._get_first_error(str1)
self.assertEqual(err_cde, "8", err_str)
self.assertEqual(err_cde, "SEG_8_segment_empty", err_str)

def test_segment_trailing_space(self):
str1 = "ISA*00* *00* *ZZ*ZZ000 *ZZ*ZZ001 *030828*1128*U*00401*000010121*0*T*:~ \n"
str1 += "ZZ*0019~ \n"
(err_cde, err_str) = self._get_first_error(str1)
self.assertEqual(err_cde, "1", err_str)
self.assertEqual(err_cde, "SEG_1_leading_space", err_str)


class FileString(X12fileTestCase):
Expand Down Expand Up @@ -606,7 +606,7 @@ def test_LX_increment_bad(self):
str1 += "GE*1*17~\n"
str1 += "IEA*1*000010121~\n"
(err_cde, err_str) = self._get_first_error(str1, "837")
self.assertEqual(err_cde, "LX", err_str)
self.assertEqual(err_cde, "SEG_8_lx_count_mismatch", err_str)


class InterchangeVersion(X12fileTestCase):
Expand Down
Loading