<commit_before>import os
import uuid

import mock
import requests

from nose import SkipTest

from depot._compat import PY2, unicode_text


S3Storage = None
FILE_CONTENT = b'HELLO WORLD'


class TestS3FileStorage(object):
    def setup(self):
        try:
            global S3Storage
            from depot.io.awss3 import S3Storage
        except ImportError:
            raise SkipTest('Boto not installed')

        env = os.environ
        access_key_id = env.get('AWS_ACCESS_KEY_ID')
        secret_access_key = env.get('AWS_SECRET_ACCESS_KEY')
        if access_key_id is None or secret_access_key is None:
            raise SkipTest('Amazon S3 credentials not available')

        PID = os.getpid()
        NODE = str(uuid.uuid1()).rsplit('-', 1)[-1]  # Travis runs multiple tests concurrently
        self.default_bucket_name = 'filedepot-%s' % (access_key_id.lower(), )
        self.cred = (access_key_id, secret_access_key)
        self.fs = S3Storage(access_key_id, secret_access_key,
                            'filedepot-testfs-%s-%s-%s' % (access_key_id.lower(), NODE, PID))

    def test_fileoutside_depot(self):
        fid = str(uuid.uuid1())
        key = self.fs._bucket_driver.new_key(fid)
        key.set_contents_from_string(FILE_CONTENT)

        f = self.fs.get(fid)
        assert f.read() == FILE_CONTENT

    def test_invalid_modified(self):
        fid = str(uuid.uuid1())
        key = self.fs._bucket_driver.new_key(fid)
        key.set_metadata('x-depot-modified', 'INVALID')
        key.set_contents_from_string(FILE_CONTENT)

        f = self.fs.get(fid)
        assert f.last_modified is None, f.last_modified

    def test_creates_bucket_when_missing(self):
        with mock.patch('boto.s3.connection.S3Connection.lookup', return_value=None):
            with mock.patch('boto.s3.connection.S3Connection.lookup',
                            return_value='YES') as mock_create:
                fs = S3Storage(*self.cred)
                mock_create.assert_called_with(self.default_bucket_name)

    def test_default_bucket_name(self):
        with mock.patch('boto.s3.connection.S3Connection.lookup', return_value='YES'):
            fs = S3Storage(*self.cred)
            assert fs._bucket_driver.bucket == 'YES'

    def test_public_url(self):
        fid = str(uuid.uuid1())
        key = self.fs._bucket_driver.new_key(fid)
        key.set_contents_from_string(FILE_CONTENT)

        f = self.fs.get(fid)
        assert '.s3.amazonaws.com' in f.public_url, f.public_url
        assert f.public_url.endswith('/%s' % fid), f.public_url

    def test_content_disposition(self):
        if not PY2:
            raise SkipTest('Test is for Python2.X only')
        file_id = self.fs.create(b'content', unicode_text('test.txt'), 'text/plain')
        test_file = self.fs.get(file_id)
        response = requests.get(test_file.public_url)
        assert response.headers['Content-Disposition'] == "inline;filename=test.txt;filename*=utf-8''test.txt"

    def teardown(self):
        keys = [key.name for key in self.fs._bucket_driver.bucket]
        if keys:
            self.fs._bucket_driver.bucket.delete_keys(keys)

        try:
            self.fs._conn.delete_bucket(self.fs._bucket_driver.bucket.name)
        except:
            pass<commit_msg>This works also on Py3<commit_after>import os
import uuid

import mock
import requests

from nose import SkipTest

from depot._compat import PY2, unicode_text


S3Storage = None
FILE_CONTENT = b'HELLO WORLD'


class TestS3FileStorage(object):
    def setup(self):
        try:
            global S3Storage
            from depot.io.awss3 import S3Storage
        except ImportError:
            raise SkipTest('Boto not installed')

        env = os.environ
        access_key_id = env.get('AWS_ACCESS_KEY_ID')
        secret_access_key = env.get('AWS_SECRET_ACCESS_KEY')
        if access_key_id is None or secret_access_key is None:
            raise SkipTest('Amazon S3 credentials not available')

        PID = os.getpid()
        NODE = str(uuid.uuid1()).rsplit('-', 1)[-1]  # Travis runs multiple tests concurrently
        self.default_bucket_name = 'filedepot-%s' % (access_key_id.lower(), )
        self.cred = (access_key_id, secret_access_key)
        self.fs = S3Storage(access_key_id, secret_access_key,
                            'filedepot-testfs-%s-%s-%s' % (access_key_id.lower(), NODE, PID))

    def test_fileoutside_depot(self):
        fid = str(uuid.uuid1())
        key = self.fs._bucket_driver.new_key(fid)
        key.set_contents_from_string(FILE_CONTENT)

        f = self.fs.get(fid)
        assert f.read() == FILE_CONTENT

    def test_invalid_modified(self):
        fid = str(uuid.uuid1())
        key = self.fs._bucket_driver.new_key(fid)
        key.set_metadata('x-depot-modified', 'INVALID')
        key.set_contents_from_string(FILE_CONTENT)

        f = self.fs.get(fid)
        assert f.last_modified is None, f.last_modified

    def test_creates_bucket_when_missing(self):
        with mock.patch('boto.s3.connection.S3Connection.lookup', return_value=None):
            with mock.patch('boto.s3.connection.S3Connection.lookup',
                            return_value='YES') as mock_create:
                fs = S3Storage(*self.cred)
                mock_create.assert_called_with(self.default_bucket_name)

    def test_default_bucket_name(self):
        with mock.patch('boto.s3.connection.S3Connection.lookup', return_value='YES'):
            fs = S3Storage(*self.cred)
            assert fs._bucket_driver.bucket == 'YES'

    def test_public_url(self):
        fid = str(uuid.uuid1())
        key = self.fs._bucket_driver.new_key(fid)
        key.set_contents_from_string(FILE_CONTENT)

        f = self.fs.get(fid)
        assert '.s3.amazonaws.com' in f.public_url, f.public_url
        assert f.public_url.endswith('/%s' % fid), f.public_url

    def test_content_disposition(self):
        file_id = self.fs.create(b'content', unicode_text('test.txt'), 'text/plain')
        test_file = self.fs.get(file_id)
        response = requests.get(test_file.public_url)
        assert response.headers['Content-Disposition'] == "inline;filename=test.txt;filename*=utf-8''test.txt"

    def teardown(self):
        keys = [key.name for key in self.fs._bucket_driver.bucket]
        if keys:
            self.fs._bucket_driver.bucket.delete_keys(keys)

        try:
            self.fs._conn.delete_bucket(self.fs._bucket_driver.bucket.name)
        except:
            pass
{'fraction_non_alphanumeric': 0.10974485090685521, 'fraction_numerical': 0.00783891792191823, 'mean_word_length': 2.6170094496942746, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1434795', 'n_tokens_mistral': 2141, 'n_tokens_neox': 2084, 'n_words': 370}