1# -*- coding: utf-8 -*-
2from __future__ import absolute_import, unicode_literals
3
4import os
5import pickle
6import tempfile
7
8import pytest
9from case import skip
10
11from celery import states, uuid
12from celery.backends import filesystem
13from celery.backends.filesystem import FilesystemBackend
14from celery.exceptions import ImproperlyConfigured
15
16
17@skip.if_win32()
18class test_FilesystemBackend:
19
20    def setup(self):
21        self.directory = tempfile.mkdtemp()
22        self.url = 'file://' + self.directory
23        self.path = self.directory.encode('ascii')
24
25    def test_a_path_is_required(self):
26        with pytest.raises(ImproperlyConfigured):
27            FilesystemBackend(app=self.app)
28
29    def test_a_path_in_url(self):
30        tb = FilesystemBackend(app=self.app, url=self.url)
31        assert tb.path == self.path
32
33    @pytest.mark.parametrize("url,expected_error_message", [
34        ('file:///non-existing', filesystem.E_PATH_INVALID),
35        ('url://non-conforming', filesystem.E_PATH_NON_CONFORMING_SCHEME),
36        (None, filesystem.E_NO_PATH_SET)
37    ])
38    def test_raises_meaningful_errors_for_invalid_urls(
39        self,
40        url,
41        expected_error_message
42    ):
43        with pytest.raises(
44            ImproperlyConfigured,
45            match=expected_error_message
46        ):
47            FilesystemBackend(app=self.app, url=url)
48
49    def test_localhost_is_removed_from_url(self):
50        url = 'file://localhost' + self.directory
51        tb = FilesystemBackend(app=self.app, url=url)
52        assert tb.path == self.path
53
54    def test_missing_task_is_PENDING(self):
55        tb = FilesystemBackend(app=self.app, url=self.url)
56        assert tb.get_state('xxx-does-not-exist') == states.PENDING
57
58    def test_mark_as_done_writes_file(self):
59        tb = FilesystemBackend(app=self.app, url=self.url)
60        tb.mark_as_done(uuid(), 42)
61        assert len(os.listdir(self.directory)) == 1
62
63    def test_done_task_is_SUCCESS(self):
64        tb = FilesystemBackend(app=self.app, url=self.url)
65        tid = uuid()
66        tb.mark_as_done(tid, 42)
67        assert tb.get_state(tid) == states.SUCCESS
68
69    def test_correct_result(self):
70        data = {'foo': 'bar'}
71
72        tb = FilesystemBackend(app=self.app, url=self.url)
73        tid = uuid()
74        tb.mark_as_done(tid, data)
75        assert tb.get_result(tid) == data
76
77    def test_get_many(self):
78        data = {uuid(): 'foo', uuid(): 'bar', uuid(): 'baz'}
79
80        tb = FilesystemBackend(app=self.app, url=self.url)
81        for key, value in data.items():
82            tb.mark_as_done(key, value)
83
84        for key, result in tb.get_many(data.keys()):
85            assert result['result'] == data[key]
86
87    def test_forget_deletes_file(self):
88        tb = FilesystemBackend(app=self.app, url=self.url)
89        tid = uuid()
90        tb.mark_as_done(tid, 42)
91        tb.forget(tid)
92        assert len(os.listdir(self.directory)) == 0
93
94    @pytest.mark.usefixtures('depends_on_current_app')
95    def test_pickleable(self):
96        tb = FilesystemBackend(app=self.app, url=self.url, serializer='pickle')
97        assert pickle.loads(pickle.dumps(tb))
98