MCPcopy
hub / github.com/celery/celery / test_cleanup

Method test_cleanup

t/unit/backends/test_filesystem.py:101–130  ·  view source on GitHub ↗
(self)

Source from the content-addressed store, hash-verified

99 @pytest.mark.skipif(sys.platform == 'win32', reason='Test can fail on '
100 'Windows/FAT due to low granularity of st_mtime')
101 def test_cleanup(self):
102 tb = FilesystemBackend(app=self.app, url=self.url)
103 yesterday_task_ids = [uuid() for i in range(10)]
104 today_task_ids = [uuid() for i in range(10)]
105 for tid in yesterday_task_ids:
106 tb.mark_as_done(tid, 42)
107 day_length = 0.2
108 time.sleep(day_length) # let FS mark some difference in mtimes
109 for tid in today_task_ids:
110 tb.mark_as_done(tid, 42)
111 with patch.object(tb, 'expires', 0):
112 tb.cleanup()
113 # test that zero expiration time prevents any cleanup
114 filenames = set(os.listdir(tb.path))
115 assert all(
116 tb.get_key_for_task(tid) in filenames
117 for tid in yesterday_task_ids + today_task_ids
118 )
119 # test that non-zero expiration time enables cleanup by file mtime
120 with patch.object(tb, 'expires', day_length):
121 tb.cleanup()
122 filenames = set(os.listdir(tb.path))
123 assert not any(
124 tb.get_key_for_task(tid) in filenames
125 for tid in yesterday_task_ids
126 )
127 assert all(
128 tb.get_key_for_task(tid) in filenames
129 for tid in today_task_ids
130 )

Callers

nothing calls this directly

Calls 6

cleanupMethod · 0.95
FilesystemBackendClass · 0.90
mark_as_doneMethod · 0.80
objectMethod · 0.80
get_key_for_taskMethod · 0.80
sleepMethod · 0.45

Tested by

no test coverage detected