Add s3 location tests including video

Author: Unshure

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

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
 

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
 
 

tests/strands/models/test_bedrock.py

@@ -1787,33 +1787,6 @@ def test_format_request_filters_image_content_blocks(model, model_id):
     assert "metadata" not in image_block
 
 
-def test_format_request_filters_nested_image_s3_fields(model, model_id):
-    """Test that s3Location is used when present, even if bytes is also in source."""
-    messages = [
-        {
-            "role": "user",
-            "content": [
-                {
-                    "image": {
-                        "format": "png",
-                        "source": {
-                            "bytes": b"image_data",
-                            "s3Location": {"uri": "s3://my-bucket/image.png", "bucketOwner": "123456789012"},
-                        },
-                    }
-                }
-            ],
-        }
-    ]
-
-    formatted_request = model._format_request(messages)
-    image_source = formatted_request["messages"][0]["content"][0]["image"]["source"]
-
-    # s3Location should be used when present
-    assert image_source == {"s3Location": {"uri": "s3://my-bucket/image.png", "bucketOwner": "123456789012"}}
-    assert "bytes" not in image_source
-
-
 def test_format_request_image_s3_location_only(model, model_id):
     """Test that image with only s3Location is properly formatted."""
     messages = [

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()
 

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,
 )
 

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()
 

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(

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: