import pytest from octopus.utils.aws import s3 @pytest.fixture(params=[ ('s3://bucket_name/', ('bucket_name', '')), ('s3://bucket_name/folder1/folder2/', ('bucket_name', 'folder1/folder2/')), ('s3://bucket_name/folder1/folder2/some_key', ('bucket_name', 'folder1/folder2/some_key')) ]) def s3_url_fixture(request): """Creates valid S3 urls as fixtures Return: tuple: param created by the fixture """ return request.param @pytest.fixture(params=['something_wrong', 'S3://', 'http://', '']) def s3_url_fixture_with_error(request): """Creates invalid S3 urls as fixtures Return: str: param created by the fixture """ return request.param def test_extract_bucket_path(s3_url_fixture): """Test that the bucket name and the bucket path are properly extracted from a given S3 url. """ (s3_url, expected) = s3_url_fixture assert s3.extract_bucket_path(s3_url) == expected def test_extract_bucket_path_with_errors(s3_url_fixture_with_error): """Test that an error is triggered when passing an invalid S3 url. """ with pytest.raises(Exception): s3.extract_bucket_path(s3_url_fixture_with_error) def test_get_destination_s3key_path(): with pytest.raises(AssertionError): s3.get_destination_s3key_path(None, None)