Skip to content

Refactor integration tests #210

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
agoncharov-reef wants to merge 4 commits into
base: replication_enhancement
Choose a base branch
from
+504 −416
Open

Refactor integration tests #210

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
66c478e
Refactor integration tests
agoncharov-reef May 6, 2022
cc87ad4
Yapf fix
agoncharov-reef May 6, 2022
4b7b73d
Replace small fixtures with constants
agoncharov-reef May 17, 2022
51dda82
FILE_NAME -> TEST_FILE_NAME, FILE_CONTENTS -> TEST_FILE_CONTENTS
agoncharov-reef May 18, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 9 additions & 28 deletions test/integration/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,40 +135,21 @@ def upload_url_dict(raw_api, api_url, account_auth_token, bucket_id) -> dict:
return raw_api.get_upload_url(api_url, account_auth_token, bucket_id)


@pytest.fixture(scope='module')
def file_name() -> str:
return 'test.txt'


@pytest.fixture(scope='module')
def file_contents() -> bytes:
return b'hello world'


@pytest.fixture(scope='session')
def sse_none() -> EncryptionSetting:
return EncryptionSetting(mode=EncryptionMode.NONE)


@pytest.fixture(scope='session')
def sse_b2_aes() -> EncryptionSetting:
return EncryptionSetting(
mode=EncryptionMode.SSE_B2,
algorithm=EncryptionAlgorithm.AES256,
)
TEST_FILE_NAME = 'test.txt'
TEST_FILE_CONTENTS = b'hello world'


@pytest.fixture(scope='module')
def file_dict(raw_api, upload_url_dict, file_name, file_contents, sse_b2_aes) -> dict:
def file_dict(raw_api, upload_url_dict, sse_b2_aes) -> dict:
return raw_api.upload_file(
upload_url_dict['uploadUrl'],
upload_url_dict['authorizationToken'],
file_name,
len(file_contents),
TEST_FILE_NAME,
len(TEST_FILE_CONTENTS),
'text/plain',
hex_sha1_of_stream(BytesIO(file_contents), len(file_contents)),
hex_sha1_of_stream(BytesIO(TEST_FILE_CONTENTS), len(TEST_FILE_CONTENTS)),
{'color': 'blue'},
BytesIO(file_contents),
BytesIO(TEST_FILE_CONTENTS),
server_side_encryption=sse_b2_aes,
)

Expand All @@ -179,12 +160,12 @@ def file_id(file_dict) -> str:


@pytest.fixture(scope='module')
def download_auth_dict(raw_api, api_url, account_auth_token, bucket_id, file_name) -> dict:
def download_auth_dict(raw_api, api_url, account_auth_token, bucket_id) -> dict:
return raw_api.get_download_authorization(
api_url,
account_auth_token,
bucket_id,
file_name[:-2],
TEST_FILE_NAME[:-2],
12345,
)

Expand Down
102 changes: 52 additions & 50 deletions test/integration/test_raw_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,13 +14,16 @@

import pytest

from b2sdk.encryption.setting import EncryptionMode, EncryptionSetting
from b2sdk.encryption.setting import SSE_B2_AES, SSE_NONE
from b2sdk.file_lock import BucketRetentionSetting, RetentionMode, RetentionPeriod
from b2sdk.replication.setting import ReplicationConfiguration, ReplicationDestinationConfiguration, ReplicationRule, ReplicationSourceConfiguration
from b2sdk.replication.types import ReplicationStatus
from b2sdk.utils import hex_sha1_of_stream

from .bucket_cleaner import _clean_and_delete_bucket, _cleanup_old_buckets
from .conftest import TEST_FILE_CONTENTS, TEST_FILE_NAME


"""
Try each of the calls to the raw api.

Expand Down Expand Up @@ -191,16 +194,15 @@ def test_replication(
account_auth_token,
source_bucket_dict['bucketId'],
)
file_contents = b'hello world'
file_dict = raw_api.upload_file(
upload_url_dict['uploadUrl'],
upload_url_dict['authorizationToken'],
'test.txt',
len(file_contents),
len(TEST_FILE_CONTENTS),
'text/plain',
hex_sha1_of_stream(io.BytesIO(file_contents), len(file_contents)),
hex_sha1_of_stream(io.BytesIO(TEST_FILE_CONTENTS), len(TEST_FILE_CONTENTS)),
{'color': 'blue'},
io.BytesIO(file_contents),
io.BytesIO(TEST_FILE_CONTENTS),
)

assert ReplicationStatus[file_dict['replicationStatus'].upper()
Expand Down Expand Up @@ -248,15 +250,15 @@ def test_replication(


def test_update_bucket(
raw_api, api_url, account_auth_token, account_id, bucket_id, sse_none, sse_b2_aes
raw_api, api_url, account_auth_token, account_id, bucket_id,
):
for encryption_setting, default_retention in [
(
sse_none,
SSE_NONE,
BucketRetentionSetting(mode=RetentionMode.GOVERNANCE, period=RetentionPeriod(days=1))
),
(sse_b2_aes, None),
(sse_b2_aes, BucketRetentionSetting(RetentionMode.NONE)),
(SSE_B2_AES, None),
(SSE_B2_AES, BucketRetentionSetting(RetentionMode.NONE)),
]:
raw_api.update_bucket(
api_url,
Expand All @@ -281,108 +283,108 @@ def test_file_upload(file_dict):
pass


def test_list_file_versions(raw_api, api_url, account_auth_token, bucket_id, file_dict, file_name):
def test_list_file_versions(raw_api, api_url, account_auth_token, bucket_id, file_dict):
list_versions_dict = raw_api.list_file_versions(api_url, account_auth_token, bucket_id)
assert [file_name] == [f_dict['fileName'] for f_dict in list_versions_dict['files']]
assert [TEST_FILE_NAME] == [f_dict['fileName'] for f_dict in list_versions_dict['files']]


def test_download_file_by_id_with_auth(
raw_api, download_url, file_dict, account_auth_token, file_contents
raw_api, download_url, file_dict, account_auth_token,
):
url = raw_api.get_download_url_by_id(download_url, file_dict['fileId'])
with raw_api.download_file_from_url(account_auth_token, url) as response:
data = next(response.iter_content(chunk_size=len(file_contents)))
assert data == file_contents, data
data = next(response.iter_content(chunk_size=len(TEST_FILE_CONTENTS)))
assert data == TEST_FILE_CONTENTS, data


def test_download_file_by_id_no_auth(raw_api, download_url, file_dict, file_contents):
def test_download_file_by_id_no_auth(raw_api, download_url, file_dict):

@ppolewicz ppolewicz May 18, 2022

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

file_dict is an unused variable here and in a few more tests. What I think you want to do is a class-based test which inherits from a FileHandlingHelper which uploads a file and assigns it to a self.file_dict in a setUp method

@agoncharov-reef agoncharov-reef May 18, 2022

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why so complicated? I need a file uploaded for the test. I require it in test inputs.

url = raw_api.get_download_url_by_id(download_url, file_dict['fileId'])
with raw_api.download_file_from_url(None, url) as response:
data = next(response.iter_content(chunk_size=len(file_contents)))
assert data == file_contents, data
data = next(response.iter_content(chunk_size=len(TEST_FILE_CONTENTS)))
assert data == TEST_FILE_CONTENTS, data


def test_download_file_by_name_with_auth(
raw_api, download_url, file_name, bucket_name, account_auth_token, file_contents
raw_api, download_url, bucket_name, account_auth_token, file_dict
):
url = raw_api.get_download_url_by_name(download_url, bucket_name, file_name)
url = raw_api.get_download_url_by_name(download_url, bucket_name, TEST_FILE_NAME)
with raw_api.download_file_from_url(account_auth_token, url) as response:
data = next(response.iter_content(chunk_size=len(file_contents)))
assert data == file_contents, data
data = next(response.iter_content(chunk_size=len(TEST_FILE_CONTENTS)))
assert data == TEST_FILE_CONTENTS, data


def test_download_file_by_name_no_auth(
raw_api, download_url, file_name, bucket_name, file_contents
raw_api, download_url, bucket_name, file_dict
):
url = raw_api.get_download_url_by_name(download_url, bucket_name, file_name)
url = raw_api.get_download_url_by_name(download_url, bucket_name, TEST_FILE_NAME)
with raw_api.download_file_from_url(None, url) as response:
data = next(response.iter_content(chunk_size=len(file_contents)))
assert data == file_contents, data
data = next(response.iter_content(chunk_size=len(TEST_FILE_CONTENTS)))
assert data == TEST_FILE_CONTENTS, data


def test_get_download_authorization(download_auth_dict):
pass


def test_download_file_by_name_download_auth(
download_auth_dict, raw_api, download_url, bucket_name, file_name, file_contents
download_auth_dict, raw_api, download_url, bucket_name, file_dict
):
download_auth_token = download_auth_dict['authorizationToken']
url = raw_api.get_download_url_by_name(download_url, bucket_name, file_name)
url = raw_api.get_download_url_by_name(download_url, bucket_name, TEST_FILE_NAME)
with raw_api.download_file_from_url(download_auth_token, url) as response:
data = next(response.iter_content(chunk_size=len(file_contents)))
assert data == file_contents, data
data = next(response.iter_content(chunk_size=len(TEST_FILE_CONTENTS)))
assert data == TEST_FILE_CONTENTS, data


def test_list_file_names(raw_api, api_url, account_auth_token, bucket_id, file_name):
def test_list_file_names(raw_api, api_url, account_auth_token, bucket_id, file_dict):
list_names_dict = raw_api.list_file_names(api_url, account_auth_token, bucket_id)
assert [file_name] == [f_dict['fileName'] for f_dict in list_names_dict['files']]
assert [TEST_FILE_NAME] == [f_dict['fileName'] for f_dict in list_names_dict['files']]


def test_list_file_names_start_count(raw_api, api_url, account_auth_token, bucket_id, file_name):
def test_list_file_names_start_count(raw_api, api_url, account_auth_token, bucket_id, file_dict):
list_names_dict = raw_api.list_file_names(
api_url, account_auth_token, bucket_id, start_file_name=file_name, max_file_count=5
api_url, account_auth_token, bucket_id, start_file_name=TEST_FILE_NAME, max_file_count=5
)
assert [file_name] == [f_dict['fileName'] for f_dict in list_names_dict['files']]
assert [TEST_FILE_NAME] == [f_dict['fileName'] for f_dict in list_names_dict['files']]


def test_copy_file(raw_api, api_url, account_auth_token, file_id):
copy_file_name = 'test_copy.txt'
raw_api.copy_file(api_url, account_auth_token, file_id, copy_file_name)


def test_get_file_info_by_id(raw_api, api_url, account_auth_token, file_id, file_name):
def test_get_file_info_by_id(raw_api, api_url, account_auth_token, file_id):
file_info_dict = raw_api.get_file_info_by_id(api_url, account_auth_token, file_id)
assert file_info_dict['fileName'] == file_name
assert file_info_dict['fileName'] == TEST_FILE_NAME


def test_get_file_info_by_name_no_auth(
raw_api, api_url, account_id, bucket_name, file_name, file_id, download_url
raw_api, api_url, account_id, bucket_name, file_id, download_url
):
info_headers = raw_api.get_file_info_by_name(download_url, None, bucket_name, file_name)
info_headers = raw_api.get_file_info_by_name(download_url, None, bucket_name, TEST_FILE_NAME)
assert info_headers['x-bz-file-id'] == file_id


def test_get_file_info_by_name_with_auth(
raw_api, download_url, account_auth_token, bucket_name, file_name, file_id
raw_api, download_url, account_auth_token, bucket_name, file_id
):
info_headers = raw_api.get_file_info_by_name(
download_url, account_auth_token, bucket_name, file_name
download_url, account_auth_token, bucket_name, TEST_FILE_NAME
)
assert info_headers['x-bz-file-id'] == file_id


def test_get_file_info_by_name_download_auth(
raw_api, download_url, download_auth_token, bucket_name, file_name, file_id
raw_api, download_url, download_auth_token, bucket_name, file_id
):
info_headers = raw_api.get_file_info_by_name(
download_url, download_auth_token, bucket_name, file_name
download_url, download_auth_token, bucket_name, TEST_FILE_NAME
)
assert info_headers['x-bz-file-id'] == file_id


def test_hide_file(raw_api, api_url, account_auth_token, bucket_id, file_name):
raw_api.hide_file(api_url, account_auth_token, bucket_id, file_name)
def test_hide_file(raw_api, api_url, account_auth_token, bucket_id, file_dict):
raw_api.hide_file(api_url, account_auth_token, bucket_id, TEST_FILE_NAME)


class TestLargeFile:
Expand All @@ -392,16 +394,16 @@ def file_info(self) -> dict:

@pytest.fixture(scope='class')
def large_file_id(
self, raw_api, api_url, account_auth_token, bucket_id, file_name, sse_b2_aes, file_info
self, raw_api, api_url, account_auth_token, bucket_id, file_info, file_dict
) -> str:
large_info = raw_api.start_large_file(
api_url,
account_auth_token,
bucket_id,
file_name,
TEST_FILE_NAME,
'text/plain',
file_info,
server_side_encryption=sse_b2_aes,
server_side_encryption=SSE_B2_AES,
)
return large_info['fileId']

Expand All @@ -414,7 +416,7 @@ def part_sha1(self, part_contents) -> bytes:
return hex_sha1_of_stream(io.BytesIO(part_contents), len(part_contents))

def test_upload_part(
self, raw_api, api_url, account_auth_token, large_file_id, file_contents, part_contents,
self, raw_api, api_url, account_auth_token, large_file_id, part_contents,
part_sha1
):
upload_part_dict = raw_api.get_upload_part_url(api_url, account_auth_token, large_file_id)
Expand All @@ -434,12 +436,12 @@ def test_list_parts(self, raw_api, api_url, account_auth_token, large_file_id):
assert [1, 2] == [part['partNumber'] for part in parts_response['parts']]

def test_list_unfinished_large_files(
self, raw_api, api_url, account_auth_token, bucket_id, file_name, file_info
self, raw_api, api_url, account_auth_token, bucket_id, file_info, file_dict
):
unfinished_list = raw_api.list_unfinished_large_files(
api_url, account_auth_token, bucket_id
)
assert [file_name] == [f_dict['fileName'] for f_dict in unfinished_list['files']]
assert [TEST_FILE_NAME] == [f_dict['fileName'] for f_dict in unfinished_list['files']]
assert file_info == unfinished_list['files'][0]['fileInfo']

def test_finish_large_file(
Expand Down