59b7976f3e86f370058c47cd947b278b53f25ce1
[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     def test_get_blob_size(self):
49         # get a backend
50         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
51         # write a blob with size=10
52         path = backend._get_path('user', 'blob_id', '')
53         mkdir_p(os.path.split(path)[0])
54         with open(path, "w") as f:
55             f.write("0123456789")
56         # check it's size
57         size = backend.get_blob_size('user', 'blob_id', '')
58         self.assertEquals(10, size)
59
60     @pytest.mark.usefixtures("method_tmpdir")
61     @mock.patch.object(_blobs.static, 'File')
62     @mock.patch.object(_blobs.FilesystemBlobsBackend, '_get_path',
63                        Mock(return_value='path'))
64     def test_read_blob(self, file_mock):
65         render_mock = Mock()
66         file_mock.return_value = render_mock
67         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
68         request = DummyRequest([''])
69         resource = backend.read_blob('user', 'blob_id')
70         resource.render_GET(request)
71
72         backend._get_path.assert_called_once_with('user', 'blob_id', '')
73         ctype = 'application/octet-stream'
74         _blobs.static.File.assert_called_once_with('path', defaultType=ctype)
75         render_mock.render_GET.assert_called_once_with(request)
76
77     @pytest.mark.usefixtures("method_tmpdir")
78     @mock.patch.object(os.path, 'isfile')
79     @mock.patch.object(_blobs.FilesystemBlobsBackend, '_get_path',
80                        Mock(return_value='path'))
81     @defer.inlineCallbacks
82     def test_cannot_overwrite(self, isfile):
83         isfile.return_value = True
84         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
85         with pytest.raises(_blobs.BlobExists):
86             fd = Mock()
87             yield backend.write_blob('user', 'blob_id', fd)
88
89     @pytest.mark.usefixtures("method_tmpdir")
90     @mock.patch.object(os.path, 'isfile')
91     @defer.inlineCallbacks
92     def test_write_cannot_exceed_quota(self, isfile):
93         isfile.return_value = False
94         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
95         backend.get_total_storage = lambda x: 100
96         backend.quota = 90
97         with pytest.raises(_blobs.QuotaExceeded):
98             fd = Mock()
99             yield backend.write_blob('user', 'blob_id', fd)
100
101     @pytest.mark.usefixtures("method_tmpdir")
102     def test_get_path_partitioning_by_default(self):
103         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
104         backend.path = '/somewhere/'
105         path = backend._get_path('user', 'blob_id', '')
106         expected = '/somewhere/user/default/b/blo/blob_i/blob_id'
107         self.assertEquals(path, expected)
108
109     @pytest.mark.usefixtures("method_tmpdir")
110     def test_get_path_custom(self):
111         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
112         backend.path = '/somewhere/'
113         path = backend._get_path('user', 'blob_id', 'wonderland')
114         expected = '/somewhere/user/wonderland/b/blo/blob_i/blob_id'
115         self.assertEquals(expected, path)
116
117     @pytest.mark.usefixtures("method_tmpdir")
118     def test_get_path_namespace_traversal_raises(self):
119         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
120         backend.path = '/somewhere/'
121         with pytest.raises(Exception):
122             backend._get_path('user', 'blob_id', '..')
123
124     @pytest.mark.usefixtures("method_tmpdir")
125     @mock.patch('leap.soledad.server._blobs.os.walk')
126     def test_list_blobs(self, walk_mock):
127         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
128         _ = None
129         walk_mock.return_value = [('', _, ['blob_0']), ('', _, ['blob_1'])]
130         result = backend.list_blobs('user')
131         self.assertEquals(result, ['blob_0', 'blob_1'])
132
133     @pytest.mark.usefixtures("method_tmpdir")
134     @mock.patch('leap.soledad.server._blobs.os.walk')
135     def test_list_blobs_limited_by_namespace(self, walk_mock):
136         backend = _blobs.FilesystemBlobsBackend(self.tempdir)
137         _ = None
138         walk_mock.return_value = [('', _, ['blob_0']), ('', _, ['blob_1'])]
139         result = backend.list_blobs('user', namespace='incoming')
140         self.assertEquals(result, ['blob_0', 'blob_1'])
141         target_dir = os.path.join(self.tempdir, 'user', 'incoming')
142         walk_mock.assert_called_once_with(target_dir)
143
144     @pytest.mark.usefixtures("method_tmpdir")
145     def test_path_validation_on_read_blob(self):
146         blobs_path, request = self.tempdir, DummyRequest([''])
147         backend = _blobs.FilesystemBlobsBackend(blobs_path=blobs_path)
148         with pytest.raises(Exception):
149             backend.read_blob('..', '..', request)
150         with pytest.raises(Exception):
151             backend.read_blob('user', '../../../', request)
152         with pytest.raises(Exception):
153             backend.read_blob('../../../', 'blob_id', request)
154         with pytest.raises(Exception):
155             backend.read_blob('user', 'blob_id', request, namespace='..')
156
157     @pytest.mark.usefixtures("method_tmpdir")
158     @defer.inlineCallbacks
159     def test_path_validation_on_write_blob(self):
160         blobs_path, request = self.tempdir, DummyRequest([''])
161         backend = _blobs.FilesystemBlobsBackend(blobs_path=blobs_path)
162         with pytest.raises(Exception):
163             yield backend.write_blob('..', '..', request)
164         with pytest.raises(Exception):
165             yield backend.write_blob('user', '../../../', request)
166         with pytest.raises(Exception):
167             yield backend.write_blob('../../../', 'id1', request)
168         with pytest.raises(Exception):
169             yield backend.write_blob('user', 'id2', request, namespace='..')
170
171     @pytest.mark.usefixtures("method_tmpdir")
172     @mock.patch('leap.soledad.server._blobs.os.unlink')
173     @defer.inlineCallbacks
174     def test_delete_blob(self, unlink_mock):
175         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
176         # write a blob...
177         path = backend._get_path('user', 'blob_id', '')
178         mkdir_p(os.path.split(path)[0])
179         with open(path, "w") as f:
180             f.write("bl0b")
181         # ...and delete it
182         yield backend.delete_blob('user', 'blob_id')
183         unlink_mock.assert_any_call(backend._get_path('user',
184                                                       'blob_id'))
185         unlink_mock.assert_any_call(backend._get_path('user',
186                                                       'blob_id') + '.flags')
187
188     @pytest.mark.usefixtures("method_tmpdir")
189     @mock.patch('leap.soledad.server._blobs.os.unlink')
190     @defer.inlineCallbacks
191     def test_delete_blob_custom_namespace(self, unlink_mock):
192         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
193         # write a blob...
194         path = backend._get_path('user', 'blob_id', 'trash')
195         mkdir_p(os.path.split(path)[0])
196         with open(path, "w") as f:
197             f.write("bl0b")
198         # ...and delete it
199         yield backend.delete_blob('user', 'blob_id', namespace='trash')
200         unlink_mock.assert_any_call(backend._get_path('user',
201                                                       'blob_id',
202                                                       'trash'))
203         unlink_mock.assert_any_call(backend._get_path('user',
204                                                       'blob_id',
205                                                       'trash') + '.flags')
206
207     @pytest.mark.usefixtures("method_tmpdir")
208     @defer.inlineCallbacks
209     def test_write_blob_using_namespace(self):
210         backend = _blobs.FilesystemBlobsBackend(blobs_path=self.tempdir)
211         request = DummyRequest([''])
212         request.content = BytesIO('content')
213         yield backend.write_blob('user', 'blob_id', request,
214                                  namespace='custom')
215         default = yield backend.list_blobs('user', request)
216         custom = yield backend.list_blobs('user', request, namespace='custom')
217         self.assertEquals([], json.loads(default))
218         self.assertEquals(['blob_id'], json.loads(custom))