483
|
1 import unittest
|
|
2 import time
|
|
3 import os
|
|
4 from helpers import OrthancTestCase, Helpers
|
|
5
|
|
6 from orthanc_api_client import OrthancApiClient, generate_test_dicom_file
|
|
7 from orthanc_tools import OrthancTestDbPopulator
|
|
8
|
|
9 import pathlib
|
|
10 import glob
|
|
11 here = pathlib.Path(__file__).parent.resolve()
|
|
12
|
|
13
|
|
14 class TestDelayedDeletion(OrthancTestCase):
|
|
15
|
|
16 @classmethod
|
|
17 def prepare(cls):
|
|
18 print('-------------- preparing TestDelayedDeletion tests')
|
|
19
|
|
20 cls.clear_storage(storage_name="DelayedDeletion")
|
|
21
|
|
22 config = {
|
|
23 "DelayedDeletion": {
|
|
24 "Enable": True,
|
|
25 "ThrottleDelayMs": 200
|
|
26 }
|
|
27 }
|
|
28
|
|
29 config_path = cls.generate_configuration(
|
|
30 config_name="delayed_deletion",
|
|
31 storage_name="DelayedDeletion",
|
|
32 config=config,
|
|
33 plugins=Helpers.plugins
|
|
34 )
|
|
35
|
|
36 if Helpers.break_before_preparation:
|
|
37 print(f"++++ It is now time to start your Orthanc under tests with configuration file '{config_path}' +++++")
|
|
38 input("Press Enter to continue")
|
|
39 else:
|
|
40 print('-------------- launching DelayedDeletion preparation')
|
|
41
|
|
42 cls.launch_orthanc_to_prepare_db(
|
|
43 config_name="delayed_deletion",
|
|
44 config_path=config_path,
|
|
45 storage_name="DelayedDeletion",
|
|
46 config=config,
|
|
47 plugins=Helpers.plugins
|
|
48 )
|
|
49
|
|
50 populator = OrthancTestDbPopulator(
|
|
51 api_client=cls.o,
|
|
52 studies_count=2,
|
|
53 random_seed=42
|
|
54 )
|
|
55 populator.execute()
|
|
56
|
|
57 cls.files_count_after_preparation = len(glob.glob(os.path.join(cls.get_storage_path("DelayedDeletion"), "**"), recursive=True))
|
|
58
|
|
59 all_studies_ids = cls.o.studies.get_all_ids()
|
|
60 # delete all studies and exit Orthanc one seconds later
|
|
61 cls.o.studies.delete(orthanc_ids = all_studies_ids)
|
|
62 time.sleep(1)
|
|
63
|
|
64 if Helpers.break_before_preparation:
|
|
65 print(f"++++ It is now time stop your Orthanc +++++")
|
|
66 input("Press Enter to continue")
|
|
67 else:
|
|
68 cls.kill_orthanc()
|
|
69
|
|
70 cls.files_count_after_stop = len(glob.glob(os.path.join(cls.get_storage_path("DelayedDeletion"), "**"), recursive=True))
|
|
71
|
|
72 # speed up deletion for the second part of the tests
|
|
73 config["DelayedDeletion"]["ThrottleDelayMs"] = 0
|
|
74
|
|
75 config_path = cls.generate_configuration(
|
|
76 config_name="delayed_deletion",
|
|
77 storage_name="DelayedDeletion",
|
|
78 config=config,
|
|
79 plugins=Helpers.plugins
|
|
80 )
|
|
81
|
|
82 print('-------------- prepared DelayedDeletion tests')
|
|
83 if Helpers.break_after_preparation:
|
|
84 print(f"++++ It is now time to start your Orthanc under tests with configuration file '{config_path}' +++++")
|
|
85 input("Press Enter to continue")
|
|
86 else:
|
|
87 print('-------------- launching DelayedDeletion tests')
|
|
88 cls.launch_orthanc_under_tests(
|
|
89 config_path=config_path,
|
|
90 config_name="delayed_deletion",
|
|
91 storage_name="DelayedDeletion",
|
|
92 plugins=Helpers.plugins
|
|
93 )
|
|
94
|
|
95 print('-------------- waiting for orthanc-under-tests to be available')
|
|
96 cls.o.wait_started()
|
|
97
|
|
98
|
|
99
|
|
100 def test_resumes_pending_deletion(self):
|
|
101
|
|
102 completed = False
|
|
103 while not completed:
|
|
104 print('-------------- waiting for DelayedDeletion to finish processing')
|
|
105 time.sleep(1)
|
|
106 plugin_status = self.o.get_json("/plugins/delayed-deletion/status")
|
|
107 completed = plugin_status["FilesPendingDeletion"] == 0
|
|
108
|
|
109 self.assertTrue(completed) |