I'm trying to write a test that verifies that register_extracts_by_location is able to read from an s3 bucket and grab the files. When writing the moto mock test, I get an error stating that the bucket doesn't exist.

Here's the register_extracts_by_location method:

class ProcessTracker:
    # ... other methods and init here.

    def register_extracts_by_location(self, location_path, location_name=None):
        """
        For a given location, find all files and attempt to register them.
        :param location_name: Name of the location
        :param location_path: Path of the location
        :return:
        """
        location = LocationTracker(location_path=location_path, location_name=location_name)

        if location.location_type.location_type_name == "s3":
            s3 = boto3.resource("s3")

            path = location.location_path

            if path.startswith("s3://"):
                path = path[len("s3://")]

            bucket = s3.Bucket(path)

            for file in bucket.objects.all():
                ExtractTracker(process_run=self
                               , filename=file
                               , location=location
                               , status='ready')
        else:
            for file in os.listdir(location_path):
                ExtractTracker(process_run=self
                               , filename=file
                               , location=location
                               , status='ready')

and the relevant part of the test is here:

   def test_register_extracts_by_location_s3(self):
        """
        Testing that when the location is s3, all the extracts are registered and set to 'ready' status.
        The process/extract relationship should also be set to 'ready' since that is the last status the process set
        the extracts to.
        :return:
        """
        process_status = aliased(ExtractStatus)
        extract_status = aliased(ExtractStatus)

        expected_keys = 'test_local_dir_1.csv', 'test_local_dir_2.csv'

        with moto.mock_s3():
            conn = boto3.resource('s3', region_name='us-east-1')
            conn.create_bucket(Bucket='test_bucket')

            for file in expected_keys:
                conn.Object('test_bucket', file)

            self.process_tracker.register_extracts_by_location(location_path='s3://test_bucket')

It appears that boto3 is still going out and connecting, but I'm not sure at this point. The error received is:

botocore.errorfactory.NoSuchBucket: An error occurred (NoSuchBucket) when calling the ListObjects operation: The specified bucket does not exist

0 Answers