Mercurial > hg > orthanc-tests

comparison NewTests/MaxStorage/test_max_storage_reject.py @ 624:543e372d2265

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
added a PG max storage test
author Alain Mazy <am@osimis.io>
date Wed, 14 Feb 2024 11:25:40 +0100
parents NewTests/MaxStorageReject/test_max_storage_reject.py@47c43fdaf91d
children
comparison
equal deleted inserted replaced
623:3bb4b2589030 624:543e372d2265
1 import unittest
2 import time
3 import subprocess
4 import pprint
5 import os
6 from helpers import OrthancTestCase, Helpers
7
8 from orthanc_api_client import OrthancApiClient, generate_test_dicom_file
9 from orthanc_api_client import exceptions as orthanc_exceptions
10
11 import pathlib
12 here = pathlib.Path(__file__).parent.resolve()
13
14
15 def count_files_in_storage(path):
16 all_files = []
17 for root, dirs, files in os.walk(path):
18 for file in files:
19 if len(file) == 36:
20 all_files.append(file)
21
22 return len(all_files)
23
24
25 class TestMaxStorageReject(OrthancTestCase):
26
27 @classmethod
28 def prepare(cls):
29 test_name = "MaxStorageReject"
30 storage_name = "max_storage_reject"
31
32 cls.clear_storage(storage_name=storage_name)
33
34 config_path = cls.generate_configuration(
35 config_name=f"{test_name}_under_test",
36 storage_name=storage_name,
37 config={
38 "MaximumPatientCount": 2,
39 "MaximumStorageMode": "Reject"
40 },
41 plugins=Helpers.plugins
42 )
43
44 print(f'-------------- prepared {test_name} tests')
45 if Helpers.break_after_preparation:
46 print(f"++++ It is now time to start your Orthanc under tests with configuration file '{config_path}' +++++")
47 input("Press Enter to continue")
48 else:
49 print(f'-------------- launching {test_name} tests')
50 cls.launch_orthanc_under_tests(
51 config_path=config_path,
52 config_name=f"{test_name}_under_test",
53 storage_name=storage_name,
54 plugins=Helpers.plugins
55 )
56
57 print('-------------- waiting for orthanc-under-tests to be available')
58 cls.o.wait_started()
59
60
61 def test_upload_3_patients_rest_api(self):
62
63 if self.o.get_system()["ApiVersion"] > 20: # from Orthanc 1.12.1
64 self.o.delete_all_content()
65
66 # make sure the 3rd patient does not make it into the storage (through the Rest API)
67 self.o.upload_file(here / "../../Database/Brainix/Flair/IM-0001-0001.dcm")
68 self.o.upload_file(here / "../../Database/Knix/Loc/IM-0001-0001.dcm")
69 with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
70 self.o.upload_file(here / "../../Database/Phenix/IM-0001-0001.dcm")
71 self.assertEqual(507, ctx.exception.http_status_code)
72 self.assertEqual(2, len(self.o.studies.get_all_ids()))
73 self.assertEqual(2, count_files_in_storage(self.get_storage_path("max_storage_reject")))
74
75 def upload_with_store_scu(self, path):
76 subprocess.check_call([Helpers.find_executable('storescu'),
77 "-xs",
78 Helpers.get_orthanc_ip(),
79 str(Helpers.get_orthanc_dicom_port()),
80 path])
81
82 def test_upload_3_patients_c_store(self):
83
84 if self.o.get_system()["ApiVersion"] > 20: # from Orthanc 1.12.1
85 self.o.delete_all_content()
86
87 # make sure the 3rd patient does not make it into the storage (through StoreSCU)
88 self.upload_with_store_scu(here / "../../Database/Brainix/Flair/IM-0001-0001.dcm")
89 self.upload_with_store_scu(here / "../../Database/Knix/Loc/IM-0001-0001.dcm")
90 with self.assertRaises(subprocess.CalledProcessError) as ctx:
91 self.upload_with_store_scu(here / "../../Database/Phenix/IM-0001-0001.dcm")
92 self.assertEqual(2, len(self.o.studies.get_all_ids()))
93 self.assertEqual(2, count_files_in_storage(self.get_storage_path("max_storage_reject")))
94
95 def test_upload_3_patients_dicomweb(self):
96
97 if self.o.get_system()["ApiVersion"] > 20: # from Orthanc 1.12.1
98 self.o.delete_all_content()
99
100 # make sure the 3rd patient does not make it into the storage (through DicomWeb)
101 self.o.upload_files_dicom_web([here / "../../Database/Brainix/Flair/IM-0001-0001.dcm"])
102 self.o.upload_files_dicom_web([here / "../../Database/Knix/Loc/IM-0001-0001.dcm"])
103
104 with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
105 self.o.upload_files_dicom_web([here / "../../Database/Phenix/IM-0001-0001.dcm"])
106 self.assertEqual(400, ctx.exception.http_status_code)
107
108 self.assertEqual(2, len(self.o.studies.get_all_ids()))
109 self.assertEqual(2, count_files_in_storage(self.get_storage_path("max_storage_reject")))
110
111 def test_upload_3_patients_dicomweb_in_one_query(self):
112
113 if self.o.get_system()["ApiVersion"] > 20: # from Orthanc 1.12.1
114 self.o.delete_all_content()
115
116 # make sure the 3rd patient does not make it into the storage (through DicomWeb)
117 r = self.o.upload_files_dicom_web([
118 here / "../../Database/Brainix/Flair/IM-0001-0001.dcm",
119 here / "../../Database/Knix/Loc/IM-0001-0001.dcm",
120 here / "../../Database/Phenix/IM-0001-0001.dcm"
121 ])
122
123 # pprint.pprint(r)
124 self.assertEqual(2, len(self.o.studies.get_all_ids()))
125 self.assertIn('00081198', r)
126 self.assertEqual(0xA700, r['00081198']['Value'][0]['00081197']['Value'][0]) # one failed instance with out-of-resource status
127 self.assertEqual(2, count_files_in_storage(self.get_storage_path("max_storage_reject")))