test(examples): add e2e tests for AgentCore zero-code example

Author: mesutoezdil-adfinis

tests/integration/test_live_agents.py

@@ -38,6 +38,8 @@
     reason="GOOGLE_API_KEY not set",
 )
 
+_skip_no_aws = pytest.mark.skipif(not os.environ.get("AWS_DEFAULT_REGION"), reason="AWS_DEFAULT_REGION not set")
+
 
 def _run_agent(
     script: str,
@@ -365,6 +367,39 @@ def test_session_visible_via_api(self, live_servers):
         assert session_name in session_ids
 
 
+_AGENTCORE_ENV = {"OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental"}
+_AGENTCORE_SCRIPT = "examples/zero-code-examples/agentcore/run.py"
+
+
+@_skip_no_aws
+class TestAgentCoreZeroCode:
+    def test_session_created_spans_only(self, live_servers):
+        main_port, otlp_http_port, mgr = live_servers
+        session_name = "e2e-agentcore"
+        result = _run_agent(_AGENTCORE_SCRIPT, otlp_http_port, session_name, extra_env=_AGENTCORE_ENV, timeout=60)
+        assert result.returncode == 0, f"Agent failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
+        wait_for_session_complete_sync(mgr, session_name, timeout=60)
+        s = mgr.sessions[session_name]
+        assert s.is_complete and s.source == "otlp" and len(s.spans) > 0
+
+    def test_invocations_extracted(self, live_servers):
+        main_port, otlp_http_port, mgr = live_servers
+        session_name = "e2e-agentcore-inv"
+        result = _run_agent(_AGENTCORE_SCRIPT, otlp_http_port, session_name, extra_env=_AGENTCORE_ENV, timeout=60)
+        assert result.returncode == 0, f"Agent failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
+        wait_for_session_complete_sync(mgr, session_name, timeout=60)
+        assert len(mgr.sessions[session_name].invocations) > 0
+
+    def test_session_visible_via_api(self, live_servers):
+        main_port, otlp_http_port, mgr = live_servers
+        session_name = "e2e-agentcore-api"
+        result = _run_agent(_AGENTCORE_SCRIPT, otlp_http_port, session_name, extra_env=_AGENTCORE_ENV, timeout=60)
+        assert result.returncode == 0
+        wait_for_session_complete_sync(mgr, session_name, timeout=60)
+        data = httpx.get(f"http://127.0.0.1:{main_port}/api/streaming/sessions").json()["data"]
+        assert session_name in [s["sessionId"] for s in data]
+
+
 @_skip_no_openai
 class TestAgentRerun:
     """Verify that re-running an agent with the same session_name creates