Add s3 location tests including video

Unshure · Unshure · commit 2e47315be937 · 2026-01-27T12:14:36.000-05:00
diff --git a/src/strands/models/bedrock.py b/src/strands/models/bedrock.py
@@ -494,10 +494,10 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An
                 source = document["source"]
                 if "s3Location" in source:
                     s3_loc = source["s3Location"]
-                    formatted_s3: dict[str, Any] = {"uri": s3_loc["uri"]}
+                    formatted_document_s3: dict[str, Any] = {"uri": s3_loc["uri"]}
                     if "bucketOwner" in s3_loc:
-                        formatted_s3["bucketOwner"] = s3_loc["bucketOwner"]
-                    result["source"] = {"s3Location": formatted_s3}
+                        formatted_document_s3["bucketOwner"] = s3_loc["bucketOwner"]
+                    result["source"] = {"s3Location": formatted_document_s3}
                 elif "bytes" in source:
                     result["source"] = {"bytes": source["bytes"]}
 
@@ -520,16 +520,16 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An
         if "image" in content:
             image = content["image"]
             source = image["source"]
-            formatted_source: dict[str, Any] = {}
+            formatted_image_source: dict[str, Any] = {}
             if "s3Location" in source:
                 s3_loc = source["s3Location"]
-                formatted_s3: dict[str, Any] = {"uri": s3_loc["uri"]}
+                formatted_image_s3: dict[str, Any] = {"uri": s3_loc["uri"]}
                 if "bucketOwner" in s3_loc:
-                    formatted_s3["bucketOwner"] = s3_loc["bucketOwner"]
-                formatted_source = {"s3Location": formatted_s3}
+                    formatted_image_s3["bucketOwner"] = s3_loc["bucketOwner"]
+                formatted_image_source = {"s3Location": formatted_image_s3}
             elif "bytes" in source:
-                formatted_source = {"bytes": source["bytes"]}
-            result = {"format": image["format"], "source": formatted_source}
+                formatted_image_source = {"bytes": source["bytes"]}
+            result = {"format": image["format"], "source": formatted_image_source}
             return {"image": result}
 
         # https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ReasoningContentBlock.html
@@ -591,16 +591,16 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An
         if "video" in content:
             video = content["video"]
             source = video["source"]
-            formatted_source: dict[str, Any] = {}
+            formatted_video_source: dict[str, Any] = {}
             if "s3Location" in source:
                 s3_loc = source["s3Location"]
-                formatted_s3: dict[str, Any] = {"uri": s3_loc["uri"]}
+                formatted_video_s3: dict[str, Any] = {"uri": s3_loc["uri"]}
                 if "bucketOwner" in s3_loc:
-                    formatted_s3["bucketOwner"] = s3_loc["bucketOwner"]
-                formatted_source = {"s3Location": formatted_s3}
+                    formatted_video_s3["bucketOwner"] = s3_loc["bucketOwner"]
+                formatted_video_source = {"s3Location": formatted_video_s3}
             elif "bytes" in source:
-                formatted_source = {"bytes": source["bytes"]}
-            result = {"format": video["format"], "source": formatted_source}
+                formatted_video_source = {"bytes": source["bytes"]}
+            result = {"format": video["format"], "source": formatted_video_source}
             return {"video": result}
 
         # https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_CitationsContentBlock.html
diff --git a/src/strands/models/gemini.py b/src/strands/models/gemini.py
@@ -242,13 +242,12 @@ def _format_request_content(self, messages: Messages) -> list[genai.types.Conten
                     continue
                 parts.append(self._format_request_content_part(content, tool_use_id_to_name))
 
-            if parts:
-                contents.append(
-                    genai.types.Content(
-                        parts=parts,
-                        role="user" if message["role"] == "user" else "model",
-                    )
+            contents.append(
+                genai.types.Content(
+                    parts=parts,
+                    role="user" if message["role"] == "user" else "model",
                 )
+            )
 
         return contents
 
diff --git a/tests/strands/models/test__validation.py b/tests/strands/models/test__validation.py
@@ -1,7 +1,5 @@
 """Tests for model validation helper functions."""
 
-import pytest
-
 from strands.models._validation import _has_s3_source
 
 
diff --git a/tests/strands/tools/mcp/test_mcp_client.py b/tests/strands/tools/mcp/test_mcp_client.py
@@ -632,7 +632,7 @@ def test_call_tool_sync_embedded_nested_base64_textual_mime(mock_transport, mock
 def test_call_tool_sync_embedded_image_blob(mock_transport, mock_session):
     """EmbeddedResource.resource (blob with image MIME) should map to image content."""
     # Read yellow.png file
-    with open("tests_integ/yellow.png", "rb") as image_file:
+    with open("tests_integ/resources/yellow.png", "rb") as image_file:
         png_data = image_file.read()
     payload = base64.b64encode(png_data).decode()
 
diff --git a/tests/strands/types/test_media.py b/tests/strands/types/test_media.py
@@ -1,12 +1,9 @@
 """Tests for media type definitions."""
 
 from strands.types.media import (
-    DocumentFormat,
     DocumentSource,
-    ImageFormat,
     ImageSource,
     S3Location,
-    VideoFormat,
     VideoSource,
 )
 
diff --git a/tests_integ/conftest.py b/tests_integ/conftest.py
@@ -133,14 +133,21 @@ def pytest_sessionstart(session):
 
 @pytest.fixture
 def yellow_img(pytestconfig):
-    path = pytestconfig.rootdir / "tests_integ/yellow.png"
+    path = pytestconfig.rootdir / "tests_integ/resources/yellow.png"
     with open(path, "rb") as fp:
         return fp.read()
 
 
 @pytest.fixture
 def letter_pdf(pytestconfig):
-    path = pytestconfig.rootdir / "tests_integ/letter.pdf"
+    path = pytestconfig.rootdir / "tests_integ/resources/letter.pdf"
+    with open(path, "rb") as fp:
+        return fp.read()
+
+
+@pytest.fixture
+def blue_video(pytestconfig):
+    path = pytestconfig.rootdir / "tests_integ/resources/blue.mp4"
     with open(path, "rb") as fp:
         return fp.read()
 
diff --git a/tests_integ/mcp/echo_server.py b/tests_integ/mcp/echo_server.py
@@ -90,7 +90,7 @@ def get_weather(location: Literal["New York", "London", "Tokyo"] = "New York"):
             ]
         elif location.lower() == "tokyo":
             # Read yellow.png file for weather icon
-            with open("tests_integ/yellow.png", "rb") as image_file:
+            with open("tests_integ/resources/yellow.png", "rb") as image_file:
                 png_data = image_file.read()
             return [
                 EmbeddedResource(
diff --git a/tests_integ/mcp/test_mcp_client.py b/tests_integ/mcp/test_mcp_client.py
@@ -43,7 +43,7 @@ def calculator(x: int, y: int) -> int:
     @mcp.tool(description="Generates a custom image")
     def generate_custom_image() -> MCPImageContent:
         try:
-            with open("tests_integ/yellow.png", "rb") as image_file:
+            with open("tests_integ/resources/yellow.png", "rb") as image_file:
                 encoded_image = base64.b64encode(image_file.read())
                 return MCPImageContent(type="image", data=encoded_image, mimeType="image/png")
         except Exception as e:
diff --git a/tests_integ/resources/blue.mp4 b/tests_integ/resources/blue.mp4
diff --git a/tests_integ/resources/letter.pdf b/tests_integ/resources/letter.pdf
diff --git a/tests_integ/resources/yellow.png b/tests_integ/resources/yellow.png
diff --git a/tests_integ/test_a2a_executor.py b/tests_integ/test_a2a_executor.py
@@ -17,7 +17,7 @@
 async def test_a2a_executor_with_real_image():
     """Test A2A server processes a real image file correctly via HTTP."""
     # Read the test image file
-    test_image_path = os.path.join(os.path.dirname(__file__), "yellow.png")
+    test_image_path = os.path.join(os.path.dirname(__file__), "resources/yellow.png")
     with open(test_image_path, "rb") as f:
         original_image_bytes = f.read()
 
@@ -80,7 +80,7 @@ async def test_a2a_executor_with_real_image():
 def test_a2a_executor_image_roundtrip():
     """Test that image data survives the A2A base64 encoding/decoding roundtrip."""
     # Read the test image
-    test_image_path = os.path.join(os.path.dirname(__file__), "yellow.png")
+    test_image_path = os.path.join(os.path.dirname(__file__), "resources/yellow.png")
     with open(test_image_path, "rb") as f:
         original_bytes = f.read()
 
diff --git a/tests_integ/test_bedrock_s3_location.py b/tests_integ/test_bedrock_s3_location.py
@@ -5,35 +5,35 @@
 import boto3
 import pytest
 
-from strands.models.bedrock import BedrockModel
+from strands import Agent
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture
 def boto_session():
     """Create a boto3 session for testing."""
-    return boto3.Session(region_name="us-east-1")
+    return boto3.Session(region_name="us-west-2")
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture
 def account_id(boto_session):
     """Get the current AWS account ID."""
     sts_client = boto_session.client("sts")
     return sts_client.get_caller_identity()["Account"]
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture
 def s3_client(boto_session):
     """Create an S3 client."""
     return boto_session.client("s3")
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture
 def test_bucket(s3_client, account_id):
     """Create a test S3 bucket for the tests.
 
     Creates a bucket with account-specific name and cleans it up after tests.
     """
-    bucket_name = f"strands-integ-test-resources-{account_id}"
+    bucket_name = f"strands-integ-tests-resources-{account_id}"
 
     # Create the bucket if it doesn't exist
     try:
@@ -43,7 +43,7 @@ def test_bucket(s3_client, account_id):
         try:
             s3_client.create_bucket(
                 Bucket=bucket_name,
-                CreateBucketConfiguration={"LocationConstraint": "us-east-1"},
+                CreateBucketConfiguration={"LocationConstraint": "us-west-2"},
             )
             print(f"Created test bucket: {bucket_name}")
             # Wait for bucket to be available
@@ -57,7 +57,7 @@ def test_bucket(s3_client, account_id):
     # Objects will be overwritten on subsequent runs
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture
 def s3_document(s3_client, test_bucket, letter_pdf):
     """Upload a test document to S3 and return its URI."""
     document_key = "test-documents/letter.pdf"
@@ -74,7 +74,7 @@ def s3_document(s3_client, test_bucket, letter_pdf):
     return f"s3://{test_bucket}/{document_key}"
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture
 def s3_image(s3_client, test_bucket, yellow_img):
     """Upload a test image to S3 and return its URI."""
     image_key = "test-images/yellow.png"
@@ -91,112 +91,83 @@ def s3_image(s3_client, test_bucket, yellow_img):
     return f"s3://{test_bucket}/{image_key}"
 
 
-@pytest.fixture(scope="module")
-def bedrock_model():
-    """Create a BedrockModel for testing."""
-    return BedrockModel(region_name="us-east-1")
+@pytest.fixture
+def s3_video(s3_client, test_bucket, blue_video):
+    """Upload a test video to S3 and return its URI."""
+    video_key = "test-videos/blue.mp4"
 
+    # Upload the video using existing blue_video fixture
+    s3_client.put_object(
+        Bucket=test_bucket,
+        Key=video_key,
+        Body=blue_video,
+        ContentType="video/mp4",
+    )
+    print(f"Uploaded test video to s3://{test_bucket}/{video_key}")
 
-@pytest.mark.integ
-def test_document_s3_location_format(bedrock_model, s3_document):
-    """Test that Bedrock correctly formats a document with S3 location.
+    return f"s3://{test_bucket}/{video_key}"
 
-    This test uploads a document to S3, then verifies the s3Location reference
-    is correctly formatted in the Bedrock request.
-    """
+
+def test_document_s3_location(s3_document, account_id):
+    """Test that Bedrock correctly formats a document with S3 location."""
     messages = [
         {
             "role": "user",
             "content": [
-                {"text": "Please summarize this document:"},
+                {"text": "Please tell me about this document?"},
                 {
                     "document": {
                         "format": "pdf",
-                        "name": "letter.pdf",
-                        "source": {
-                            "s3Location": {
-                                "uri": s3_document,
-                            }
-                        },
-                    }
+                        "name": "letter",
+                        "source": {"s3Location": {"uri": s3_document, "bucketOwner": account_id}},
+                    },
                 },
             ],
         },
     ]
 
-    # Format the request to verify it's correctly structured
-    formatted_request = bedrock_model._format_request(messages)
+    agent = Agent(model="amazon.nova-2-lite-v1:0")
+    result = agent(messages)
 
-    # Verify the S3 location is properly formatted
-    document_content = formatted_request["messages"][0]["content"][1]
-    assert "document" in document_content
-    assert "source" in document_content["document"]
-    assert "s3Location" in document_content["document"]["source"]
-    assert document_content["document"]["source"]["s3Location"]["uri"] == s3_document
+    assert "amazon" in str(result).lower()
 
 
-@pytest.mark.integ
-def test_image_s3_location_format(bedrock_model, s3_image):
+def test_image_s3_location(s3_image):
     """Test that Bedrock correctly formats an image with S3 location."""
     messages = [
         {
             "role": "user",
             "content": [
-                {"text": "What do you see in this image?"},
+                {"text": "Please tell me about this image?"},
                 {
                     "image": {
                         "format": "png",
-                        "source": {
-                            "s3Location": {
-                                "uri": s3_image,
-                            }
-                        },
-                    }
+                        "source": {"s3Location": {"uri": s3_image}},
+                    },
                 },
             ],
         },
     ]
 
-    # Format the request to verify it's correctly structured
-    formatted_request = bedrock_model._format_request(messages)
+    agent = Agent(model="amazon.nova-2-lite-v1:0")
+    result = agent(messages)
 
-    # Verify the S3 location is properly formatted
-    image_content = formatted_request["messages"][0]["content"][1]
-    assert "image" in image_content
-    assert "source" in image_content["image"]
-    assert "s3Location" in image_content["image"]["source"]
-    assert image_content["image"]["source"]["s3Location"]["uri"] == s3_image
+    assert "yellow" in str(result).lower()
 
 
-@pytest.mark.integ
-def test_s3_location_with_bucket_owner(bedrock_model, s3_document, account_id):
-    """Test S3 location with optional bucketOwner field."""
+def test_video_s3_location(s3_video):
+    """Test that Bedrock correctly formats a video with S3 location."""
     messages = [
         {
             "role": "user",
             "content": [
-                {"text": "Read this document:"},
-                {
-                    "document": {
-                        "format": "pdf",
-                        "name": "letter.pdf",
-                        "source": {
-                            "s3Location": {
-                                "uri": s3_document,
-                                "bucketOwner": account_id,
-                            }
-                        },
-                    }
-                },
+                {"text": "Describe the colors is in this video?"},
+                {"video": {"format": "mp4", "source": {"s3Location": {"uri": s3_video}}}},
             ],
         },
     ]
 
-    # Format the request to verify bucketOwner is included
-    formatted_request = bedrock_model._format_request(messages)
-
-    document_content = formatted_request["messages"][0]["content"][1]
-    s3_location = document_content["document"]["source"]["s3Location"]
+    agent = Agent(model="us.amazon.nova-pro-v1:0")
+    result = agent(messages)
 
-    assert s3_location["uri"] == s3_document
-    assert s3_location["bucketOwner"] == account_id
+    assert "blue" in str(result).lower()

Original file line number	Diff line number	Diff line change
`@@ -90,7 +90,7 @@ def get_weather(location: Literal["New York", "London", "Tokyo"] = "New York"):`
`90`	`90`	`]`
`91`	`91`	`elif location.lower() == "tokyo":`
`92`	`92`	`# Read yellow.png file for weather icon`
`93`		`- with open("tests_integ/yellow.png", "rb") as image_file:`
	`93`	`+ with open("tests_integ/resources/yellow.png", "rb") as image_file:`
`94`	`94`	`png_data = image_file.read()`
`95`	`95`	`return [`
`96`	`96`	`EmbeddedResource(`