|
648
|
1 import unittest
|
|
|
2 import time
|
|
|
3 from helpers import OrthancTestCase, Helpers
|
|
|
4
|
|
|
5 from orthanc_api_client import OrthancApiClient, generate_test_dicom_file
|
|
|
6
|
|
|
7 import pathlib
|
|
|
8 here = pathlib.Path(__file__).parent.resolve()
|
|
|
9
|
|
|
10
|
|
|
11 class TestHousekeeper2(OrthancTestCase):
|
|
|
12
|
|
|
13 @classmethod
|
|
|
14 def prepare(cls):
|
|
|
15 print('-------------- preparing TestHousekeeper2 tests')
|
|
|
16
|
|
|
17 cls.clear_storage(storage_name="housekeeper2")
|
|
|
18
|
|
|
19 cls.launch_orthanc_to_prepare_db(
|
|
|
20 config_name="housekeeper2_preparation",
|
|
|
21 storage_name="housekeeper2",
|
|
|
22 config={
|
|
|
23 "Housekeeper": {
|
|
|
24 "Enable": False
|
|
|
25 }
|
|
|
26 },
|
|
|
27 plugins=Helpers.plugins
|
|
|
28 )
|
|
|
29
|
|
|
30 # upload a study and keep track of data before housekeeper runs
|
|
|
31 cls.o.upload_folder(here / "../../Database/Knix/Loc")
|
|
|
32
|
|
|
33 cls.instance_before, cls.series_before, cls.study_before, cls.patient_before, cls.instance_metadata_before = cls.get_infos()
|
|
|
34
|
|
|
35 cls.kill_orthanc()
|
|
|
36 time.sleep(3)
|
|
|
37
|
|
|
38 # generate config for orthanc-under-tests (change StorageCompression and add ExtraMainDicomTags)
|
|
|
39 config_path = cls.generate_configuration(
|
|
|
40 config_name="housekeeper2_under_test",
|
|
|
41 storage_name="housekeeper2",
|
|
|
42 config={
|
|
|
43 "IngestTranscoding": "1.2.840.10008.1.2.4.80",
|
|
|
44 "ExtraMainDicomTags": {
|
|
|
45 "Patient" : ["PatientWeight", "PatientAge"],
|
|
|
46 "Study": ["NameOfPhysiciansReadingStudy"],
|
|
|
47 "Series": ["ScanOptions"],
|
|
|
48 "Instance": ["Rows", "Columns", "DerivationCodeSequence"]
|
|
|
49 },
|
|
|
50 "Housekeeper": {
|
|
|
51 "Enable": True
|
|
|
52 },
|
|
|
53 "KeepAliveTimeout": 2
|
|
|
54 },
|
|
|
55 plugins=Helpers.plugins
|
|
|
56 )
|
|
|
57
|
|
|
58 print('-------------- prepared TestHousekeeper2 tests')
|
|
|
59 if Helpers.break_after_preparation:
|
|
|
60 print(f"++++ It is now time to start your Orthanc under tests with configuration file '{config_path}' +++++")
|
|
|
61 input("Press Enter to continue")
|
|
|
62 else:
|
|
|
63 print('-------------- launching TestHousekeeper2 tests')
|
|
|
64 cls.launch_orthanc_under_tests(
|
|
|
65 config_path=config_path,
|
|
|
66 config_name="housekeeper2_under_test",
|
|
|
67 storage_name="housekeeper2",
|
|
|
68 plugins=Helpers.plugins,
|
|
|
69 enable_verbose=False
|
|
|
70 )
|
|
|
71
|
|
|
72 print('-------------- waiting for orthanc-under-tests to be available')
|
|
|
73 cls.o.wait_started()
|
|
|
74
|
|
|
75 completed = False
|
|
|
76 while not completed:
|
|
|
77 print('-------------- waiting for housekeeper2 to finish processing')
|
|
|
78 time.sleep(1)
|
|
|
79 housekeeper_status = cls.o.get_json("plugins/housekeeper/status")
|
|
|
80 completed = (housekeeper_status["LastProcessedConfiguration"]["IngestTranscoding"] == "1.2.840.10008.1.2.4.80") \
|
|
|
81 and (housekeeper_status["LastChangeToProcess"] == housekeeper_status["LastProcessedChange"])
|
|
|
82
|
|
|
83
|
|
|
84 @classmethod
|
|
|
85 def get_infos(cls):
|
|
|
86 instance_id = cls.o.lookup(
|
|
|
87 needle="1.2.840.113619.2.176.2025.1499492.7040.1171286241.704",
|
|
|
88 filter="Instance"
|
|
|
89 )[0]
|
|
|
90
|
|
|
91 instance_info = cls.o.get_json(endpoint=f"instances/{instance_id}")
|
|
|
92
|
|
|
93 series_id = instance_info["ParentSeries"]
|
|
|
94 series_info = cls.o.get_json(endpoint=f"series/{series_id}")
|
|
|
95
|
|
|
96 study_id = series_info["ParentStudy"]
|
|
|
97 study_info = cls.o.get_json(endpoint=f"studies/{study_id}")
|
|
|
98
|
|
|
99 patient_id = study_info["ParentPatient"]
|
|
|
100 patient_info = cls.o.get_json(endpoint=f"patients/{patient_id}")
|
|
|
101
|
|
|
102 instance_metadata = cls.o.get_json(endpoint=f"instances/{instance_id}/metadata?expand")
|
|
|
103 return instance_info, series_info, study_info, patient_info, instance_metadata
|
|
|
104
|
|
|
105
|
|
|
106
|
|
|
107 def test_before_after_reconstruction(self):
|
|
|
108 if self.o.is_orthanc_version_at_least(1, 12, 4):
|
|
|
109 # make sure it has run once !
|
|
|
110 housekeeper_status = self.o.get_json("housekeeper/status")
|
|
|
111 self.assertIsNotNone(housekeeper_status["LastTimeStarted"])
|
|
|
112
|
|
|
113 instance_after, series_after, study_after, patient_after, instance_metadata_after = self.get_infos()
|
|
|
114
|
|
|
115 # extra tags were not in DB before reconstruction
|
|
|
116 self.assertNotIn("Rows", self.instance_before["MainDicomTags"])
|
|
|
117 self.assertNotIn("DerivationCodeSequence", self.instance_before["MainDicomTags"])
|
|
|
118 self.assertNotIn("ScanOptions", self.series_before["MainDicomTags"])
|
|
|
119 self.assertNotIn("NameOfPhysiciansReadingStudy", self.study_before["MainDicomTags"])
|
|
|
120 self.assertNotIn("PatientWeight", self.patient_before["MainDicomTags"])
|
|
|
121
|
|
|
122 # extra tags are in DB after reconstruction
|
|
|
123 self.assertIn("Rows", instance_after["MainDicomTags"])
|
|
|
124 self.assertIn("DerivationCodeSequence", instance_after["MainDicomTags"])
|
|
|
125 self.assertIn("ScanOptions", series_after["MainDicomTags"])
|
|
|
126 self.assertIn("NameOfPhysiciansReadingStudy", study_after["MainDicomTags"])
|
|
|
127 self.assertIn("PatientWeight", patient_after["MainDicomTags"])
|
|
|
128
|
|
|
129 # instance has been transcoded and we can still access the tags
|
|
|
130 self.assertTrue(self.instance_metadata_before["TransferSyntax"] != instance_metadata_after["TransferSyntax"])
|
|
|
131 self.o.instances.get_tags(instance_after["ID"])
|
|
|
132
|
|
|
133 # the reception date and other metadata have not been updated
|
|
|
134 self.assertEqual(self.instance_metadata_before["ReceptionDate"], instance_metadata_after["ReceptionDate"])
|
|
|
135 self.assertEqual(self.instance_metadata_before["Origin"], instance_metadata_after["Origin"])
|
|
|
136 self.assertNotEqual(self.instance_before["FileUuid"], instance_after["FileUuid"]) # files ID have changed
|
