test: add stream delivery integ tests and TESTING.md

Author: Hweinstock

.github/workflows/integration-testing.yml

@@ -117,6 +117,12 @@ jobs:
             timeout: 10
             extra-deps: ""
             ignore: ""
+          # TODO: expand to full tests_integ/memory once test stability is addressed
+          - group: memory
+            path: tests_integ/memory/test_memory_client.py -k "TestMemoryClient"
+            timeout: 10
+            extra-deps: ""
+            ignore: ""
           - group: evaluation
             path: tests_integ/evaluation
             timeout: 15
@@ -151,6 +157,8 @@ jobs:
         env:
           AWS_REGION: us-west-2
           PYTHONUNBUFFERED: 1
+          MEMORY_KINESIS_ARN: ${{ secrets.MEMORY_KINESIS_ARN }}
+          MEMORY_ROLE_ARN: ${{ secrets.MEMORY_ROLE_ARN }}
         id: tests
         timeout-minutes: ${{ matrix.timeout }}
         run: |

TESTING.md

@@ -0,0 +1,58 @@
+# 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, fails if unset)
+export MEMORY_KINESIS_ARN=<Kinesis Data Stream ARN>
+export MEMORY_ROLE_ARN=<IAM role ARN trusted by bedrock-agentcore.amazonaws.com with kinesis:PutRecord/PutRecords permissions>
+uv run pytest tests_integ/memory/test_memory_client.py -xvs -k "TestMemoryClient"
+
+# 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
+
+Integration tests run via `integration-testing.yml` on PRs to main. The workflow runs runtime, memory (stream delivery only), and evaluation tests in parallel.
+
+Stream delivery tests fail in CI unless `MEMORY_KINESIS_ARN` and `MEMORY_ROLE_ARN` secrets are configured on the repo. Other memory integration tests are not yet run in CI due to provisioning times and flaky LLM-dependent assertions — CI support is planned once test stability is addressed.
+
+## Conventions
+
+- Each test file should map 1:1 to a source code file when possible.
+- Each feature area should correspond to a single test class named after the class under test (e.g. `TestMemoryClient`).
+- Expensive resources (e.g. memories) should be created the minimum number of times, ideally once in `setup_class`.
+- All resources created during tests must be cleaned up in `teardown_class`.
+- Use `self.__class__.memory_ids` to track resources for cleanup across test methods.

tests_integ/memory/test_memory_client.py

@@ -3,8 +3,11 @@
 import logging
 import os
 import time
+import uuid
 from datetime import datetime
 
+import pytest
+
 from bedrock_agentcore.memory import MemoryClient
 
 # Use INFO level logging for cleaner output
@@ -408,5 +411,70 @@ def main():
             logger.error("Failed to delete test memory: %s", e)
 
 
+@pytest.mark.integration
+class TestMemoryClient:
+    """Integration tests for streamDeliveryResources on MemoryClient."""
+
+    @classmethod
+    def setup_class(cls):
+        cls.kinesis_stream_arn = os.environ.get("MEMORY_KINESIS_ARN")
+        cls.execution_role_arn = os.environ.get("MEMORY_ROLE_ARN")
+
+        if not cls.kinesis_stream_arn or not cls.execution_role_arn:
+            pytest.fail("MEMORY_KINESIS_ARN and MEMORY_ROLE_ARN must be set")
+
+        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_and_update(self):
+        """Create memory with FULL_CONTENT delivery, update to METADATA_ONLY via passthrough."""
+        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["id"]
+        self.__class__.memory_ids.append(memory_id)
+
+        assert memory["streamDeliveryResources"] == delivery_config
+
+        # Test update via MemoryClient.__getattr__ passthrough to boto3 client.
+        # Uses camelCase params because the passthrough forwards directly to boto3
+        # without the snake_case translation that explicit SDK methods provide.
+        updated_config = self._make_delivery_config("METADATA_ONLY")
+        response = self.client.update_memory(
+            memoryId=memory_id,
+            clientToken=str(uuid.uuid4()),
+            streamDeliveryResources=updated_config,
+        )
+        assert response["memory"]["streamDeliveryResources"] == updated_config
+
+
 if __name__ == "__main__":
     main()