Mercurial > hg > orthanc-tests
comparison NewTests/helpers.py @ 473:4ee85b016a40
added NewTests framework - only the Housekeeper tests right now
author | Alain Mazy <am@osimis.io> |
---|---|
date | Sat, 30 Apr 2022 19:38:34 +0200 |
parents | |
children | 6917a26881ed |
comparison
equal
deleted
inserted
replaced
472:d9ceb0fd5995 | 473:4ee85b016a40 |
---|---|
1 import unittest | |
2 from orthanc_api_client import OrthancApiClient | |
3 import subprocess | |
4 import json | |
5 import time | |
6 import typing | |
7 import shutil | |
8 from threading import Thread | |
9 | |
10 | |
11 import pathlib | |
12 here = pathlib.Path(__file__).parent.resolve() | |
13 | |
14 | |
15 default_base_config = { | |
16 "AuthenticationEnabled": False, | |
17 "RemoteAccessAllowed": True | |
18 } | |
19 | |
20 class Helpers: | |
21 | |
22 orthanc_under_tests_hostname: str = 'localhost' | |
23 orthanc_under_tests_http_port: int = 8042 | |
24 orthanc_under_tests_exe: str = None | |
25 orthanc_previous_version_exe: str = None | |
26 orthanc_under_tests_docker_image: str = None | |
27 skip_preparation: bool = False | |
28 break_after_preparation: bool = False | |
29 plugins: typing.List[str] = [] | |
30 | |
31 @classmethod | |
32 def get_orthanc_url(cls): | |
33 return f"http://{cls.orthanc_under_tests_hostname}:{cls.orthanc_under_tests_http_port}" | |
34 | |
35 class OrthancTestCase(unittest.TestCase): | |
36 | |
37 o: OrthancApiClient = None # the orthanc under tests api client | |
38 _orthanc_process = None | |
39 _orthanc_is_running = False | |
40 _orthanc_logger_thread = None | |
41 | |
42 @classmethod | |
43 def setUpClass(cls): | |
44 | |
45 cls.o = OrthancApiClient(Helpers.get_orthanc_url()) | |
46 cls._prepare() | |
47 | |
48 @classmethod | |
49 def tearDownClass(cls): | |
50 if not Helpers.break_after_preparation: | |
51 cls.kill_orthanc() | |
52 | |
53 @classmethod | |
54 def prepare(cls): | |
55 pass # to override | |
56 | |
57 @classmethod | |
58 def _prepare(cls): | |
59 if not Helpers.skip_preparation: | |
60 cls.prepare() | |
61 | |
62 @classmethod | |
63 def get_storage_path(cls, storage_name: str): | |
64 return str(here / "storages" / f"{storage_name}") | |
65 | |
66 @classmethod | |
67 def generate_configuration(cls, config_name: str, config: object, storage_name: str, plugins = []): | |
68 | |
69 # add plugins and default storge directory | |
70 config["Plugins"] = plugins | |
71 | |
72 if not "StorageDirectory" in config: | |
73 config["StorageDirectory"] = cls.get_storage_path(storage_name=storage_name) | |
74 | |
75 if not "Name" in config: | |
76 config["Name"] = config_name | |
77 | |
78 if not "HttpPort" in config: | |
79 config["HttpPort"] = Helpers.orthanc_under_tests_http_port | |
80 | |
81 # copy the values from the base config | |
82 for k, v in default_base_config.items(): | |
83 if not k in config: | |
84 config[k] = v | |
85 | |
86 # save to disk | |
87 path = str(here / "configurations" / f"{config_name}.json") | |
88 with open(path, "w") as f: | |
89 json.dump(config, f, indent=4) | |
90 | |
91 return path | |
92 | |
93 @classmethod | |
94 def clear_storage(cls, storage_name: str): | |
95 storage_path = cls.get_storage_path(storage_name=storage_name) | |
96 shutil.rmtree(storage_path) | |
97 | |
98 | |
99 @classmethod | |
100 def launch_orthanc_to_prepare_db(cls, config_name: str, config: object, storage_name: str, plugins = []): | |
101 # generate the configuration file | |
102 config_path = cls.generate_configuration( | |
103 config_name=config_name, | |
104 storage_name=storage_name, | |
105 config=config, | |
106 plugins=plugins | |
107 ) | |
108 | |
109 # run orthanc | |
110 if Helpers.orthanc_previous_version_exe: | |
111 cls.launch_orthanc( | |
112 exe_path=Helpers.orthanc_previous_version_exe, | |
113 config_path=config_path | |
114 ) | |
115 else: | |
116 raise RuntimeError("No orthanc_previous_version_exe defined, can not launch Orthanc") | |
117 | |
118 @classmethod | |
119 def launch_orthanc(cls, exe_path: str, config_path: str): | |
120 cls._orthanc_process = subprocess.Popen( | |
121 [exe_path, "--verbose", config_path], | |
122 stdout=subprocess.PIPE, | |
123 stderr=subprocess.PIPE | |
124 ) | |
125 | |
126 cls.o.wait_started(10) | |
127 if not cls.o.is_alive(): | |
128 output = cls.get_orthanc_process_output() | |
129 print("Orthanc output\n" + output) | |
130 | |
131 raise RuntimeError(f"Orthanc failed to start '{exe_path}', conf = '{config_path}'. Check output above") | |
132 | |
133 @classmethod | |
134 def kill_orthanc(cls): | |
135 cls._orthanc_process.kill() | |
136 output = cls.get_orthanc_process_output() | |
137 print("Orthanc output\n" + output) | |
138 cls._orthanc_process = None | |
139 | |
140 @classmethod | |
141 def get_orthanc_process_output(cls): | |
142 outputs = cls._orthanc_process.communicate() | |
143 output = "" | |
144 for o in outputs: | |
145 output += o.decode('utf-8') | |
146 return output |