|
701
|
1 import unittest
|
|
|
2 import time
|
|
|
3 import os
|
|
|
4 import threading
|
|
|
5 from helpers import OrthancTestCase, Helpers
|
|
|
6
|
|
|
7 from orthanc_api_client import OrthancApiClient, ChangeType
|
|
|
8 from orthanc_api_client.exceptions import HttpError
|
|
|
9 from orthanc_api_client import helpers as OrthancHelpers
|
|
|
10
|
|
|
11 from orthanc_tools import OrthancTestDbPopulator
|
|
|
12
|
|
|
13 import pathlib
|
|
|
14 import subprocess
|
|
|
15 import glob
|
|
|
16 here = pathlib.Path(__file__).parent.resolve()
|
|
|
17
|
|
|
18
|
|
|
19 class TestReadOnlyPG(OrthancTestCase):
|
|
|
20
|
|
|
21 @classmethod
|
|
|
22 def terminate(cls):
|
|
|
23
|
|
|
24 if Helpers.is_docker():
|
|
|
25 subprocess.run(["docker", "rm", "-f", "pg-server"])
|
|
|
26 else:
|
|
|
27 cls.pg_service_process.terminate()
|
|
|
28
|
|
|
29
|
|
|
30 @classmethod
|
|
|
31 def prepare(cls):
|
|
|
32 test_name = "ReadOnlyPG"
|
|
|
33 cls._storage_name = "read-only-pg" #actually not used since we are using PG storage
|
|
|
34 network_name = "read-only-pg"
|
|
|
35
|
|
|
36 print(f'-------------- preparing {test_name} tests')
|
|
|
37
|
|
|
38 pg_hostname = "localhost"
|
|
|
39 if Helpers.is_docker():
|
|
|
40 pg_hostname = "pg-server"
|
|
|
41 cls.create_docker_network(network_name)
|
|
|
42
|
|
|
43 config = {
|
|
|
44 "PostgreSQL" : {
|
|
|
45 "EnableStorage": True,
|
|
|
46 "EnableIndex": True,
|
|
|
47 "Host": pg_hostname,
|
|
|
48 "Port": 5432,
|
|
|
49 "Database": "postgres",
|
|
|
50 "Username": "postgres",
|
|
|
51 "Password": "postgres",
|
|
|
52 "IndexConnectionsCount": 10,
|
|
|
53 "MaximumConnectionRetries" : 20,
|
|
|
54 "ConnectionRetryInterval" : 1,
|
|
|
55 "TransactionMode": "ReadCommitted",
|
|
|
56 "EnableVerboseLogs": True
|
|
|
57 },
|
|
|
58 "AuthenticationEnabled": False,
|
|
|
59 "OverwriteInstances": True,
|
|
|
60 "ReadOnly": False, # disable for preparation
|
|
|
61 "DicomWeb": {
|
|
|
62 "EnableMetadataCache": False # disable for preparation
|
|
|
63 }
|
|
|
64 }
|
|
|
65
|
|
|
66 # launch the docker PG server
|
|
|
67 print('--------------- launching PostgreSQL server ------------------')
|
|
|
68
|
|
|
69 pg_cmd = [
|
|
|
70 "docker", "run", "--rm",
|
|
|
71 "-p", "5432:5432",
|
|
|
72 "--name", "pg-server",
|
|
|
73 "--env", "POSTGRES_HOST_AUTH_METHOD=trust"
|
|
|
74 ]
|
|
|
75
|
|
|
76 if Helpers.is_docker():
|
|
|
77 pg_cmd.extend(["--network", network_name])
|
|
|
78 pg_cmd.append("postgres:15")
|
|
|
79
|
|
|
80 cls.pg_service_process = subprocess.Popen(pg_cmd)
|
|
|
81 time.sleep(5)
|
|
|
82
|
|
|
83 print('--------------- launching Orthanc to prepare DB ------------------')
|
|
|
84 cls.launch_orthanc_to_prepare_db(
|
|
|
85 config_name=f"{test_name}",
|
|
|
86 storage_name=cls._storage_name,
|
|
|
87 config=config,
|
|
|
88 plugins=Helpers.plugins,
|
|
|
89 docker_network=network_name
|
|
|
90 )
|
|
|
91
|
|
|
92 # upload a study
|
|
|
93 cls.uploaded_instances_ids = cls.o.upload_folder(here / "../../Database/Knix/Loc")
|
|
|
94 cls.one_instance_id = cls.uploaded_instances_ids[0]
|
|
|
95 cls.one_series_id = cls.o.instances.get_parent_series_id(cls.one_instance_id)
|
|
|
96 cls.one_study_id = cls.o.series.get_parent_study_id(cls.one_series_id)
|
|
|
97 cls.one_patient_id = cls.o.studies.get_parent_patient_id(cls.one_study_id)
|
|
|
98
|
|
|
99 cls.kill_orthanc()
|
|
|
100
|
|
|
101 print('--------------- stopped preparation Orthanc ------------------')
|
|
|
102
|
|
|
103 time.sleep(3)
|
|
|
104
|
|
|
105 # modify config for the readonly version
|
|
|
106 config["ReadOnly"] = True
|
|
|
107 config["DicomWeb"]["EnableMetadataCache"] = True
|
|
|
108
|
|
|
109 config_path = cls.generate_configuration(
|
|
|
110 config_name=f"{test_name}",
|
|
|
111 storage_name=cls._storage_name,
|
|
|
112 config=config,
|
|
|
113 plugins=Helpers.plugins
|
|
|
114 )
|
|
|
115
|
|
|
116 if Helpers.break_after_preparation:
|
|
|
117 print(f"++++ It is now time to start your Orthanc under tests with configuration file '{config_path}' +++++")
|
|
|
118 input("Press Enter to continue")
|
|
|
119 else:
|
|
|
120 cls.launch_orthanc_under_tests(
|
|
|
121 config_name=f"{test_name}",
|
|
|
122 storage_name=cls._storage_name,
|
|
|
123 config=config,
|
|
|
124 plugins=Helpers.plugins,
|
|
|
125 docker_network=network_name
|
|
|
126 )
|
|
|
127
|
|
|
128 cls.o = OrthancApiClient(cls.o._root_url)
|
|
|
129 cls.o.wait_started()
|
|
|
130
|
|
|
131
|
|
|
132 def test_write_methods_fail(self):
|
|
|
133 self.assertRaises(Exception, lambda: self.o.upload_folder(here / "../../Database/Knix/Loc"))
|
|
|
134 self.assertRaises(Exception, lambda: self.o.instances.delete(self.one_instance_id))
|
|
|
135 self.assertRaises(Exception, lambda: self.o.series.delete(self.one_series_id))
|
|
|
136 self.assertRaises(Exception, lambda: self.o.studies.delete(self.one_study_id))
|
|
|
137 self.assertRaises(Exception, lambda: self.o.patients.delete(self.one_patient_id))
|
|
|
138
|
|
|
139 tags = self.o.instances.get_tags(self.one_instance_id)
|
|
|
140
|
|
|
141
|
|
|
142
|
|
|
143 def test_read_methods_succeed(self):
|
|
|
144 # nothing should raise
|
|
|
145 tags = self.o.instances.get_tags(self.one_instance_id)
|
|
|
146
|
|
|
147 self.o.get_json(f"/dicom-web/studies/{tags['StudyInstanceUID']}/metadata")
|
|
|
148 self.o.get_json(f"/dicom-web/studies/{tags['StudyInstanceUID']}/series/{tags['SeriesInstanceUID']}/metadata")
|
|
|
149 self.o.get_json(f"/statistics")
|
