Mock S3 bucket in tests

example/download.yaml

@@ -12,7 +12,7 @@
 - url: https://drive.google.com/uc?id=10ojJffrPSl12OMcu4gyx0fak2CNu6qOs
   local_name: gdrive_test_2.txt
 
-- url: s3://monarch-kg-test/kghub_downloader_test_file.yaml
+- url: s3://monarch-test/kghub_downloader_test_file.yaml
   local_name: s3_test.yaml
 
 - url: git://Knowledge-Graph-Hub/kg-microbe/testfile.zip

test/conftest.py

@@ -0,0 +1,38 @@
+"""Fixtures for s3."""
+
+import os
+
+import boto3  # type: ignore
+import moto
+import pytest
+
+
+@pytest.fixture(scope="function")
+def mock_aws_credentials():
+    """Fixture to mock AWS Credentials for moto."""
+    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
+    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"  # noqa: S105
+    os.environ["AWS_SECURITY_TOKEN"] = "testing"  # noqa: S105
+    os.environ["AWS_SESSION_TOKEN"] = "testing"  # noqa: S105
+    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
+
+
+@pytest.fixture(scope="function")
+def mock_empty_bucket(mock_aws_credentials):
+    """Fixture to mock an empty AWS bucket."""
+    moto_fake = moto.mock_aws()
+    try:
+        moto_fake.start()
+        conn = boto3.resource("s3")
+        conn.create_bucket(Bucket="monarch-test")  # or the name of the bucket you use
+        yield conn
+    finally:
+        moto_fake.stop()
+
+@pytest.fixture(scope="function")
+def mock_s3_test_file(mock_empty_bucket):
+    """Fixture to populate the mock S3 bucket with a test file."""
+    s3 = boto3.client("s3")
+    s3.put_object(Body="test data", Bucket="monarch-test", Key="kghub_downloader_test_file.yaml")
+    yield
+    s3.delete_object(Bucket="monarch-test", Key="kghub_downloader_test_file.yaml")

test/integration/test_download.py

@@ -3,6 +3,8 @@
 from os.path import exists
 from pathlib import Path
 
+import pytest
+
 from kghub_downloader import download, model
 from kghub_downloader.download_utils import download_from_yaml
 
@@ -48,8 +50,9 @@ def test_google_drive(self):
         download.google_drive(resource, output_file, False)
         self._assert_file_exists(output_file)
 
+    @pytest.mark.usefixtures('mock_s3_test_file')
     def test_s3(self):
-        resource = model.DownloadableResource(url="s3://monarch-kg-test/kghub_downloader_test_file.yaml")
+        resource = model.DownloadableResource(url="s3://monarch-test/kghub_downloader_test_file.yaml")
         output_file = output_files["s3"]
         download.s3(resource, output_file, False)
         self._assert_file_exists(output_file)
@@ -60,6 +63,7 @@ def test_git(self):
         download.git(resource, output_file, False)
         self._assert_file_exists(output_file)
 
+    @pytest.mark.usefixtures('mock_s3_test_file')
     def test_yaml_spec_download(self):
         download_from_yaml(yaml_file="example/download.yaml", output_dir="test/output")
         for file in output_files.values():

test/unit/test_mirror.py

@@ -1,12 +1,9 @@
-import os
 from unittest import mock
 
-import boto3
-import moto
-import pytest
-
 from kghub_downloader.upload import mirror_to_bucket
 
+# ruff: noqa: D100, D103
+
 
 # proper test
 @mock.patch("google.cloud.storage.Client")
@@ -24,38 +21,15 @@ def test_mirror(client):
     blob.assert_called_with("kghub_test_upload.txt")
 
 
-@pytest.fixture(scope="session")
-def aws_credentials():
-    """Mocked AWS Credentials for moto."""
-    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
-    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
-    os.environ["AWS_SECURITY_TOKEN"] = "testing"
-    os.environ["AWS_SESSION_TOKEN"] = "testing"
-    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
-
-
-@pytest.fixture(scope="function")
-def empty_bucket(aws_credentials):
-    moto_fake = moto.mock_aws()
-    try:
-        moto_fake.start()
-        conn = boto3.resource("s3")
-        conn.create_bucket(Bucket="monarch-test")  # or the name of the bucket you use
-        yield conn
-    finally:
-        moto_fake.stop()
-
-
-def test_mirror_to_bucket_s3(empty_bucket):
-    # Call the function under test
+def test_mirror_to_bucket_s3(mock_empty_bucket):
     result = mirror_to_bucket(
         local_file="test/resources/testfile.txt",
         bucket_url="s3://monarch-test/",
         remote_file="kghub_test_upload.txt",
     )
 
     # Check if the file was created in the bucket
-    bucket = empty_bucket.Bucket("monarch-test")
+    bucket = mock_empty_bucket.Bucket("monarch-test")
     files_in_bucket = list(bucket.objects.all())
     assert len(files_in_bucket) == 1
     assert files_in_bucket[0].key == "kghub_test_upload.txt"