diff --git a/tests/unit/tools/test_aws_s3_tool.py b/tests/unit/tools/test_aws_s3_tool.py index 9c4c34e0b..8be8ebc5c 100644 --- a/tests/unit/tools/test_aws_s3_tool.py +++ b/tests/unit/tools/test_aws_s3_tool.py @@ -1,63 +1,52 @@ import boto3 import pytest +from moto import mock_aws from griptape.tools import AwsS3Tool -from tests.utils.aws import mock_aws_credentials class TestAwsS3Tool: - @pytest.fixture(autouse=True) - def _run_before_and_after_tests(self): - mock_aws_credentials() - - def test_get_bucket_acl(self): + @pytest.fixture() + def session(self): + mock = mock_aws() + mock.start() + yield boto3.Session(region_name="us-east-1") + mock.stop() + + def test_get_bucket_acl(self, session): value = {"bucket_name": "bucket_test"} - assert "error getting bucket acl" in AwsS3Tool(session=boto3.Session()).get_bucket_acl({"values": value}).value + assert "error getting bucket acl" in AwsS3Tool(session=session).get_bucket_acl({"values": value}).value - def test_get_bucket_policy(self): + def test_get_bucket_policy(self, session): value = {"bucket_name": "bucket_test"} - assert ( - "error getting bucket policy" - in AwsS3Tool(session=boto3.Session()).get_bucket_policy({"values": value}).value - ) + assert "error getting bucket policy" in AwsS3Tool(session=session).get_bucket_policy({"values": value}).value - def test_get_object_acl(self): + def test_get_object_acl(self, session): value = {"bucket_name": "bucket_test", "object_key": "key_test"} - assert "error getting object acl" in AwsS3Tool(session=boto3.Session()).get_object_acl({"values": value}).value + assert "error getting object acl" in AwsS3Tool(session=session).get_object_acl({"values": value}).value - def test_list_s3_buckets(self): - assert "error listing s3 buckets" in AwsS3Tool(session=boto3.Session()).list_s3_buckets({}).value + def test_list_s3_buckets(self, session): + assert AwsS3Tool(session=session).list_s3_buckets({}).value == [] - def test_list_objects(self): + def test_list_objects(self, session): value = {"bucket_name": "bucket_test"} - assert ( - "error listing objects in bucket" - in AwsS3Tool(session=boto3.Session()).list_objects({"values": value}).value - ) + assert "error listing objects in bucket" in AwsS3Tool(session=session).list_objects({"values": value}).value - def test_upload_memory_artifacts_to_s3(self): + def test_upload_memory_artifacts_to_s3(self, session): value = { "memory_name": "foobar", "bucket_name": "bucket_test", "artifact_namespace": "foo", "object_key": "test.txt", } - assert ( - "memory not found" - in AwsS3Tool(session=boto3.Session()).upload_memory_artifacts_to_s3({"values": value}).value - ) + assert "memory not found" in AwsS3Tool(session=session).upload_memory_artifacts_to_s3({"values": value}).value - def test_upload_content_to_s3(self): + def test_upload_content_to_s3(self, session): value = {"content": "foobar", "bucket_name": "bucket_test", "object_key": "test.txt"} - assert ( - "error uploading objects" - in AwsS3Tool(session=boto3.Session()).upload_content_to_s3({"values": value}).value - ) + assert "uploaded successfully" in AwsS3Tool(session=session).upload_content_to_s3({"values": value}).value - def test_download_objects(self): + def test_download_objects(self, session): value = {"objects": {"bucket_name": "bucket_test", "object_key": "test.txt"}} - assert ( - "error downloading objects" in AwsS3Tool(session=boto3.Session()).download_objects({"values": value}).value - ) + assert "error downloading objects" in AwsS3Tool(session=session).download_objects({"values": value}).value