Mercurial > hg > orthanc-tests
annotate NewTests/DelayedDeletion/test_delayed_deletion.py @ 484:ddfabe1fbee1
new tests for DelayedDeletion plugin
| author | Alain Mazy <am@osimis.io> |
|---|---|
| date | Wed, 22 Jun 2022 16:06:46 +0200 |
| parents | 45c3fe035fed |
| children | 6144ef431512 |
| rev | line source |
|---|---|
| 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 | |
|
484
ddfabe1fbee1
new tests for DelayedDeletion plugin
Alain Mazy <am@osimis.io>
parents:
483
diff
changeset
|
109 self.assertTrue(completed) |
|
ddfabe1fbee1
new tests for DelayedDeletion plugin
Alain Mazy <am@osimis.io>
parents:
483
diff
changeset
|
110 files_count_after_delayed_deletion_is_complete = len(glob.glob(os.path.join(self.get_storage_path("DelayedDeletion"), "**"), recursive=True)) |
|
ddfabe1fbee1
new tests for DelayedDeletion plugin
Alain Mazy <am@osimis.io>
parents:
483
diff
changeset
|
111 self.assertGreater(10, files_count_after_delayed_deletion_is_complete) # only the sqlite files shall remain (and . and ..) |
|
ddfabe1fbee1
new tests for DelayedDeletion plugin
Alain Mazy <am@osimis.io>
parents:
483
diff
changeset
|
112 |