Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
primitive.AffectedSOPClassUID = '1.1.1'
primitive.AffectedSOPInstanceUID = '1.2.1'
primitive.Priority = 0x02
primitive.MoveOriginatorApplicationEntityTitle = 'UNITTEST'
primitive.MoveOriginatorMessageID = 3
# Test encode without dataset
dimse_msg = C_STORE_RQ()
dimse_msg.primitive_to_message(primitive)
p_data_list = dimse_msg.encode_msg(12, 16)
self.assertEqual(p_data_list[0].presentation_data_value_list[0][1][0:1], b'\x01')
self.assertEqual(p_data_list[-1].presentation_data_value_list[0][1][0:1], b'\x03')
self.assertEqual(dimse_msg.ID, 12)
# Test encode with dataset
ds = Dataset()
ds.PatientID = 'Test1101'
ds.PatientName = 'Tube^HeNe'
primitive.DataSet = BytesIO(encode(ds, True, True))
dimse_msg = C_STORE_RQ()
dimse_msg.primitive_to_message(primitive)
p_data_list = dimse_msg.encode_msg(13, 10)
self.assertEqual(p_data_list[0].presentation_data_value_list[0][1][0:1], b'\x01')
self.assertEqual(p_data_list[-1].presentation_data_value_list[0][1][0:1], b'\x02')
self.assertEqual(p_data_list[-2].presentation_data_value_list[0][1][0:1], b'\x00')
self.assertEqual(p_data_list[-10].presentation_data_value_list[0][1][0:1], b'\x03')
self.assertEqual(dimse_msg.ID, 13)
p_data_list = dimse_msg.encode_msg(1, 31682)
self.assertEqual(p_data_list[0].presentation_data_value_list[0][1], c_store_rq_cmd)
def setUp(self):
"""Run prior to each test"""
self.ds = Dataset()
self.ds.PatientName = '*'
self.ds.QueryRetrieveLevel = "PATIENT"
self.assertEqual(ds.LUTDescriptor[0], 1)
self.assertEqual(ds[0x00283002].VR, 'US')
self.assertEqual(ds.LUTData, 256)
self.assertEqual(ds[0x00283006].VR, 'US')
# If LUTDescriptor[0] is not 1 then LUTData VR is 'OW'
ref_ds.LUTDescriptor = b'\x02\x00\x00\x01\x10\x00' # 2\256\16
ref_ds.LUTData = b'\x00\x01\x00\x02'
ds = correct_ambiguous_vr(deepcopy(ref_ds), True) # Little endian
self.assertEqual(ds.LUTDescriptor[0], 2)
self.assertEqual(ds[0x00283002].VR, 'US')
self.assertEqual(ds.LUTData, b'\x00\x01\x00\x02')
self.assertEqual(ds[0x00283006].VR, 'OW')
# If no LUTDescriptor then VR should be unchanged
ref_ds = Dataset()
ref_ds.LUTData = b'\x00\x01'
ds = correct_ambiguous_vr(deepcopy(ref_ds), True)
self.assertEqual(ds.LUTData, b'\x00\x01')
self.assertEqual(ds[0x00283006].VR, 'US or OW')
"""Test reading only preamble and command set"""
preamble = b'\x00' * 128
prefix = b'DICM'
command = (b'\x00\x00\x00\x00\x04\x00\x00\x00\x38'
b'\x00\x00\x00\x00\x00\x02\x00\x12\x00\x00'
b'\x00\x31\x2e\x32\x2e\x38\x34\x30\x2e\x31'
b'\x30\x30\x30\x38\x2e\x31\x2e\x31\x00\x00'
b'\x00\x00\x01\x02\x00\x00\x00\x30\x00\x00'
b'\x00\x10\x01\x02\x00\x00\x00\x07\x00\x00'
b'\x00\x00\x08\x02\x00\x00\x00\x01\x01')
bytestream = preamble + prefix + command
fp = BytesIO(bytestream)
ds = read_file(fp, force=True)
self.assertTrue('MessageID' in ds)
self.assertEqual(ds.file_meta, Dataset())
def test_meta_unchanged(self):
"""Test that the meta dataset doesn't change when writing it"""
# Empty
meta = Dataset()
write_file_meta_info(self.fp, meta, enforce_standard=False)
self.assertEqual(meta, Dataset())
# Incomplete
meta = Dataset()
meta.MediaStorageSOPClassUID = '1.1'
meta.MediaStorageSOPInstanceUID = '1.2'
meta.TransferSyntaxUID = '1.3'
meta.ImplementationClassUID = '1.4'
ref_meta = deepcopy(meta)
write_file_meta_info(self.fp, meta, enforce_standard=False)
self.assertEqual(meta, ref_meta)
# Conformant
meta = Dataset()
meta.FileMetaInformationGroupLength = 62 # Correct length
def test_file_meta_unchanged(self):
"""Test no file_meta elements are added if missing."""
ds = read_file(rtplan_name)
ds.file_meta = Dataset()
ds.save_as(self.fp, write_like_original=True)
self.assertEqual(ds.file_meta, Dataset())
def test_presentation_context_transfer(self):
"""Check the presentation context generation transfer syntax"""
# str -> UID
ae = AE(scu_sop_class=['1.1'], transfer_syntax=['1.2.840.10008.1.2'])
tran_syn = ae.presentation_contexts_scu[0].TransferSyntax[0]
self.assertEqual(tran_syn, UID('1.2.840.10008.1.2'))
self.assertTrue(isinstance(tran_syn, UID))
# UID no change
ae = AE(scu_sop_class=['1.2'], transfer_syntax=[b'1.2.840.10008.1.2'])
tran_syn = ae.presentation_contexts_scu[0].TransferSyntax[0]
self.assertEqual(tran_syn, UID('1.2.840.10008.1.2'))
self.assertTrue(isinstance(tran_syn, UID))
# bytes -> UID
ae = AE(scu_sop_class=['1.3'], transfer_syntax=[UID('1.2.840.10008.1.2')])
tran_syn = ae.presentation_contexts_scu[0].TransferSyntax[0]
self.assertEqual(tran_syn, UID('1.2.840.10008.1.2'))
self.assertTrue(isinstance(tran_syn, UID))
# If not transfer syntax raise
with self.assertRaises(ValueError):
ae = AE(scu_sop_class=['1.3'], transfer_syntax=['1.2.840'])
# If not str, bytes, UID raise
with self.assertRaises(ValueError):
ae = AE(scu_sop_class=['1.3'], transfer_syntax=[123456])
# If not valid UID raise
with self.assertRaises(ValueError):
ae = AE(scu_sop_class=['1.3'], transfer_syntax=['1.2.3.4.5.'])
def test_properties(self):
""" Check property setters and getters """
item = SOPClassExtendedNegotiationSubItem()
# SOP Class UID
item.sop_class_uid = '1.1.1'
self.assertEqual(item.sop_class_uid, UID('1.1.1'))
self.assertEqual(item.sop_class_uid_length, 5)
self.assertTrue(isinstance(item.sop_class_uid, UID))
item.sop_class_uid = b'1.1.2.1'
self.assertEqual(item.sop_class_uid, UID('1.1.2.1'))
self.assertEqual(item.sop_class_uid_length, 7)
self.assertTrue(isinstance(item.sop_class_uid, UID))
item.sop_class_uid = UID('1.1.3.1.1')
self.assertEqual(item.sop_class_uid, UID('1.1.3.1.1'))
self.assertEqual(item.sop_class_uid_length, 9)
self.assertTrue(isinstance(item.sop_class_uid, UID))
self.assertEqual(item.UID, item.sop_class_uid)
with self.assertRaises(TypeError):
item.sop_class_uid = 10002
pdu = A_ASSOCIATE_RQ()
pdu.Decode(a_associate_rq_user_id_ext_neg)
item = pdu.user_information.ext_neg[0]
def test_properies(self):
""" Check property setters and getters """
uid = ImplementationClassUIDSubItem()
uid.implementation_class_uid = '1.2.826.0.1.3680043.9.3811.0.9.1'
self.assertEqual(uid.implementation_class_uid, UID('1.2.826.0.1.3680043.9.3811.0.9.1'))
uid.implementation_class_uid = b'1.2.826.0.1.3680043.9.3811.0.9.2'
self.assertEqual(uid.implementation_class_uid, UID('1.2.826.0.1.3680043.9.3811.0.9.2'))
uid.implementation_class_uid = UID('1.2.826.0.1.3680043.9.3811.0.9.3')
self.assertEqual(uid.implementation_class_uid, UID('1.2.826.0.1.3680043.9.3811.0.9.3'))
with self.assertRaises(TypeError):
uid.implementation_class_uid = 10002
def test_to_primitive(self):
""" Check converting to primitive """
pdu = A_ASSOCIATE_RQ()
pdu.Decode(a_associate_rq_com_ext_neg)
item = pdu.user_information.common_ext_neg[0]
result = item.ToParams()
check = SOPClassCommonExtendedNegotiation()
check.sop_class_uid = UID('1.2.840.10008.5.1.4.1.1.4')
check.service_class_uid = UID('1.2.840.10008.4.2')
check.related_general_sop_class_identification = [UID('1.2.840.10008.5.1.4.1.1.88.22')]
self.assertEqual(result, check)