diff options
Diffstat (limited to 'tests/test_transform_lambda.py')
| -rw-r--r-- | tests/test_transform_lambda.py | 44 |
1 files changed, 26 insertions, 18 deletions
diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 00f3d83..5ed743e 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -1,4 +1,8 @@ -from src.transform_lambda import read_from_s3_subfolder_to_df, list_existing_s3_files, bucket_name +from src.transform_lambda import ( + read_from_s3_subfolder_to_df, + list_existing_s3_files, + bucket_name, +) from moto import mock_aws import pytest import pandas as pd @@ -6,14 +10,15 @@ import os import boto3 from botocore.exceptions import ClientError import numpy as np + # import caplog import logging - logger = logging.getLogger() logger.setLevel(logging.INFO) + @pytest.fixture(scope="class") def aws_credentials(): os.environ["AWS_ACCESS_KEY_ID"] = "testing" @@ -92,42 +97,45 @@ class TestReadFromS3: assert result["Foods"].eq(expected_foods_df, axis="columns").all(axis=None) assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None) + class TestListExistingFiles: def test_functions_receives_error_if_no_bucket(self, s3_client, caplog): caplog.set_level(logging.INFO) with pytest.raises(ClientError): - list_existing_s3_files('rando_bucket', client=s3_client) + list_existing_s3_files("rando_bucket", client=s3_client) - assert "Error listing S3 objects: An error occurred (NoSuchBucket) when calling the ListObjectsV2 operation: The specified bucket does not exist" in caplog.text + assert ( + "Error listing S3 objects: An error occurred (NoSuchBucket) when calling the ListObjectsV2 operation: The specified bucket does not exist" + in caplog.text + ) def test_recieves_logger_error_if_no_files_listed(self, s3_client, caplog): caplog.set_level(logging.INFO) s3_client.create_bucket( - Bucket='mock_bucket', - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"} + Bucket="mock_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, ) - response = list_existing_s3_files('mock_bucket', client=s3_client) - assert 'The bucket is empty' in caplog.text + response = list_existing_s3_files("mock_bucket", client=s3_client) + assert "The bucket is empty" in caplog.text def test_retrieves_existing_files(self, s3_client, caplog): caplog.set_level(logging.INFO) - s3_client.upload_file( - "tests/dummy.txt", 'mock_bucket', "dummy.txt" - ) - result = list_existing_s3_files('mock_bucket', client=s3_client) + s3_client.upload_file("tests/dummy.txt", "mock_bucket", "dummy.txt") + result = list_existing_s3_files("mock_bucket", client=s3_client) assert result == ["dummy.txt"] + class TestBucketName: def test_functions_retrieves_bucket(self, s3_client): s3_client.create_bucket( - Bucket='mock_bucket', - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"} + Bucket="mock_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, ) - - bucket = bucket_name('mock_bucket', s3_client) - assert bucket == 'mock_bucket' - # def test_
\ No newline at end of file + bucket = bucket_name("mock_bucket", s3_client) + assert bucket == "mock_bucket" + + # def test_ |