[715671f] | 1 | import os |
---|
| 2 | from io import BytesIO |
---|
| 3 | from cgi import FieldStorage |
---|
| 4 | |
---|
| 5 | import pytest |
---|
| 6 | |
---|
| 7 | from pyramid.testing import DummyRequest |
---|
| 8 | from pyramid.httpexceptions import HTTPFound |
---|
| 9 | |
---|
| 10 | from webob.multidict import MultiDict |
---|
| 11 | |
---|
| 12 | from ow.models.root import OpenWorkouts |
---|
| 13 | from ow.models.user import User |
---|
| 14 | from ow.models.bulk import BulkFile, BulkFiles |
---|
| 15 | from ow.schemas.bulk import BulkFileSchema |
---|
| 16 | from ow.utilities import create_blob |
---|
| 17 | |
---|
| 18 | import ow.views.bulk as bulk_views |
---|
| 19 | |
---|
| 20 | |
---|
| 21 | class TestBulkViews(object): |
---|
| 22 | |
---|
| 23 | # paths to sample compressed files we can use for testing. |
---|
| 24 | bulk_filenames = ( |
---|
| 25 | # valid FIT files, in different compressed formats |
---|
| 26 | 'fixtures/bulk-fit.zip', |
---|
| 27 | 'fixtures/bulk-fit.tgz', |
---|
| 28 | # valid GPX files, also in different formats |
---|
| 29 | 'fixtures/bulk-gpx.zip', |
---|
| 30 | 'fixtures/bulk-gpx.tgz', |
---|
| 31 | # empty ZIP file |
---|
| 32 | 'fixtures/bulk-empty.zip', |
---|
| 33 | # ZIP file containing invalid files in different formats |
---|
| 34 | 'fixtures/bulk-invalid.zip' |
---|
| 35 | ) |
---|
| 36 | |
---|
| 37 | def open_uploaded_file(self, path): |
---|
| 38 | """ |
---|
| 39 | Open the uploaded compressed file fixture from disk |
---|
| 40 | """ |
---|
| 41 | uploaded_file_path = os.path.join( |
---|
| 42 | os.path.dirname(os.path.dirname(__file__)), path) |
---|
| 43 | uploaded_file = open(uploaded_file_path, 'rb') |
---|
| 44 | return uploaded_file |
---|
| 45 | |
---|
| 46 | def close_uploaded_file(self, uploaded_file): |
---|
| 47 | """ |
---|
| 48 | Close the opened uploaded compressed file |
---|
| 49 | """ |
---|
| 50 | uploaded_file.close() |
---|
| 51 | |
---|
| 52 | def create_filestorage(self, uploaded_file): |
---|
| 53 | """ |
---|
| 54 | Create a FileStorage instance from an open compressed tracking file, |
---|
| 55 | suitable for testing file uploads later |
---|
| 56 | """ |
---|
| 57 | storage = FieldStorage() |
---|
| 58 | storage.filename = os.path.basename(uploaded_file.name) |
---|
| 59 | storage.file = BytesIO(uploaded_file.read()) |
---|
| 60 | storage.name = os.path.basename(uploaded_file.name) |
---|
| 61 | # This prevents FormEncode validator from thinking we are providing |
---|
| 62 | # more than one file for the upload, which crashes the tests |
---|
| 63 | storage.list = None |
---|
| 64 | return storage |
---|
| 65 | |
---|
| 66 | @pytest.fixture |
---|
| 67 | def root(self): |
---|
| 68 | root = OpenWorkouts() |
---|
| 69 | root['_bulk_files'] = BulkFiles() |
---|
| 70 | john = User(firstname='John', lastname='Doe', |
---|
| 71 | email='john.doe@example.net') |
---|
| 72 | john.password = 's3cr3t' |
---|
| 73 | root.add_user(john) |
---|
| 74 | |
---|
| 75 | """ |
---|
| 76 | uid = str(john.uid) |
---|
| 77 | bulk_file = BulkFile(uid=uid) |
---|
| 78 | bulk_file_path = os.path.join( |
---|
| 79 | os.path.dirname(os.path.dirname(__file__)), |
---|
| 80 | 'fixtures/bulk-fit.zip') |
---|
| 81 | with open(bulk_file_path, 'rb') as _bulk_file: |
---|
| 82 | bulk_file.compressed_file = create_blob( |
---|
| 83 | _bulk_file.read(), file_extension='zip', binary=True) |
---|
| 84 | bulk_file.file_name = 'bulk-fit.zip' |
---|
| 85 | bulk_file.file_type = 'zip' |
---|
| 86 | root['_bulk_files'].add_bulk_file(bulk_file) |
---|
| 87 | """ |
---|
| 88 | |
---|
| 89 | return root |
---|
| 90 | |
---|
| 91 | @pytest.fixture |
---|
| 92 | def dummy_request(self, root): |
---|
| 93 | request = DummyRequest() |
---|
| 94 | request.root = root |
---|
| 95 | return request |
---|
| 96 | |
---|
| 97 | def test_add_bulk_file_get(self, dummy_request): |
---|
| 98 | """ |
---|
| 99 | Test the view that renders the "add bulk file" form |
---|
| 100 | """ |
---|
| 101 | request = dummy_request |
---|
| 102 | user = request.root.users[0] |
---|
| 103 | assert len(request.root['_bulk_files']) == 0 |
---|
| 104 | response = bulk_views.add_bulk_file(user, request) |
---|
| 105 | assert 'form' in response |
---|
| 106 | assert len(response['form'].form.errors) == 0 |
---|
| 107 | assert isinstance(response['form'].form.schema, BulkFileSchema) |
---|
| 108 | # GET request, no bulk files have been saved |
---|
| 109 | assert len(request.root['_bulk_files']) == 0 |
---|
| 110 | |
---|
| 111 | def test_add_bulk_file_post_invalid(self, dummy_request): |
---|
| 112 | """ |
---|
| 113 | POST request to add a bulk file, without providing the required form |
---|
| 114 | data (actually, any bulk file at all). |
---|
| 115 | """ |
---|
| 116 | request = dummy_request |
---|
| 117 | user = request.root.users[0] |
---|
| 118 | request.method = 'POST' |
---|
| 119 | request.POST = MultiDict({'submit': True}) |
---|
| 120 | assert len(request.root['_bulk_files']) == 0 |
---|
| 121 | response = bulk_views.add_bulk_file(user, request) |
---|
| 122 | assert 'form' in response |
---|
| 123 | # Only one required field in this case, the bulk file |
---|
| 124 | assert len(response['form'].form.errors) == 1 |
---|
| 125 | # no bulk files have been saved |
---|
| 126 | assert len(request.root['_bulk_files']) == 0 |
---|
| 127 | |
---|
| 128 | def test_add_bulk_file_post_invalid_bytes(self, dummy_request): |
---|
| 129 | """ |
---|
| 130 | POST request to add a workout, without uploading a compressed file, |
---|
| 131 | which sends an empty bytes object (b'') |
---|
| 132 | """ |
---|
| 133 | request = dummy_request |
---|
| 134 | user = request.root.users[0] |
---|
| 135 | request.method = 'POST' |
---|
| 136 | request.POST = MultiDict({ |
---|
| 137 | 'compressed_file': b'', |
---|
| 138 | 'submit': True, |
---|
| 139 | }) |
---|
| 140 | assert len(request.root['_bulk_files']) == 0 |
---|
| 141 | response = bulk_views.add_bulk_file(user, request) |
---|
| 142 | assert 'form' in response |
---|
| 143 | # Only one required field in this case, the tracking file |
---|
| 144 | assert len(response['form'].form.errors) == 1 |
---|
| 145 | # no bulk files have been saved |
---|
| 146 | assert len(request.root['_bulk_files']) == 0 |
---|
| 147 | |
---|
| 148 | @pytest.mark.parametrize('filename', bulk_filenames) |
---|
| 149 | def test_add_workout_post_valid(self, filename, dummy_request): |
---|
| 150 | """ |
---|
| 151 | POST request to add a bulk file |
---|
| 152 | """ |
---|
| 153 | request = dummy_request |
---|
| 154 | uploaded_file = self.open_uploaded_file(filename) |
---|
| 155 | filestorage = self.create_filestorage(uploaded_file) |
---|
| 156 | user = request.root.users[0] |
---|
| 157 | request.method = 'POST' |
---|
| 158 | request.POST = MultiDict({ |
---|
| 159 | 'compressed_file': filestorage, |
---|
| 160 | 'submit': True, |
---|
| 161 | }) |
---|
| 162 | assert len(request.root['_bulk_files']) == 0 |
---|
| 163 | response = bulk_views.add_bulk_file(user, request) |
---|
| 164 | assert isinstance(response, HTTPFound) |
---|
| 165 | assert response.location == request.resource_url(user, 'bulk-files') |
---|
| 166 | assert len(request.root['_bulk_files']) == 1 |
---|
| 167 | bulk_file = request.root['_bulk_files'].values()[0] |
---|
| 168 | assert isinstance(bulk_file, BulkFile) |
---|
| 169 | assert bulk_file.uid == str(user.uid) |
---|
| 170 | assert bulk_file.file_name == os.path.basename(filename) |
---|
| 171 | assert bulk_file.workout_ids == [] |
---|
| 172 | assert bulk_file.loaded_info == {} |
---|
| 173 | assert not bulk_file.loaded |
---|
| 174 | self.close_uploaded_file(uploaded_file) |
---|
| 175 | |
---|
| 176 | def test_bulk_files_get(self, dummy_request): |
---|
| 177 | """ |
---|
| 178 | Test the view that renders the "list of all bulk files for a user" page |
---|
| 179 | """ |
---|
| 180 | request = dummy_request |
---|
| 181 | user = request.root.users[0] |
---|
| 182 | # empty _bulk_files |
---|
| 183 | assert len(request.root['_bulk_files']) == 0 |
---|
| 184 | response = bulk_views.bulk_files(user, request) |
---|
| 185 | assert response == {'bulk_files': []} |
---|
| 186 | # let's add one bulk file |
---|
| 187 | uid = str(user.uid) |
---|
| 188 | bulk_file = BulkFile(uid=uid) |
---|
| 189 | bulk_file_path = os.path.join( |
---|
| 190 | os.path.dirname(os.path.dirname(__file__)), |
---|
| 191 | 'fixtures/bulk-fit.zip') |
---|
| 192 | with open(bulk_file_path, 'rb') as _bulk_file: |
---|
| 193 | bulk_file.compressed_file = create_blob( |
---|
| 194 | _bulk_file.read(), file_extension='zip', binary=True) |
---|
| 195 | bulk_file.file_name = 'bulk-fit.zip' |
---|
| 196 | bulk_file.file_type = 'zip' |
---|
| 197 | request.root['_bulk_files'].add_bulk_file(bulk_file) |
---|
| 198 | assert len(request.root['_bulk_files']) == 1 |
---|
| 199 | response = bulk_views.bulk_files(user, request) |
---|
| 200 | assert response == {'bulk_files': [bulk_file]} |
---|
| 201 | # now, add another one for another (faked) user |
---|
| 202 | another_bulk_file = BulkFile(uid='faked-user-uid') |
---|
| 203 | another_bulk_file_path = os.path.join( |
---|
| 204 | os.path.dirname(os.path.dirname(__file__)), |
---|
| 205 | 'fixtures/bulk-gpx.zip') |
---|
| 206 | with open(another_bulk_file_path, 'rb') as _bulk_file: |
---|
| 207 | bulk_file.compressed_file = create_blob( |
---|
| 208 | _bulk_file.read(), file_extension='zip', binary=True) |
---|
| 209 | another_bulk_file.file_name = 'bulk-gpx.zip' |
---|
| 210 | another_bulk_file.file_type = 'zip' |
---|
| 211 | request.root['_bulk_files'].add_bulk_file(another_bulk_file) |
---|
| 212 | # there are 2 bulk files, 1 per user, so asking for one user |
---|
| 213 | # returns only one |
---|
| 214 | assert len(request.root['_bulk_files']) == 2 |
---|
| 215 | response = bulk_views.bulk_files(user, request) |
---|
| 216 | assert response == {'bulk_files': [bulk_file]} |
---|
| 217 | assert response['bulk_files'][0].uid == str(user.uid) |
---|
[8340661] | 218 | |
---|
| 219 | def test_delete_bulk_file_get(self, dummy_request): |
---|
| 220 | request = dummy_request |
---|
| 221 | user = request.root.users[0] |
---|
| 222 | # let's add one bulk file |
---|
| 223 | uid = str(user.uid) |
---|
| 224 | bulk_file = BulkFile(uid=uid) |
---|
| 225 | bulk_file_path = os.path.join( |
---|
| 226 | os.path.dirname(os.path.dirname(__file__)), |
---|
| 227 | 'fixtures/bulk-fit.zip') |
---|
| 228 | with open(bulk_file_path, 'rb') as _bulk_file: |
---|
| 229 | bulk_file.compressed_file = create_blob( |
---|
| 230 | _bulk_file.read(), file_extension='zip', binary=True) |
---|
| 231 | bulk_file.file_name = 'bulk-fit.zip' |
---|
| 232 | bulk_file.file_type = 'zip' |
---|
| 233 | request.root['_bulk_files'].add_bulk_file(bulk_file) |
---|
| 234 | assert len(request.root['_bulk_files']) == 1 |
---|
| 235 | response = bulk_views.delete_bulk_file(bulk_file, request) |
---|
| 236 | assert response == {'user': user} |
---|
| 237 | assert len(request.root['_bulk_files']) == 1 |
---|
| 238 | |
---|
| 239 | def test_delete_bulk_file_post_invalid(self, dummy_request): |
---|
| 240 | request = dummy_request |
---|
| 241 | request.method = 'POST' |
---|
| 242 | # invalid, missing confirmation delete hidden value |
---|
| 243 | request.POST = MultiDict({'submit': True}) |
---|
| 244 | user = request.root.users[0] |
---|
| 245 | # let's add one bulk file |
---|
| 246 | uid = str(user.uid) |
---|
| 247 | bulk_file = BulkFile(uid=uid) |
---|
| 248 | bulk_file_path = os.path.join( |
---|
| 249 | os.path.dirname(os.path.dirname(__file__)), |
---|
| 250 | 'fixtures/bulk-fit.zip') |
---|
| 251 | with open(bulk_file_path, 'rb') as _bulk_file: |
---|
| 252 | bulk_file.compressed_file = create_blob( |
---|
| 253 | _bulk_file.read(), file_extension='zip', binary=True) |
---|
| 254 | bulk_file.file_name = 'bulk-fit.zip' |
---|
| 255 | bulk_file.file_type = 'zip' |
---|
| 256 | request.root['_bulk_files'].add_bulk_file(bulk_file) |
---|
| 257 | assert len(request.root['_bulk_files']) == 1 |
---|
| 258 | response = bulk_views.delete_bulk_file(bulk_file, request) |
---|
| 259 | assert response == {'user': user} |
---|
| 260 | assert len(request.root['_bulk_files']) == 1 |
---|
| 261 | |
---|
| 262 | def test_delete_bulk_file_post_valid(self, dummy_request): |
---|
| 263 | request = dummy_request |
---|
| 264 | request.method = 'POST' |
---|
| 265 | # invalid, missing confirmation delete hidden value |
---|
| 266 | request.POST = MultiDict({'submit': True, 'delete': 'yes'}) |
---|
| 267 | user = request.root.users[0] |
---|
| 268 | # let's add one bulk file |
---|
| 269 | uid = str(user.uid) |
---|
| 270 | bulk_file = BulkFile(uid=uid) |
---|
| 271 | bulk_file_path = os.path.join( |
---|
| 272 | os.path.dirname(os.path.dirname(__file__)), |
---|
| 273 | 'fixtures/bulk-fit.zip') |
---|
| 274 | with open(bulk_file_path, 'rb') as _bulk_file: |
---|
| 275 | bulk_file.compressed_file = create_blob( |
---|
| 276 | _bulk_file.read(), file_extension='zip', binary=True) |
---|
| 277 | bulk_file.file_name = 'bulk-fit.zip' |
---|
| 278 | bulk_file.file_type = 'zip' |
---|
| 279 | request.root['_bulk_files'].add_bulk_file(bulk_file) |
---|
| 280 | assert len(request.root['_bulk_files']) == 1 |
---|
| 281 | response = bulk_views.delete_bulk_file(bulk_file, request) |
---|
| 282 | # after a successful delete, we send the user back to his dashboard |
---|
| 283 | assert isinstance(response, HTTPFound) |
---|
| 284 | assert response.location == request.resource_url(user, 'bulk-files') |
---|
| 285 | assert len(request.root['_bulk_files']) == 0 |
---|