"""Unit tests for S3 connector.""" from unittest.mock import Mock, patch import pytest from botocore.exceptions import ClientError from src.connectors.s3 import ( S3Connector, get_s3_client, get_s3_connector, ) from src.errors import TransientError class TestS3ConnectorInit: """Tests for S3Connector initialization.""" def test_init_stores_s3_client(self): """Test S3Connector stores S3 client reference.""" mock_client = Mock() connector = S3Connector(mock_client) assert connector.s3_client is mock_client class TestS3ConnectorObjectExists: """Tests for S3Connector.object_exists method.""" def test_object_exists_returns_true_when_file_exists(self): """Test object_exists returns True when S3 object exists.""" mock_client = Mock() mock_client.head_object.return_value = {'ContentLength': 1024} connector = S3Connector(mock_client) result = connector.object_exists('test-bucket', 'path/to/file.csv') assert result is True mock_client.head_object.assert_called_once_with( Bucket='test-bucket', Key='path/to/file.csv' ) def test_object_exists_returns_false_for_404(self): """Test object_exists returns False when object doesn't exist (404).""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( {'ResponseMetadata': {'HTTPStatusCode': 404}}, 'HeadObject', ) connector = S3Connector(mock_client) result = connector.object_exists('test-bucket', 'missing/file.csv') assert result is False def test_object_exists_raises_transient_for_429(self): """Test object_exists raises TransientError for 429 TooManyRequests.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': {'Code': 'TooManyRequests', 'Message': 'Slow down'}, 'ResponseMetadata': {'HTTPStatusCode': 429}, }, 'HeadObject', ) connector = S3Connector(mock_client) with pytest.raises(TransientError) as exc_info: connector.object_exists('test-bucket', 'file.csv') assert 'S3 service error' in str(exc_info.value) assert 'TooManyRequests' in str(exc_info.value) def test_object_exists_raises_transient_for_500(self): """Test object_exists raises TransientError for 500 InternalError.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': {'Code': 'InternalError', 'Message': 'Server error'}, 'ResponseMetadata': {'HTTPStatusCode': 500}, }, 'HeadObject', ) connector = S3Connector(mock_client) with pytest.raises(TransientError) as exc_info: connector.object_exists('test-bucket', 'file.csv') assert 'S3 service error' in str(exc_info.value) assert 'InternalError' in str(exc_info.value) def test_object_exists_raises_transient_for_503(self): """Test object_exists raises TransientError for 503 ServiceUnavailable.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': { 'Code': 'ServiceUnavailable', 'Message': 'Service unavailable', }, 'ResponseMetadata': {'HTTPStatusCode': 503}, }, 'HeadObject', ) connector = S3Connector(mock_client) with pytest.raises(TransientError) as exc_info: connector.object_exists('test-bucket', 'file.csv') assert 'S3 service error' in str(exc_info.value) def test_object_exists_propagates_non_transient_errors(self): """Test object_exists propagates non-transient errors (403, 400, etc).""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': {'Code': 'AccessDenied', 'Message': 'Access denied'}, 'ResponseMetadata': {'HTTPStatusCode': 403}, }, 'HeadObject', ) connector = S3Connector(mock_client) with pytest.raises(ClientError) as exc_info: connector.object_exists('test-bucket', 'file.csv') assert exc_info.value.response['Error']['Code'] == 'AccessDenied' def test_object_exists_handles_slow_down_error(self): """Test object_exists handles SlowDown error as transient.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': { 'Code': 'SlowDown', 'Message': 'Please reduce your request rate', }, 'ResponseMetadata': {'HTTPStatusCode': 503}, }, 'HeadObject', ) connector = S3Connector(mock_client) with pytest.raises(TransientError) as exc_info: connector.object_exists('test-bucket', 'file.csv') assert 'S3 service error' in str(exc_info.value) assert 'SlowDown' in str(exc_info.value) class TestGetS3Client: """Tests for get_s3_client function.""" @patch('src.connectors.s3.boto3.client') def test_get_s3_client_creates_client_with_default_config(self, mock_boto_client): """Test get_s3_client creates boto3 S3 client with default config.""" mock_client = Mock() mock_boto_client.return_value = mock_client result = get_s3_client() assert result is mock_client mock_boto_client.assert_called_once() call_args = mock_boto_client.call_args assert call_args[0][0] == 's3' # Verify config has signature_version config = call_args[1]['config'] assert config.signature_version == 's3v4' @patch('src.connectors.s3.boto3.client') def test_get_s3_client_accepts_custom_config(self, mock_boto_client): """Test get_s3_client accepts custom Config.""" from botocore.client import Config custom_config = Config(signature_version='s3', region_name='us-west-2') mock_client = Mock() mock_boto_client.return_value = mock_client result = get_s3_client(config=custom_config) assert result is mock_client mock_boto_client.assert_called_once_with('s3', config=custom_config) class TestGetS3Connector: """Tests for get_s3_connector function.""" def test_get_s3_connector_with_provided_client(self): """Test get_s3_connector uses provided S3 client.""" mock_client = Mock() connector = get_s3_connector(s3_client=mock_client) assert isinstance(connector, S3Connector) assert connector.s3_client is mock_client @patch('src.connectors.s3.get_s3_client') def test_get_s3_connector_creates_default_client(self, mock_get_client): """Test get_s3_connector creates default client when none provided.""" mock_client = Mock() mock_get_client.return_value = mock_client connector = get_s3_connector() assert isinstance(connector, S3Connector) assert connector.s3_client is mock_client mock_get_client.assert_called_once() @patch('src.connectors.s3.get_s3_client') def test_get_s3_connector_without_client_argument(self, mock_get_client): """Test get_s3_connector works when called without arguments.""" mock_client = Mock() mock_get_client.return_value = mock_client connector = get_s3_connector() assert connector.s3_client is mock_client class TestHandleTransientErrorsDecorator: """Tests for handle_transient_errors decorator.""" def test_decorator_allows_successful_execution(self): """Test decorator allows successful function execution.""" mock_client = Mock() mock_client.head_object.return_value = {'ContentLength': 100} connector = S3Connector(mock_client) result = connector.object_exists('bucket', 'key') assert result is True def test_decorator_converts_429_to_transient_error(self): """Test decorator converts 429 status to TransientError.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': {'Code': 'ThrottlingException', 'Message': 'Rate exceeded'}, 'ResponseMetadata': {'HTTPStatusCode': 429}, }, 'HeadObject', ) connector = S3Connector(mock_client) with pytest.raises(TransientError): connector.object_exists('bucket', 'key') def test_decorator_converts_500_to_transient_error(self): """Test decorator converts 500 status to TransientError.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': {'Code': 'InternalError', 'Message': 'Internal error'}, 'ResponseMetadata': {'HTTPStatusCode': 500}, }, 'HeadObject', ) connector = S3Connector(mock_client) with pytest.raises(TransientError): connector.object_exists('bucket', 'key') def test_decorator_converts_503_to_transient_error(self): """Test decorator converts 503 status to TransientError.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': {'Code': 'ServiceUnavailable', 'Message': 'Unavailable'}, 'ResponseMetadata': {'HTTPStatusCode': 503}, }, 'HeadObject', ) connector = S3Connector(mock_client) with pytest.raises(TransientError): connector.object_exists('bucket', 'key') def test_decorator_does_not_convert_403(self): """Test decorator doesn't convert 403 to TransientError.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': {'Code': 'AccessDenied', 'Message': 'Denied'}, 'ResponseMetadata': {'HTTPStatusCode': 403}, }, 'HeadObject', ) connector = S3Connector(mock_client) with pytest.raises(ClientError) as exc_info: connector.object_exists('bucket', 'key') assert not isinstance(exc_info.value, TransientError) class TestS3ConnectorEdgeCases: """Tests for S3Connector edge cases.""" def test_object_exists_with_empty_bucket(self): """Test object_exists with empty bucket name.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': {'Code': 'InvalidBucketName', 'Message': 'Invalid bucket'}, 'ResponseMetadata': {'HTTPStatusCode': 400}, }, 'HeadObject', ) connector = S3Connector(mock_client) with pytest.raises(ClientError) as exc_info: connector.object_exists('', 'some-key') assert exc_info.value.response['Error']['Code'] == 'InvalidBucketName' def test_object_exists_with_empty_key(self): """Test object_exists with empty key.""" mock_client = Mock() # S3 might still accept empty key, but it's an edge case mock_client.head_object.return_value = {'ContentLength': 0} connector = S3Connector(mock_client) result = connector.object_exists('bucket', '') assert result is True mock_client.head_object.assert_called_once_with(Bucket='bucket', Key='') def test_object_exists_400_bad_request(self): """Test object_exists with 400 Bad Request.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': {'Code': 'BadRequest', 'Message': 'Bad Request'}, 'ResponseMetadata': {'HTTPStatusCode': 400}, }, 'HeadObject', ) connector = S3Connector(mock_client) # Should propagate, not convert to TransientError with pytest.raises(ClientError) as exc_info: connector.object_exists('bucket', 'key') assert exc_info.value.response['Error']['Code'] == 'BadRequest' assert not isinstance(exc_info.value, TransientError) def test_object_exists_401_unauthorized(self): """Test object_exists with 401 Unauthorized.""" mock_client = Mock() mock_client.head_object.side_effect = ClientError( { 'Error': {'Code': 'Unauthorized', 'Message': 'Unauthorized'}, 'ResponseMetadata': {'HTTPStatusCode': 401}, }, 'HeadObject', ) connector = S3Connector(mock_client) # Should propagate, not convert to TransientError with pytest.raises(ClientError) as exc_info: connector.object_exists('bucket', 'key') assert exc_info.value.response['Error']['Code'] == 'Unauthorized' def test_object_exists_with_very_long_key(self): """Test object_exists with maximum length S3 key (1024 chars).""" mock_client = Mock() mock_client.head_object.return_value = {'ContentLength': 100} connector = S3Connector(mock_client) # S3 keys can be up to 1024 characters very_long_key = 'a' * 1024 result = connector.object_exists('bucket', very_long_key) assert result is True mock_client.head_object.assert_called_once_with( Bucket='bucket', Key=very_long_key )