test: add stream delivery integ tests and TESTING.md

Author: Hweinstock

TESTING.md

@@ -0,0 +1,54 @@
+# Integration Tests
+
+See [CONTRIBUTING.md](CONTRIBUTING.md) for initial setup (Python, `uv sync`, etc).
+
+All integration tests require AWS credentials with `bedrock-agentcore:*` and `bedrock-agentcore-control:*` permissions.
+
+```bash
+export AWS_PROFILE=<your-profile>
+export BEDROCK_TEST_REGION=us-west-2
+```
+
+## Runtime
+
+```bash
+uv run pytest tests_integ/runtime -xvs --log-cli-level=INFO
+```
+
+## Memory
+
+```bash
+# Control plane (CRUD on memories/strategies, 5-10 min due to provisioning)
+uv run pytest tests_integ/memory/test_controlplane.py -xvs -k "integration"
+
+# Stream delivery (requires env vars below, skips if unset)
+export BEDROCK_TEST_KINESIS_STREAM_ARN=<Kinesis Data Stream ARN>
+export BEDROCK_TEST_EXECUTION_ROLE_ARN=<IAM role ARN trusted by bedrock-agentcore.amazonaws.com with kinesis:PutRecord/PutRecords permissions>
+uv run pytest tests_integ/memory/test_controlplane.py -xvs -k "TestStreamDeliveryResources"
+
+# Client and developer experience
+uv run pytest tests_integ/memory/test_memory_client.py -xvs
+uv run pytest tests_integ/memory/test_devex.py -xvs
+```
+
+## Identity
+
+```bash
+uv run pytest tests_integ/identity -xvs
+```
+
+## Tools (Code Interpreter, Browser)
+
+```bash
+uv run pytest tests_integ/tools -xvs
+```
+
+## CI
+
+| Workflow | Tests | Required |
+|---|---|---|
+| `integration-testing.yml` | `tests_integ/runtime` | Yes |
+| `integration-testing.yml` | `tests_integ/memory` | Yes |
+| `integration-testing.yml` | `tests_integ/evaluation` | Yes |
+
+Stream delivery tests (`TestStreamDeliveryResources`) will skip in CI unless `BEDROCK_TEST_KINESIS_STREAM_ARN` and `BEDROCK_TEST_EXECUTION_ROLE_ARN` secrets are configured on the repo.

tests_integ/memory/test_stream_delivery.py

@@ -0,0 +1,70 @@
+"""Integration tests for streamDeliveryResources on MemoryClient.
+
+Requires environment variables:
+    BEDROCK_TEST_KINESIS_STREAM_ARN: ARN of a pre-existing Kinesis stream
+    BEDROCK_TEST_EXECUTION_ROLE_ARN: ARN of an IAM role the memory service can assume
+
+Run with:
+    pytest tests_integ/memory/test_stream_delivery.py -xvs --log-cli-level=INFO
+"""
+
+import os
+import time
+
+import pytest
+
+from bedrock_agentcore.memory import MemoryClient
+
+
+@pytest.mark.integration
+class TestStreamDeliveryResources:
+    """Integration tests for streamDeliveryResources."""
+
+    @classmethod
+    def setup_class(cls):
+        cls.kinesis_stream_arn = os.environ.get("BEDROCK_TEST_KINESIS_STREAM_ARN")
+        cls.execution_role_arn = os.environ.get("BEDROCK_TEST_EXECUTION_ROLE_ARN")
+
+        if not cls.kinesis_stream_arn or not cls.execution_role_arn:
+            pytest.skip("BEDROCK_TEST_KINESIS_STREAM_ARN and BEDROCK_TEST_EXECUTION_ROLE_ARN required")
+
+        cls.region = os.environ.get("BEDROCK_TEST_REGION", "us-west-2")
+        cls.client = MemoryClient(region_name=cls.region)
+        cls.test_prefix = f"test_stream_{int(time.time())}"
+        cls.memory_ids = []
+
+    @classmethod
+    def teardown_class(cls):
+        for memory_id in cls.memory_ids:
+            try:
+                cls.client.delete_memory(memory_id)
+            except Exception as e:
+                print(f"Failed to delete memory {memory_id}: {e}")
+
+    def _make_delivery_config(self, level="FULL_CONTENT"):
+        return {
+            "resources": [
+                {
+                    "kinesis": {
+                        "dataStreamArn": self.kinesis_stream_arn,
+                        "contentConfigurations": [{"type": "MEMORY_RECORDS", "level": level}],
+                    }
+                }
+            ]
+        }
+
+    def test_stream_delivery_create(self):
+        """Create memory with stream delivery config and verify via get_memory."""
+        delivery_config = self._make_delivery_config("FULL_CONTENT")
+
+        memory = self.client.create_memory_and_wait(
+            name=f"{self.test_prefix}_stream",
+            strategies=[],
+            memory_execution_role_arn=self.execution_role_arn,
+            stream_delivery_resources=delivery_config,
+        )
+
+        memory_id = memory.get("memoryId", memory.get("id"))
+        self.__class__.memory_ids.append(memory_id)
+
+        assert memory["streamDeliveryResources"] == delivery_config