fail github actions when coverage threshold is not met

.github/workflows/ci.yml

@@ -71,7 +71,7 @@ jobs:
 
       - name: Run tests with coverage
         run: |
-          uv run pytest tests/ --cov=src --cov-report=xml --cov-report=html --cov-fail-under=56
+          uv run pytest tests/ --cov=src --cov-report=xml --cov-report=html --cov-precision=2
 
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v5

.pre-commit-config.yaml

@@ -53,8 +53,8 @@ repos:
           --cov=src,
           --cov-report=term-missing,
           --cov-report=html,
-          --cov-fail-under=90,
           --cov-branch,
+          --cov-precision=2,
           tests/
         ]
 

pyproject.toml

@@ -100,7 +100,7 @@ concurrency = ["thread", "multiprocessing"]
 
 [tool.coverage.report]
 show_missing = true
-fail_under = 56
+fail_under = 90
 skip_covered = false
 skip_empty = false
 

tests/bedrock_agentcore/identity/test_auth.py

@@ -257,8 +257,8 @@ async def test_async_func(param1, api_key=None):
                         provider_name="test-provider", agent_identity_token="test-agent-token"
                     )
 
-    def test_sync_function_decoration(self):
-        """Test decorator with sync function."""
+    def test_sync_function_decoration_no_running_loop(self):
+        """Test decorator with sync function when no asyncio loop is running."""
         # Mock IdentityClient
         with patch("bedrock_agentcore.identity.auth.IdentityClient") as mock_identity_client_class:
             mock_client = Mock()
@@ -289,6 +289,44 @@ def test_sync_func(param1, my_key=None):
 
                             assert result == "param1=value1, key=test-api-key"
 
+    def test_sync_function_decoration_with_running_loop(self):
+        """Test decorator with sync function when asyncio loop is running."""
+        # Mock IdentityClient
+        with patch("bedrock_agentcore.identity.auth.IdentityClient") as mock_identity_client_class:
+            mock_client = Mock()
+            mock_identity_client_class.return_value = mock_client
+
+            # Mock _get_workload_access_token
+            with patch(
+                "bedrock_agentcore.identity.auth._get_workload_access_token", new_callable=AsyncMock
+            ) as mock_get_agent_token:
+                mock_get_agent_token.return_value = "test-agent-token"
+
+                # Mock client.get_api_key
+                mock_client.get_api_key = AsyncMock(return_value="test-api-key")
+
+                # Mock _get_region
+                with patch("bedrock_agentcore.identity.auth._get_region", return_value="us-west-2"):
+
+                    @requires_api_key(provider_name="test-provider")
+                    def test_sync_func(param1, api_key=None):
+                        return f"param1={param1}, key={api_key}"
+
+                    # Mock asyncio.get_running_loop to succeed (loop is running)
+                    with patch("asyncio.get_running_loop"):
+                        with patch("concurrent.futures.ThreadPoolExecutor") as mock_executor_class:
+                            mock_executor = Mock()
+                            mock_executor_class.return_value.__enter__.return_value = mock_executor
+
+                            mock_future = Mock()
+                            mock_future.result.return_value = "test-api-key"
+                            mock_executor.submit.return_value = mock_future
+
+                            result = test_sync_func("value1")
+
+                            assert result == "param1=value1, key=test-api-key"
+                            mock_executor.submit.assert_called_once()
+
 
 class TestSetUpLocalAuth:
     """Test _set_up_local_auth function."""

tests/bedrock_agentcore/runtime/test_context.py

@@ -0,0 +1,30 @@
+"""Tests for Bedrock AgentCore context functionality."""
+
+import contextvars
+
+from bedrock_agentcore.runtime.context import BedrockAgentCoreContext
+
+
+class TestBedrockAgentCoreContext:
+    """Test BedrockAgentCoreContext functionality."""
+
+    def test_set_and_get_workload_access_token(self):
+        """Test setting and getting workload access token."""
+        token = "test-token-123"
+
+        BedrockAgentCoreContext.set_workload_access_token(token)
+        result = BedrockAgentCoreContext.get_workload_access_token()
+
+        assert result == token
+
+    def test_get_workload_access_token_when_none_set(self):
+        """Test getting workload access token when none is set."""
+        # Run this test in a completely fresh context to avoid interference from other tests
+        ctx = contextvars.Context()
+
+        def test_in_new_context():
+            result = BedrockAgentCoreContext.get_workload_access_token()
+            return result
+
+        result = ctx.run(test_in_new_context)
+        assert result is None