test_purge_http_cache_at_shutdown.py (4162B)
1 # This Source Code Form is subject to the terms of the Mozilla Public 2 # License, v. 2.0. If a copy of the MPL was not distributed with this 3 # file, You can obtain one at http://mozilla.org/MPL/2.0/. 4 5 import os 6 from pathlib import Path 7 8 from marionette_driver import Wait 9 from marionette_harness import MarionetteTestCase 10 11 12 class PurgeHTTPCacheAtShutdownTestCase(MarionetteTestCase): 13 def setUp(self): 14 super().setUp() 15 self.marionette.enforce_gecko_prefs({ 16 "privacy.sanitize.sanitizeOnShutdown": True, 17 "privacy.clearOnShutdown.cache": True, 18 "network.cache.shutdown_purge_in_background_task": True, 19 }) 20 21 self.profile_path = Path(self.marionette.profile_path) 22 self.cache_path = self.profile_path.joinpath("cache2") 23 24 def tearDown(self): 25 self.marionette.cleanup() 26 super().tearDown() 27 28 def cacheDirExists(self): 29 return self.cache_path.exists() 30 31 def renamedDirExists(self): 32 return any( 33 child.name.endswith(".purge.bg_rm") for child in self.profile_path.iterdir() 34 ) 35 36 def initLockDir(self): 37 self.lock_dir = None 38 with self.marionette.using_context("chrome"): 39 path = self.marionette.execute_script( 40 """ 41 return Services.dirsvc.get("UpdRootD", Ci.nsIFile).parent.parent.path; 42 """ 43 ) 44 os.makedirs(path, exist_ok=True) 45 self.lock_dir = Path(path) 46 47 def assertNoLocks(self): 48 locks = [ 49 x 50 for x in self.lock_dir.iterdir() 51 if x.is_file() and "-cachePurge" in x.name 52 ] 53 self.assertEqual(locks, [], "All locks should have been removed") 54 55 # Tests are run in lexicographical order, so test_a* is run first to 56 # cleanup locks that may be there from previous runs. 57 def test_asetup(self): 58 self.initLockDir() 59 60 self.marionette.quit() 61 62 Wait(self.marionette, timeout=60).until( 63 lambda _: not self.cacheDirExists() and not self.renamedDirExists(), 64 message="Cache directory must be removed after orderly shutdown", 65 ) 66 67 # delete locks from previous runs 68 locks = [ 69 x 70 for x in self.lock_dir.iterdir() 71 if x.is_file() and "-cachePurge" in x.name 72 ] 73 for lock in locks: 74 os.remove(lock) 75 # all locks should have been removed successfully. 76 self.assertNoLocks() 77 78 def test_ensure_cache_purge_after_in_app_quit(self): 79 self.assertTrue(self.cacheDirExists(), "Cache directory must exist") 80 self.initLockDir() 81 82 self.marionette.quit() 83 84 Wait(self.marionette, timeout=60).until( 85 lambda _: not self.cacheDirExists() and not self.renamedDirExists(), 86 message="Cache directory must be removed after orderly shutdown", 87 ) 88 89 self.assertNoLocks() 90 91 def test_longstanding_cache_purge_after_in_app_quit(self): 92 self.assertTrue(self.cacheDirExists(), "Cache directory must exist") 93 self.initLockDir() 94 95 self.marionette.set_pref( 96 "toolkit.background_tasks.remove_directory.testing.sleep_ms", 5000 97 ) 98 99 self.marionette.quit() 100 101 Wait(self.marionette, timeout=60).until( 102 lambda _: not self.cacheDirExists() and not self.renamedDirExists(), 103 message="Cache directory must be removed after orderly shutdown", 104 ) 105 106 self.assertNoLocks() 107 108 def test_ensure_cache_purge_after_forced_restart(self): 109 """ 110 Doing forced restart here to prevent the shutdown phase purging and only allow startup 111 phase one, via `CacheFileIOManager::OnDelayedStartupFinished`. 112 """ 113 self.profile_path.joinpath("foo.purge.bg_rm").mkdir() 114 self.initLockDir() 115 116 self.marionette.restart(in_app=False) 117 118 Wait(self.marionette, timeout=60).until( 119 lambda _: not self.renamedDirExists(), 120 message="Directories with .purge.bg_rm postfix must be removed at startup after" 121 "disorderly shutdown", 122 ) 123 124 self.assertNoLocks()