ckanext-s3filestore/ckanext/s3filestore/tests/test_blueprints.py at c123ba06706d389d55107df10b0db41c443fc885 · keitaroinc/ckanext-s3filestore · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
# encoding: utf-8
import six
import requests

import pytest

from ckantoolkit import config
import ckan.tests.factories as factories
from ckan.lib.helpers import url_for
import re


@pytest.mark.usefixtures(u'clean_db', u'clean_index')
class TestS3Controller(object):

    @classmethod
    def setup_class(cls):
        cls.bucket_name = config.get(u'ckanext.s3filestore.aws_bucket_name')

    def test_resource_download_url(self, resource_with_upload):
        u'''The resource url is expected for uploaded resource file.'''

        expected_url = u'/dataset/{0}/' \
                       u'resource/{1}/download/test.csv'.\
            format(resource_with_upload[u'package_id'],
                   resource_with_upload[u'id'])

        assert re.sub(r'^https?://[^/]+', '', resource_with_upload['url']) == expected_url

    def test_resource_download(self, app, resource_with_upload):
        u'''When trying to download resource
        from CKAN it should redirect to S3.'''

        user = factories.Sysadmin()
        env = {u'REMOTE_USER': six.ensure_str(user[u'name'])}

        response = app.get(
            url_for(
                u'dataset_resource.download',
                id=resource_with_upload[u'package_id'],
                resource_id=resource_with_upload[u'id'],
            ),
            extra_environ=env,
            follow_redirects=False
        )
        assert 302 == response.status_code

    def test_resource_download_not_found(self, app):
        u'''When trying to download resource from
        CKAN it should redirect to S3.'''

        user = factories.Sysadmin()
        env = {u'REMOTE_USER': six.ensure_str(user[u'name'])}

        response = app.get(
            url_for(
                u'dataset_resource.download',
                id=u'package_id',
                resource_id=u'resource_id',
            ),
            extra_environ=env,
            follow_redirects=False
        )
        assert 404 == response.status_code

    def test_resource_download_no_filename(self,
                                           app, resource_with_upload):
        '''A resource uploaded to S3 can be
        downloaded when no filename in
        url.'''
        resource_file_url = u'/dataset/{0}/resource/{1}/download' \
            .format(resource_with_upload[u'package_id'],
                    resource_with_upload[u'id'])

        response = app.get(resource_file_url,
                           follow_redirects=False)

        assert 302 == response.status_code

    def test_s3_download_link(self, app, s3_client, resource_with_upload):
        u'''A resource download from s3 test.'''

        user = factories.Sysadmin()
        env = {u'REMOTE_USER': six.ensure_str(user[u'name'])}

        response = app.get(
            url_for(
                u'dataset_resource.download',
                id=resource_with_upload[u'package_id'],
                resource_id=resource_with_upload[u'id'],
            ),
            extra_environ=env,
            follow_redirects=False
        )
        assert 302 == response.status_code
        assert response.location
        downloaded_file = requests.get(response.location)
        assert u'SnowCourseName, Number, Elev. metres,' \
               in downloaded_file.text

    def test_s3_resource_mimetype(self, resource_with_upload):
        u'''A resource mimetype test.'''

        assert u'text/csv' == resource_with_upload[u'mimetype']

    def test_organization_image_redirects_to_s3(self,
                                                app,
                                                organization_with_image):
        url = u'/uploads/group/{0}'\
            .format(organization_with_image[u'image_url'])
        response = app.get(url,
                           follow_redirects=False)
        assert 302 == response.status_code

    def test_organization_image_download_from_s3(self,
                                                 app,
                                                 organization_with_image):
        url = u'/uploads/group/{0}'\
            .format(organization_with_image[u'image_url'])
        response = app.get(url,
                           follow_redirects=False)
        assert 302 == response.status_code
        assert response.location
        image = requests.get(response.location)
        assert image.content == b"\0\0\0"