[refactor] make get_blob_size() return a deferred
[soledad.git] / tests / blobs / test_fs_backend.py
1 # -*- coding: utf-8 -*-
2 # test_fs_backend.py
3 # Copyright (C) 2017 LEAP
4 #
5 # This program is free software: you can redistribute it and/or modify
6 # it under the terms of the GNU General Public License as published by
7 # the Free Software Foundation, either version 3 of the License, or
8 # (at your option) any later version.
9 #
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU General Public License for more details.
14 #
15 # You should have received a copy of the GNU General Public License
16 # along with this program. If not, see <http://www.gnu.org/licenses/>.
17 """
18 Tests for blobs backend on server side.
19 """
20 from twisted.trial import unittest
21 from twisted.internet import defer
22 from twisted.web.test.test_web import DummyRequest
23 from leap.common.files import mkdir_p
24 from leap.soledad.server import _blobs
25 from mock import Mock
26 import mock
27 import os
28 import base64
29 import pytest
30
31
32 class FilesystemBackendTestCase(unittest.TestCase):
33
34     @pytest.mark.usefixtures("method_tmpdir")
35     def test_get_tag(self):
36         expected_tag = base64.urlsafe_b64encode('B' * 16)
37         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
38         # write a blob...
39         path = backend._get_path('user', 'blob_id', '')
40         mkdir_p(os.path.split(path)[0])
41         with open(path, "w") as f:
42             f.write('A' * 40 + 'B' * 16)
43         # ...and get its tag
44         tag = backend.get_tag('user', 'blob_id')
45         self.assertEquals(expected_tag, tag)
46
47     @pytest.mark.usefixtures("method_tmpdir")
48     @defer.inlineCallbacks
49     def test_get_blob_size(self):
50         # get a backend
51         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
52         # write a blob with size=10
53         path = backend._get_path('user', 'blob_id', '')
54         mkdir_p(os.path.split(path)[0])
55         with open(path, "w") as f:
56             f.write("0123456789")
57         # check it's size
58         size = yield backend.get_blob_size('user', 'blob_id', '')
59         self.assertEquals(10, size)
60
61     @pytest.mark.usefixtures("method_tmpdir")
62     @mock.patch.object(_blobs.static, 'File')
63     @mock.patch.object(_blobs.FilesystemBlobsBackend, '_get_path',
64                        Mock(return_value='path'))
65     def test_read_blob(self, file_mock):
66         render_mock = Mock()
67         file_mock.return_value = render_mock
68         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
69         request = DummyRequest([''])
70         resource = backend.read_blob('user', 'blob_id')
71         resource.render_GET(request)
72
73         backend._get_path.assert_called_once_with('user', 'blob_id', '')
74         ctype = 'application/octet-stream'
75         _blobs.static.File.assert_called_once_with('path', defaultType=ctype)
76         render_mock.render_GET.assert_called_once_with(request)
77
78     @pytest.mark.usefixtures("method_tmpdir")
79     @mock.patch.object(os.path, 'isfile')
80     @mock.patch.object(_blobs.FilesystemBlobsBackend, '_get_path',
81                        Mock(return_value='path'))
82     @defer.inlineCallbacks
83     def test_cannot_overwrite(self, isfile):
84         isfile.return_value = True
85         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
86         with pytest.raises(_blobs.BlobExists):
87             fd = Mock()
88             yield backend.write_blob('user', 'blob_id', fd)
89
90     @pytest.mark.usefixtures("method_tmpdir")
91     @mock.patch.object(os.path, 'isfile')
92     @defer.inlineCallbacks
93     def test_write_cannot_exceed_quota(self, isfile):
94         isfile.return_value = False
95         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
96         backend.get_total_storage = lambda x: 100
97         backend.quota = 90
98         with pytest.raises(_blobs.QuotaExceeded):
99             fd = Mock()
100             yield backend.write_blob('user', 'blob_id', fd)
101
102     @pytest.mark.usefixtures("method_tmpdir")
103     def test_get_path_partitioning_by_default(self):
104         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
105         backend.path = '/somewhere/'
106         path = backend._get_path('user', 'blob_id', '')
107         expected = '/somewhere/user/default/b/blo/blob_i/blob_id'
108         self.assertEquals(path, expected)
109
110     @pytest.mark.usefixtures("method_tmpdir")
111     def test_get_path_custom(self):
112         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
113         backend.path = '/somewhere/'
114         path = backend._get_path('user', 'blob_id', 'wonderland')
115         expected = '/somewhere/user/wonderland/b/blo/blob_i/blob_id'
116         self.assertEquals(expected, path)
117
118     @pytest.mark.usefixtures("method_tmpdir")
119     def test_get_path_namespace_traversal_raises(self):
120         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
121         backend.path = '/somewhere/'
122         with pytest.raises(Exception):
123             backend._get_path('user', 'blob_id', '..')
124
125     @pytest.mark.usefixtures("method_tmpdir")
126     @mock.patch('leap.soledad.server._blobs.os.walk')
127     def test_list_blobs(self, walk_mock):
128         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
129         _ = None
130         walk_mock.return_value = [('', _, ['blob_0']), ('', _, ['blob_1'])]
131         result = backend.list_blobs('user')
132         self.assertEquals(result, ['blob_0', 'blob_1'])
133
134     @pytest.mark.usefixtures("method_tmpdir")
135     @mock.patch('leap.soledad.server._blobs.os.walk')
136     def test_list_blobs_limited_by_namespace(self, walk_mock):
137         backend = _blobs.FilesystemBlobsBackend(self.tempdir)
138         _ = None
139         walk_mock.return_value = [('', _, ['blob_0']), ('', _, ['blob_1'])]
140         result = backend.list_blobs('user', namespace='incoming')
141         self.assertEquals(result, ['blob_0', 'blob_1'])
142         target_dir = os.path.join(self.tempdir, 'user', 'incoming')
143         walk_mock.assert_called_once_with(target_dir)
144
145     @pytest.mark.usefixtures("method_tmpdir")
146     def test_path_validation_on_read_blob(self):
147         blobs_path, request = self.tempdir, DummyRequest([''])
148         backend = _blobs.FilesystemBlobsBackend(blobs_path=blobs_path)
149         with pytest.raises(Exception):
150             backend.read_blob('..', '..', request)
151         with pytest.raises(Exception):
152             backend.read_blob('user', '../../../', request)
153         with pytest.raises(Exception):
154             backend.read_blob('../../../', 'blob_id', request)
155         with pytest.raises(Exception):
156             backend.read_blob('user', 'blob_id', request, namespace='..')
157
158     @pytest.mark.usefixtures("method_tmpdir")
159     @defer.inlineCallbacks
160     def test_path_validation_on_write_blob(self):
161         blobs_path, request = self.tempdir, DummyRequest([''])
162         backend = _blobs.FilesystemBlobsBackend(blobs_path=blobs_path)
163         with pytest.raises(Exception):
164             yield backend.write_blob('..', '..', request)
165         with pytest.raises(Exception):
166             yield backend.write_blob('user', '../../../', request)
167         with pytest.raises(Exception):
168             yield backend.write_blob('../../../', 'id1', request)
169         with pytest.raises(Exception):
170             yield backend.write_blob('user', 'id2', request, namespace='..')
171
172     @pytest.mark.usefixtures("method_tmpdir")
173     @mock.patch('leap.soledad.server._blobs.os.unlink')
174     @defer.inlineCallbacks
175     def test_delete_blob(self, unlink_mock):
176         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
177         # write a blob...
178         path = backend._get_path('user', 'blob_id', '')
179         mkdir_p(os.path.split(path)[0])
180         with open(path, "w") as f:
181             f.write("bl0b")
182         # ...and delete it
183         yield backend.delete_blob('user', 'blob_id')
184         unlink_mock.assert_any_call(backend._get_path('user',
185                                                       'blob_id'))
186         unlink_mock.assert_any_call(backend._get_path('user',
187                                                       'blob_id') + '.flags')
188
189     @pytest.mark.usefixtures("method_tmpdir")
190     @mock.patch('leap.soledad.server._blobs.os.unlink')
191     @defer.inlineCallbacks
192     def test_delete_blob_custom_namespace(self, unlink_mock):
193         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
194         # write a blob...
195         path = backend._get_path('user', 'blob_id', 'trash')
196         mkdir_p(os.path.split(path)[0])
197         with open(path, "w") as f:
198             f.write("bl0b")
199         # ...and delete it
200         yield backend.delete_blob('user', 'blob_id', namespace='trash')
201         unlink_mock.assert_any_call(backend._get_path('user',
202                                                       'blob_id',
203                                                       'trash'))
204         unlink_mock.assert_any_call(backend._get_path('user',
205                                                       'blob_id',
206                                                       'trash') + '.flags')
207
208     @pytest.mark.usefixtures("method_tmpdir")
209     @defer.inlineCallbacks
210     def test_write_blob_using_namespace(self):
211         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
212         request = DummyRequest([''])
213         request.content = BytesIO('content')
214         yield backend.write_blob('user', 'blob_id', request,
215                                  namespace='custom')
216         default = yield backend.list_blobs('user', request)
217         custom = yield backend.list_blobs('user', request, namespace='custom')
218         self.assertEquals([], json.loads(default))
219         self.assertEquals(['blob_id'], json.loads(custom))