sibblegp / b2blaze

"""

Copyright George Sibble 2018

"""

import b2blaze.b2lib

from datetime import datetime

import pytest

from b2blaze.b2_exceptions import B2Exception, B2RequestError, B2FileNotFoundError

class TestB2(object):

    """ Tests for the b2blaze library """

    @classmethod

    def setup_class(cls):

        """

        :return: None

        """

        cls.b2 = b2blaze.b2lib.B2()

        timestamp = datetime.now().strftime('%Y-%m-%d-%H%M%S')

        cls.bucket_name = 'testbucket-' + timestamp

        print('test bucket: ', cls.bucket_name)

    # Helper methods

    def test_create_b2_instance(self):

        """Create a B2 instance """

        b2 = b2blaze.b2lib.B2()

    @classmethod

    def create_bucket(cls):

        return cls.b2.buckets.create(cls.bucket_name, security=cls.b2.buckets.public)

    @classmethod

    def getbucket(cls):

        return cls.b2.buckets.get(bucket_name=cls.bucket_name) or cls.create_bucket()

    @classmethod

    def upload_textfile(cls, contents="hello there", file_name='test/hello.txt'):

        """ Upload text file with name 'test/hello.txt' """

        contents=contents.encode('utf-8')     # These fail unless encoded to UTF8

        bucket = cls.getbucket()

        return bucket.files.upload(contents=contents, file_name=file_name)

    @classmethod

    def is_b2_file(cls, obj):

        """ hacky method for checking object class/type is B2File"""

        if 'B2File' in str(type(obj)):

            return True

        return False

    ##   Tests    ##

    @pytest.mark.bucket

    @pytest.mark.files

    @pytest.mark.versions

    def test_create_bucket(self):

        """ Create a bucket by name. """

        self.bucket = self.b2.buckets.create(self.bucket_name, security=self.b2.buckets.public)

        assert self.bucket

    @pytest.mark.bucket

    def test_get_bucket(self):

        """ Get a bucket by name """ 

        bucket = self.getbucket()

        assert bucket

    @pytest.mark.bucket

    def test_get_all_buckets(self):

        """ Get buckets. Number of buckets returned should be >1 """

        buckets = self.b2.buckets.all()

        assert len(buckets) > 1, "Number of buckets returned should be >1"

    @pytest.mark.bucket

    def test_get_nonexistent_bucket(self):

        """ Get a bucket which doesn't exist should return None """ 

        bucket = self.b2.buckets.get(bucket_name='this doesnt exist')

        assert not bucket

    @pytest.mark.files

    def test_create_file_and_retrieve_by_id(self):

        """ Create a file and retrieve by ID """

        bucket = self.getbucket()

        contents='Hello World!'.encode('utf-8')     # These fail unless encoded to UTF8

        file = bucket.files.upload(contents=contents, file_name='test/hello.txt')

        file2 = bucket.files.get(file_id=file.file_id)

        # It should be a B2File

        assert self.is_b2_file(file2), 'Should be a B2File object'

    @pytest.mark.files

    def test_direct_upload_file(self):

        """ Upload binary file """

        bucket = self.getbucket()

        binary_file = open('b2blaze/test_pic.jpg', 'rb')

        uploaded_file = bucket.files.upload(contents=binary_file, file_name='test_pic2.jpg')

        binary_file.close()

        assert self.is_b2_file(uploaded_file)

    @pytest.mark.files

    def test_get_all_files(self):

        """ Get all files from a bucket. Returned objects are B2Files """

        bucket = self.getbucket()

        files = bucket.files.all()

        print('test_get_files: all files: ', len(files))

        # check type

        assert self.is_b2_file(files[0]), 'Should be a B2File object'

    @pytest.mark.versions

    @pytest.mark.files

    def test_get_all_file_versions(self):

        """ Get all file versions from a bucket """

        bucket = self.getbucket()

        file = self.upload_textfile()

        files = bucket.files.all_file_versions()

        print('test_get_all_file_versions: all versions: ', len(files['file_versions']))

        assert len(files['file_versions']) > 0, 'File versions should exist'

    @pytest.mark.files

    def test_get_file_by_name(self):

        """ Get file by name """

        bucket = self.getbucket()

        file = self.upload_textfile()

        # check type

        assert self.is_b2_file(file), 'Should be a B2File object'

    @pytest.mark.files

    def test_get_file_by_id(self):

        """ Get file by id """

        bucket = self.getbucket()

        file = self.upload_textfile()

        # check type

        assert self.is_b2_file(file), 'Should be a B2File object'

    @pytest.mark.versions

    @pytest.mark.files

    def test_get_file_versions(self):

        """ Get all versions of a file via the file.get_versions method. 

            Returned data should be a list, and items should be of type B2File

        """

        bucket = self.getbucket()

        file = bucket.files.get(file_name='test/hello.txt')

        versions = file.get_versions()

        assert len(versions) > 0, 'File should have multiple versions'

        # check type

        assert self.is_b2_file(versions[0]), 'Should be a B2File object'

    @pytest.mark.versions

    @pytest.mark.files

    def test_bucket_get_file_versions_by_name(self):

        """ Get all versions of a file by name file_list.get_versions method. 

            Returned data should be a list, and items should be of type B2File

        """

        bucket = self.getbucket()

        versions = bucket.files.get_versions(file_name='test/hello.txt')

        assert len(versions) > 0, 'File should have multiple versions'

        assert self.is_b2_file(versions[0]), 'Should be a B2File object'

    @pytest.mark.versions

    @pytest.mark.files

    def test_bucket_get_file_versions_by_id(self):

        """ Get all versions of a file by id file_list.get_versions method. 

            Returned data should be a list, and items should be of type B2File

        """

        bucket = self.getbucket()

        file = bucket.files.get(file_name='test/hello.txt')

        versions = bucket.files.get_versions(file_id=file.file_id)

        assert len(versions) > 0, 'File should have multiple versions'

        assert self.is_b2_file(versions[0]), 'Should be a B2File object'

    @pytest.mark.files

    @pytest.mark.b2errors

    def test_get_file_doesnt_exist(self):

        """ Get file which doesn't exist should raise B2FileNotFoundError, get by ID should raise B2RequestError """

        bucket = self.getbucket()

        with pytest.raises(B2FileNotFoundError):

            file = bucket.files.get(file_name='nope.txt')

        with pytest.raises(B2RequestError):

            file2 = bucket.files.get(file_id='abcd')

    @pytest.mark.files

    def test_download_file(self):

        """ Get file by id """

        bucket = self.getbucket()

        file = self.upload_textfile()

        data = file.download()

        assert len(data.read()) > 0

    @pytest.mark.files

    def test_download_url(self):

        """ Download file url should be publicly GET accessible """

        import requests

        bucket = self.getbucket()

        file = self.upload_textfile()

        url = file.url

        downloaded_file = requests.get(url)

        if downloaded_file.status_code != 200:

            print(downloaded_file.json())

            raise ValueError

    @pytest.mark.files

    @pytest.mark.b2errors

    def test_hide_file(self): 

        """ Should create + upload, then hide / soft-delete a file by name.

            File should no longer be returned when searched by name in bucket.

        """

        bucket = self.getbucket()

        upload = self.upload_textfile(contents='Delete this', file_name='test/deleteme.txt')

        # Delete

        print('test_delete_file: upload.file_id', upload.file_id)

        print('test_delete_file: upload.file_name', upload.file_name)

        upload.hide()

        # Refresh bucket; getting the the file should fail

        with pytest.raises(B2FileNotFoundError):

            bucket = self.getbucket()

            file = bucket.files.get(file_name=upload.file_name)

            assert not file, 'Deleted file should not be in files list'

    @pytest.mark.files

    @pytest.mark.versions

    def test_delete_file_version(self): 

        """ Delete a file version by name. It should still exist when searched."""

        bucket = self.getbucket()

        # Upload file & delete

        file = self.upload_textfile()

        file2 = self.upload_textfile()

        # Update versions

        versions = file.get_versions()

        assert len(versions) > 1, 'File should should have multiple version'

        # Delete version

        print('test_delete_file_version: file_name', file.file_name)

        print('test_delete_file_version: file_id', file.file_id)

        file.delete()

        # Refresh bucket; getting the the file should fail

        file2 = bucket.files.get(file_name=file.file_name)

        assert file2, 'Deleted file version only, file should still exist'

        assert self.is_b2_file(file2), 'Should be a B2File object'

    # @pytest.mark.versions

    @pytest.mark.files

    def test_delete_all_file_versions(self): 

        """ Delete all versions of a file. It should be gone completely from bucket."""

        bucket = self.getbucket()

        # Create file, make sure we have multiple versions

        contents='Hello World!'.encode('utf-8')     # These fail unless encoded to UTF8

        upload = bucket.files.upload(contents=contents, file_name='test/hello.txt')

        # Get

        # versions = bucket.files.get_versions(file_name='test/hello.txt')

        file = bucket.files.get(file_name='test/hello.txt')

        versions = file.get_versions()

        assert len(versions) > 0, 'File should should have multiple version'

        # Delete

        print('test_delete_all_file_versions: file_name', file.file_name)

        print('test_delete_all_file_versions: file_id', file.file_id)

        file.delete_all_versions(confirm=True)

        # Refresh bucket; getting the the file should fail

        with pytest.raises(B2FileNotFoundError):

            bucket = self.getbucket()

            file2 = bucket.files.get(file_name=file.file_name)

            assert not file2, 'Deleted all file versions, file should not exist'

    @pytest.mark.bucket

    def test_delete_non_empty_bucket(self):

        """ Delete bucket should fail on bucket non-empty """

        bucket = self.getbucket()

        # Upload file

        self.upload_textfile()

        assert len(bucket.files.all()) > 0, "Bucket should still contain files" 

        # Should raise exception on non-empty without confirm

        with pytest.raises(B2RequestError):   

            bucket.delete()    # Try to delete without confirmation

        # Bucket should still exist

        assert self.b2.buckets.get(bucket_name=bucket.bucket_name), 'bucket should still exist'

        # # Delete with confirmation

        # bucket.delete(delete_files=True, confirm_non_empty=True)

        # # Bucket should be gone

        # assert self.b2.buckets.get(bucket_name=bucket.bucket_name), 'bucket should not exist'

    @pytest.mark.bucket

    @pytest.mark.files

    @pytest.mark.versions

    def test_bucket_delete_all_files(self):

        """ Delete all files from bucket. """

        bucket = self.getbucket()

        self.upload_textfile()

        files = bucket.files.all()

        assert len(files) > 0, 'Bucket should still contain files'

        # Delete all files

        bucket.files.delete_all(confirm=True)

        assert len(bucket.files.all()) == 0, 'Bucket should be empty'

    @pytest.mark.bucket

    def test_delete_bucket(self):

        """ Delete empty bucket"""

        bucket = self.getbucket()

        # Ascertain it's empty

        files_new = bucket.files.all(include_hidden=True)

        assert len(files_new) == 0, "Bucket should contain no files but contains {}".format(len(files_new))

        # Delete

        bucket.delete()

        # Confirm bucket is gone. bucket.get() nonexistent should return None.

        assert not self.b2.buckets.get(bucket_name=bucket.bucket_name), 'Deleted bucket still exists'

def main():

    import pytest

    pytest_args = [ __file__, '--verbose'] 

    pytest.main(pytest_args)

if __name__ == '__main__':

    main()