test_cleanup.py 2.78 KB
Newer Older
Sascha Herzinger's avatar
Sascha Herzinger committed
1 2 3
"""This module provides tests for the janitor"""

import os
Sascha Herzinger's avatar
Sascha Herzinger committed
4
import json
Sascha Herzinger's avatar
Sascha Herzinger committed
5
from pathlib import Path
Sascha Herzinger's avatar
Sascha Herzinger committed
6
from shutil import rmtree
Sascha Herzinger's avatar
Sascha Herzinger committed
7

8
from fractalis.cleanup import janitor
Sascha Herzinger's avatar
Sascha Herzinger committed
9
from fractalis import app, redis, celery
Sascha Herzinger's avatar
Sascha Herzinger committed
10 11 12


# noinspection PyMissingOrEmptyDocstring,PyMissingTypeHints
Sascha Herzinger's avatar
Sascha Herzinger committed
13 14 15 16 17 18 19
class TestCleanup:

    data_dir = os.path.join(app.config['FRACTALIS_TMP_DIR'], 'data')

    def teardown_method(self, method):
        redis.flushall()
        rmtree(self.data_dir)
Sascha Herzinger's avatar
Sascha Herzinger committed
20 21

    def test_janitor_removes_untracked_files(self):
Sascha Herzinger's avatar
Sascha Herzinger committed
22 23
        os.makedirs(self.data_dir, exist_ok=True)
        Path(os.path.join(self.data_dir, 'abc')).touch()
24
        janitor()
Sascha Herzinger's avatar
Sascha Herzinger committed
25
        assert not os.path.exists(os.path.join(self.data_dir, 'abc'))
Sascha Herzinger's avatar
Sascha Herzinger committed
26 27

    def test_janitor_does_not_remove_tracked_files(self):
Sascha Herzinger's avatar
Sascha Herzinger committed
28 29
        os.makedirs(self.data_dir, exist_ok=True)
        Path(os.path.join(self.data_dir, 'abc')).touch()
Sascha Herzinger's avatar
Sascha Herzinger committed
30
        redis.set('data:abc', '')
31
        janitor()
Sascha Herzinger's avatar
Sascha Herzinger committed
32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88
        assert os.path.exists(os.path.join(self.data_dir, 'abc'))

    def test_janitor_removes_redis_keys_without_file(self, monkeypatch):
        os.makedirs(self.data_dir, exist_ok=True)
        path = os.path.join(self.data_dir, '123')
        data_state = {
            'file_path': path
        }
        redis.set('data:123', json.dumps(data_state))

        class FakeAsyncResult:
            def __init__(self, *args, **kwargs):
                self.state = 'SUCCESS'

            def get(self, *args, **kwargs):
                pass
        monkeypatch.setattr(celery, 'AsyncResult', FakeAsyncResult)
        janitor()
        assert not redis.exists('data:123')

    def test_janitor_doesnt_remove_redis_keys_with_file(self, monkeypatch):
        os.makedirs(self.data_dir, exist_ok=True)
        path = os.path.join(self.data_dir, '123')
        Path(path).touch()
        data_state = {
            'file_path': path
        }
        redis.set('data:123', json.dumps(data_state))

        class FakeAsyncResult:
            def __init__(self, *args, **kwargs):
                self.state = 'SUCCESS'

            def get(self, *args, **kwargs):
                pass
        monkeypatch.setattr(celery, 'AsyncResult', FakeAsyncResult)
        janitor()
        assert redis.exists('data:123')

    def test_janitor_doesnt_remove_redis_keys_if_not_finished(
            self, monkeypatch):
        os.makedirs(self.data_dir, exist_ok=True)
        path = os.path.join(self.data_dir, '123')
        data_state = {
            'file_path': path
        }
        redis.set('data:123', json.dumps(data_state))

        class FakeAsyncResult:
            def __init__(self, *args, **kwargs):
                self.state = 'SUBMITTED'

            def get(self, *args, **kwargs):
                pass
        monkeypatch.setattr(celery, 'AsyncResult', FakeAsyncResult)
        janitor()
        assert redis.exists('data:123')