comparison NewTests/ReadOnly/test_readonly_pg.py @ 701:bdb9763d53ff find-refactoring

readonly tests
author Alain Mazy <am@orthanc.team>
date Thu, 26 Sep 2024 17:58:40 +0200
parents
children
comparison
equal deleted inserted replaced
700:8561d9c88d1a 701:bdb9763d53ff
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")