Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_allow_infinite_children(self):
qpd = Segment('QPD', validation_level=VALIDATION_LEVEL.STRICT) # last field is varies
self.assertTrue(qpd.allow_infinite_children)
pid = Segment('PID', validation_level=VALIDATION_LEVEL.STRICT) # last field is not varies
self.assertFalse(pid.allow_infinite_children)
zin = Segment('ZIN', validation_level=VALIDATION_LEVEL.STRICT) # z segment
self.assertTrue(zin.allow_infinite_children)
qpd.qpd_4 = 'abc'
qpd.add_field('qpd_4')
zin.add(Field('ZIN_100', validation_level=VALIDATION_LEVEL.STRICT))
zin.zin_4 = 'abc'
zin.add_field('zin_4')
def test_add_segment_to_message_mix(self):
a = Message('OML_O33', validation_level=VALIDATION_LEVEL.TOLERANT)
msh = Segment('MSH', validation_level=VALIDATION_LEVEL.TOLERANT)
pid = Segment('PID', validation_level=VALIDATION_LEVEL.TOLERANT)
g = Group('OML_O33_PATIENT')
g.add(pid)
a.add(msh)
a.add(g)
def test_assign_complex_field_datatype_by_get(self):
p = Segment('PID')
p.pid_5 = 'test^test'
self.assertEqual(p.pid_5.xpn_1.to_er7(), 'test')
self.assertEqual(p.pid_5.xpn_2.to_er7(), 'test')
m = Message('RSP_K21', reference=self.rsp_k21_mp, validation_level=VALIDATION_LEVEL.TOLERANT)
p = m.rsp_k21_query_response.add_segment('pid')
p.pid_5 = 'test^test'
self.assertEqual(p.pid_5.xpn_1.to_er7(), 'test')
self.assertEqual(p.pid_5.xpn_2.to_er7(), 'test')
def test_add_child_with_different_version(self):
g = Group('RSP_K21_QUERY_RESPONSE', version='2.5')
s = Segment('QPD', version='2.4')
self.assertRaises(OperationNotAllowed, g.add, s)
def test_add_not_allowed_fields_to_known_segments(self):
s = Segment('PID', validation_level=VALIDATION_LEVEL.STRICT)
self.assertRaises(ChildNotValid, s.add, Field('spm_10'))
s = Segment('PID')
self.assertRaises(ChildNotValid, s.add, Field('spm_10'))
m = Message('RSP_K21', validation_level=VALIDATION_LEVEL.STRICT, reference=self.rsp_k21_mp)
s = m.add_segment('QPD')
self.assertRaises(ChildNotValid, s.add, Field('spm_10'))
m = Message('RSP_K21', validation_level=VALIDATION_LEVEL.TOLERANT, reference=self.rsp_k21_mp)
s = m.add_segment('QPD')
self.assertRaises(ChildNotValid, s.add, Field('spm_10'))
def test_add_child_with_different_validation_level(self):
g = Group('RSP_K21_QUERY_RESPONSE', validation_level=VALIDATION_LEVEL.STRICT)
s = Segment('PID', validation_level=VALIDATION_LEVEL.TOLERANT)
self.assertRaises(OperationNotAllowed, g.add, s)
g = Group('RSP_K21_QUERY_RESPONSE', validation_level=VALIDATION_LEVEL.TOLERANT)
s = Segment('PID', validation_level=VALIDATION_LEVEL.STRICT)
self.assertRaises(OperationNotAllowed, g.add, s)
qak.qak_3 = "Q22^Specimen Labeling Instructions^IHE_LABTF"
qak.qak_4 = "1"
# add the QAK segment to the RSP_K11 message
response.add(qak)
# copy the QPD segment from the incoming message
response.QPD = m.QPD
# create a PID segment
response.PID.PID_1 = '1'
response.PID.PID_5.PID_5_1 = 'PATIENT_SURNAME'
response.PID.PID_5.PID_5_2 = 'PATIENT_NAME'
response.PID.PID_6 = "19800101"
response.PID.PID_7 = "F"
# create a SPM segment
spm = Segment("SPM")
# create an OBR segment
obr = Segment("OBR")
spm.SPM_1 = '1'
spm.SPM_2 = "12345"
obr.OBR_4 = "ORDER^DESCRIPTION"
# add spm and obr to the RSP_K11 response
response.add(spm)
response.add(obr)
else:
response = LIP.nak(m)
return response.to_mllp() # encode to ER7
if _valid_z_message_name(name):
reference = ('sequence', ())
super(Message, self).__init__(name, None, reference, version,
validation_level)
else:
raise
if encoding_chars is None:
encoding_chars = get_default_encoding_chars(version)
# TODO: Change it to support message profiles
try:
msh_reference = self.structure_by_name['MSH']['ref']
except (KeyError, TypeError):
msh_reference = None
self.msh = Segment('MSH', version=self.version, validation_level=self.validation_level,
reference=msh_reference)
self.encoding_chars = encoding_chars
self.msh.msh_7 = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
self.msh.msh_12 = self.version
def add(self, obj):
super(Segment, self).add(obj)
# updates the index of the last children not allowed
if obj.name and self.allow_infinite_children:
field_index = int(obj.name[4:])
if field_index > self._last_child_index:
self._last_child_index = field_index