Skip to content

Commit

Permalink
Implement messages handling within interchange
Browse files Browse the repository at this point in the history
Fix #19
  • Loading branch information
JocelynDelalande committed Sep 8, 2020
1 parent ddc1dd7 commit a5a2e97
Show file tree
Hide file tree
Showing 2 changed files with 100 additions and 0 deletions.
36 changes: 36 additions & 0 deletions pydifact/segmentcollection.py
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,10 @@ class Interchange(AbstractSegmentsContainer):
Functional groups are not yet supported
Messages are supported, see get_message() and get_message(), but are
optional: interchange segments can be accessed without going through
messages.
https://www.stylusstudio.com/edifact/40100/UNB_.htm
https://www.stylusstudio.com/edifact/40100/UNZ_.htm
"""
Expand Down Expand Up @@ -252,6 +256,38 @@ def get_footer_segment(self) -> Segment:
self.control_reference,
)

def get_messages(self) -> List[Message]:
message = None
for segment in self.segments:
if segment.tag == 'UNH':
if not message:
message = Message(segment.elements[0], segment.elements[1])
else:
raise SyntaxError(
"Missing UNT segment before new UNH: {}".format(segment)
)
elif segment.tag == 'UNT':
if message:
yield message
else:
raise SyntaxError(
'UNT segment without matching UNH: "{}"'.format(segment)
)
else:
if message:
message.add_segment(segment)


def add_message(self, message: Message) -> "Interchange":
segments = (
[message.get_header_segment()]
+ message.segments
+ [message.get_footer_segment()]
)
self.add_segments(i for i in segments if i is not None)
return self


@classmethod
def from_segments(
cls, segments: list or collections.Iterable
Expand Down
64 changes: 64 additions & 0 deletions tests/test_segmentcollection.py
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,70 @@ def test_empty_interchange_from_str():
)


def test_interchange_messages():
i = Interchange(
sender='1234',
recipient='3333',
timestamp=datetime.datetime(2020,1,2,22,12),
control_reference='42',
syntax_identifier=('UNOC', 1),
)

m = Message(
reference_number='42z42',
identifier=('PAORES', 93, 1, 'IA'),
)

assert(len(list(i.get_messages())) == 0)

i.add_message(m)

assert(len(list(i.get_messages())) == 1)

assert str(i) == (
"UNB+UNOC:1+1234+3333+200102:2212+42'"
"UNH+42z42+PAORES:93:1:IA'"
"UNT+42z42+0'"
"UNZ+2+42'"
)


def test_interchange_messages_from_str():
i = Interchange.from_str(
"UNB+UNOC:1+1234+3333+200102:2212+42'"
"UNH+42z42+PAORES:93:1:IA'"
"UNT+42z42+0'"
"UNZ+2+42'"
)
assert str(i) == (
"UNB+UNOC:1+1234+3333+200102:2212+42'"
"UNH+42z42+PAORES:93:1:IA'"
"UNT+42z42+0'"
"UNZ+2+42'"
)


def test_faulty_interchange_messages():
i = Interchange.from_str(
"UNB+UNOC:1+1234+3333+200102:2212+42'"
"UNH+42z42+PAORES:93:1:IA'"
"UNH+42z42+PAORES:93:1:IA'"
"UNZ+2+42'"
)

with pytest.raises(SyntaxError):
list(i.get_messages())

i = Interchange.from_str(
"UNB+UNOC:1+1234+3333+200102:2212+42'"
"UNT+42z42+0'"
"UNZ+2+42'"
)

with pytest.raises(SyntaxError):
list(i.get_messages())


def test_empty_message():
m = Message(
reference_number='42z42',
Expand Down

0 comments on commit a5a2e97

Please sign in to comment.