source: OpenWorkouts-current/ow/tests/views/test_bulk.py @ 715671f

import os
from io import BytesIO
from cgi import FieldStorage

import pytest

from pyramid.testing import DummyRequest
from pyramid.httpexceptions import HTTPFound

from webob.multidict import MultiDict

from ow.models.root import OpenWorkouts
from ow.models.user import User
from ow.models.bulk import BulkFile, BulkFiles
from ow.schemas.bulk import BulkFileSchema
from ow.utilities import create_blob

import ow.views.bulk as bulk_views


class TestBulkViews(object):

    # paths to sample compressed files we can use for testing.
    bulk_filenames = (
        # valid FIT files, in different compressed formats
        'fixtures/bulk-fit.zip',
        'fixtures/bulk-fit.tgz',
        # valid GPX files, also in different formats
        'fixtures/bulk-gpx.zip',
        'fixtures/bulk-gpx.tgz',
        # empty ZIP file
        'fixtures/bulk-empty.zip',
        # ZIP file containing invalid files in different formats
        'fixtures/bulk-invalid.zip'
    )

    def open_uploaded_file(self, path):
        """
        Open the uploaded compressed file fixture from disk
        """
        uploaded_file_path = os.path.join(
            os.path.dirname(os.path.dirname(__file__)), path)
        uploaded_file = open(uploaded_file_path, 'rb')
        return uploaded_file

    def close_uploaded_file(self, uploaded_file):
        """
        Close the opened uploaded compressed file
        """
        uploaded_file.close()

    def create_filestorage(self, uploaded_file):
        """
        Create a FileStorage instance from an open compressed tracking file,
        suitable for testing file uploads later
        """
        storage = FieldStorage()
        storage.filename = os.path.basename(uploaded_file.name)
        storage.file = BytesIO(uploaded_file.read())
        storage.name = os.path.basename(uploaded_file.name)
        # This prevents FormEncode validator from thinking we are providing
        # more than one file for the upload, which crashes the tests
        storage.list = None
        return storage

    @pytest.fixture
    def root(self):
        root = OpenWorkouts()
        root['_bulk_files'] = BulkFiles()
        john = User(firstname='John', lastname='Doe',
                    email='john.doe@example.net')
        john.password = 's3cr3t'
        root.add_user(john)

        """
        uid = str(john.uid)
        bulk_file = BulkFile(uid=uid)
        bulk_file_path = os.path.join(
            os.path.dirname(os.path.dirname(__file__)),
            'fixtures/bulk-fit.zip')
        with open(bulk_file_path, 'rb') as _bulk_file:
            bulk_file.compressed_file = create_blob(
                _bulk_file.read(), file_extension='zip', binary=True)
        bulk_file.file_name = 'bulk-fit.zip'
        bulk_file.file_type = 'zip'
        root['_bulk_files'].add_bulk_file(bulk_file)
        """

        return root

    @pytest.fixture
    def dummy_request(self, root):
        request = DummyRequest()
        request.root = root
        return request

    def test_add_bulk_file_get(self, dummy_request):
        """
        Test the view that renders the "add bulk file" form
        """
        request = dummy_request
        user = request.root.users[0]
        assert len(request.root['_bulk_files']) == 0
        response = bulk_views.add_bulk_file(user, request)
        assert 'form' in response
        assert len(response['form'].form.errors) == 0
        assert isinstance(response['form'].form.schema, BulkFileSchema)
        # GET request, no bulk files have been saved
        assert len(request.root['_bulk_files']) == 0

    def test_add_bulk_file_post_invalid(self, dummy_request):
        """
        POST request to add a bulk file, without providing the required form
        data (actually, any bulk file at all).
        """
        request = dummy_request
        user = request.root.users[0]
        request.method = 'POST'
        request.POST = MultiDict({'submit': True})
        assert len(request.root['_bulk_files']) == 0
        response = bulk_views.add_bulk_file(user, request)
        assert 'form' in response
        # Only one required field in this case, the bulk file
        assert len(response['form'].form.errors) == 1
        # no bulk files have been saved
        assert len(request.root['_bulk_files']) == 0

    def test_add_bulk_file_post_invalid_bytes(self, dummy_request):
        """
        POST request to add a workout, without uploading a compressed file,
        which sends an empty bytes object (b'')
        """
        request = dummy_request
        user = request.root.users[0]
        request.method = 'POST'
        request.POST = MultiDict({
            'compressed_file': b'',
            'submit': True,
            })
        assert len(request.root['_bulk_files']) == 0
        response = bulk_views.add_bulk_file(user, request)
        assert 'form' in response
        # Only one required field in this case, the tracking file
        assert len(response['form'].form.errors) == 1
        # no bulk files have been saved
        assert len(request.root['_bulk_files']) == 0

    @pytest.mark.parametrize('filename', bulk_filenames)
    def test_add_workout_post_valid(self, filename, dummy_request):
        """
        POST request to add a bulk file
        """
        request = dummy_request
        uploaded_file = self.open_uploaded_file(filename)
        filestorage = self.create_filestorage(uploaded_file)
        user = request.root.users[0]
        request.method = 'POST'
        request.POST = MultiDict({
            'compressed_file': filestorage,
            'submit': True,
            })
        assert len(request.root['_bulk_files']) == 0
        response = bulk_views.add_bulk_file(user, request)
        assert isinstance(response, HTTPFound)
        assert response.location == request.resource_url(user, 'bulk-files')
        assert len(request.root['_bulk_files']) == 1
        bulk_file = request.root['_bulk_files'].values()[0]
        assert isinstance(bulk_file, BulkFile)
        assert bulk_file.uid == str(user.uid)
        assert bulk_file.file_name == os.path.basename(filename)
        assert bulk_file.workout_ids == []
        assert bulk_file.loaded_info == {}
        assert not bulk_file.loaded
        self.close_uploaded_file(uploaded_file)

    def test_bulk_files_get(self, dummy_request):
        """
        Test the view that renders the "list of all bulk files for a user" page
        """
        request = dummy_request
        user = request.root.users[0]
        # empty _bulk_files
        assert len(request.root['_bulk_files']) == 0
        response = bulk_views.bulk_files(user, request)
        assert response == {'bulk_files': []}
        # let's add one bulk file
        uid = str(user.uid)
        bulk_file = BulkFile(uid=uid)
        bulk_file_path = os.path.join(
            os.path.dirname(os.path.dirname(__file__)),
            'fixtures/bulk-fit.zip')
        with open(bulk_file_path, 'rb') as _bulk_file:
            bulk_file.compressed_file = create_blob(
                _bulk_file.read(), file_extension='zip', binary=True)
        bulk_file.file_name = 'bulk-fit.zip'
        bulk_file.file_type = 'zip'
        request.root['_bulk_files'].add_bulk_file(bulk_file)
        assert len(request.root['_bulk_files']) == 1
        response = bulk_views.bulk_files(user, request)
        assert response == {'bulk_files': [bulk_file]}
        # now, add another one for another (faked) user
        another_bulk_file = BulkFile(uid='faked-user-uid')
        another_bulk_file_path = os.path.join(
            os.path.dirname(os.path.dirname(__file__)),
            'fixtures/bulk-gpx.zip')
        with open(another_bulk_file_path, 'rb') as _bulk_file:
            bulk_file.compressed_file = create_blob(
                _bulk_file.read(), file_extension='zip', binary=True)
        another_bulk_file.file_name = 'bulk-gpx.zip'
        another_bulk_file.file_type = 'zip'
        request.root['_bulk_files'].add_bulk_file(another_bulk_file)
        # there are 2 bulk files, 1 per user, so asking for one user
        # returns only one
        assert len(request.root['_bulk_files']) == 2
        response = bulk_views.bulk_files(user, request)
        assert response == {'bulk_files': [bulk_file]}
        assert response['bulk_files'][0].uid == str(user.uid)