test: cover access-policy RBAC deny-path and circuit-breaker probe guard

backend/tests/test_access_policy.py

@@ -3,6 +3,7 @@
 from services.access_policy import (
     AccessRequest,
     ResourcePolicy,
+    _equivalent_roles,
     _is_system_admin_role,
     evaluate_access,
 )
@@ -316,6 +317,42 @@ def test_tenant_admin_satisfies_organization_admin_alias_deterministically():
     assert decision.reason == "allowed"
 
 
+def test_owner_without_role_or_group_permission_is_rbac_denied():
+    """Owning (or being delegated) a resource is necessary but not sufficient:
+    when every ABAC gate clears and ownership holds, a non-admin whose role and
+    group both fail the permit set is still denied (the final RBAC deny-path)."""
+    decision = evaluate_access(
+        AccessRequest(
+            user_id="alice",
+            role="member",
+            organization_id="org-acme",
+            group_ids=("sales",),
+            data_region="eu",
+            consent_scopes=("mail.read",),
+        ),
+        ResourcePolicy(
+            owner_id="alice",  # owns the resource -> ownership gate clears
+            organization_id="org-acme",
+            permitted_roles=("tenant_admin",),  # member does not satisfy tenant_admin
+            permitted_group_ids=("exec",),  # no overlap with ("sales",)
+            data_region="eu",
+            required_consent_scopes=("mail.read",),
+        ),
+    )
+
+    assert decision.allowed is False
+    assert decision.reason == "rbac_denied"
+
+
+def test_equivalent_roles_maps_known_alias_and_falls_back_for_unknown():
+    """Known roles expand to their alias set; an unrecognised role maps only to
+    itself (the fallback branch)."""
+    assert _equivalent_roles("system_admin") == frozenset(
+        {"system_admin", "platform_admin"}
+    )
+    assert _equivalent_roles("unknown_role") == frozenset({"unknown_role"})
+
+
 def test_abac_owner_denial_overrides_group_admin_rbac_allow_without_delegation():
     decision = evaluate_access(
         AccessRequest(

backend/tests/test_circuit_breaker.py

@@ -1,5 +1,7 @@
 """Tests for the provider circuit breaker."""
 
+import asyncio
+
 import pytest
 
 from services.circuit_breaker import CircuitBreaker, CircuitOpenError
@@ -87,6 +89,35 @@ async def test_half_open_probe_failure_reopens():
         await breaker.call("p", _succeeding())
 
 
+@pytest.mark.asyncio
+async def test_half_open_rejects_second_probe_while_first_in_flight():
+    """Only one half-open probe is admitted at a time: a concurrent call that
+    arrives while the single probe is still awaiting fails fast rather than
+    piling a second request onto a provider that may still be down."""
+    clock = _Clock()
+    breaker = CircuitBreaker(failure_threshold=1, cooldown_seconds=10, clock=clock)
+
+    with pytest.raises(RuntimeError):
+        await breaker.call("p", _failing())
+    clock.now = 11.0  # past cooldown -> the next call is the lone half-open probe
+
+    release = asyncio.Event()
+
+    async def slow_probe():
+        await release.wait()
+        return "recovered"
+
+    probe = asyncio.create_task(breaker.call("p", slow_probe))
+    await asyncio.sleep(0)  # let the probe start and claim the in-flight slot
+
+    # A second call while the probe is in flight is rejected fast.
+    with pytest.raises(CircuitOpenError):
+        await breaker.call("p", _succeeding())
+
+    release.set()
+    assert await probe == "recovered"
+
+
 @pytest.mark.asyncio
 async def test_keys_are_isolated():
     clock = _Clock()