Mercurial > hg > orthanc-tests
comparison NewTests/Housekeeper/test_housekeeper2.py @ 648:810390d3968a large-queries
new hk2 tests
author | Alain Mazy <am@orthanc.team> |
---|---|
date | Tue, 21 May 2024 17:10:27 +0200 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
645:a25e7a641f26 | 648:810390d3968a |
---|---|
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 |