test: add stream delivery integ tests and TESTING.md

Author: Hweinstock

TESTING.md

@@ -0,0 +1,52 @@
+# 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 |
+
+Memory integration tests are not yet run in CI. They require pre-provisioned infrastructure (Kinesis stream, IAM role), take 5-10 minutes per run due to service provisioning times, and include LLM-dependent assertions that can be flaky. Run them locally for now — CI support is planned once infrastructure and test stability are addressed.

tests_integ/memory/test_controlplane.py

@@ -211,6 +211,92 @@ def test_workflow_3_list_and_delete_memories(self):
         assert len(remaining_test_memories) == 0, f"Expected 0 test memories, found {len(remaining_test_memories)}"
 
 
+@pytest.mark.integration
+class TestStreamDeliveryResources:
+    """Integration tests for streamDeliveryResources.
+
+    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
+    """
+
+    @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 = MemoryControlPlaneClient(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=memory_id,
+                    wait_for_deletion=True,
+                    wait_for_strategies=False,
+                    max_wait=120,
+                    poll_interval=5,
+                )
+            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_and_update(self):
+        """Create memory with FULL_CONTENT delivery, update to METADATA_ONLY, verify both."""
+        # Create with FULL_CONTENT
+        delivery_config = self._make_delivery_config("FULL_CONTENT")
+
+        memory = self.client.create_memory(
+            name=f"{self.test_prefix}_stream",
+            memory_execution_role_arn=self.execution_role_arn,
+            stream_delivery_resources=delivery_config,
+            wait_for_active=True,
+            max_wait=300,
+            poll_interval=10,
+        )
+
+        memory_id = memory["id"]
+        self.__class__.memory_ids.append(memory_id)
+
+        assert memory["status"] == "ACTIVE"
+        assert memory["streamDeliveryResources"] == delivery_config
+
+        # Verify via get_memory
+        details = self.client.get_memory(memory_id)
+        assert details["streamDeliveryResources"] == delivery_config
+
+        # Update to METADATA_ONLY (update_memory returns the updated state synchronously)
+        updated_config = self._make_delivery_config("METADATA_ONLY")
+        updated = self.client.update_memory(
+            memory_id=memory_id,
+            stream_delivery_resources=updated_config,
+        )
+
+        assert updated["streamDeliveryResources"] == updated_config
+
+        # Verify update via get_memory
+        details = self.client.get_memory(memory_id)
+        assert details["streamDeliveryResources"] == updated_config
+
+
 @pytest.mark.unit
 class TestMemoryControlPlaneClientUnit:
     """Unit tests for MemoryControlPlaneClient using mocks."""