diff --git a/cli/real_clearance_composition.py b/cli/real_clearance_composition.py index cfd4e97..7faa40e 100644 --- a/cli/real_clearance_composition.py +++ b/cli/real_clearance_composition.py @@ -26,6 +26,8 @@ _CLEANUP_TIMEOUT_S: float = 5.0 _STOP_TIMEOUT_S: float = 3.0 _DIAGNOSTIC_READ_TIMEOUT_S: float = 10.0 +_SW_KEEPALIVE_POLL_INTERVAL_S: float = 0.5 +_SW_KEEPALIVE_MAX_RETRIES: int = 6 _logger = logging.getLogger(__name__) @@ -244,6 +246,16 @@ def make_sw_keepalive_checker( loop throughout the session. """ + async def _poll() -> bool | None: + attempts = _SW_KEEPALIVE_MAX_RETRIES + 1 + for i in range(attempts): + result: bool | None = await engine.read_extension_alarm("swKeepalive") + if result is True or result is None: + return result + if i < attempts - 1: + await asyncio.sleep(_SW_KEEPALIVE_POLL_INTERVAL_S) + return False + def _check() -> bool | None: _is_closed = loop.is_closed() _is_running = loop.is_running() @@ -255,7 +267,7 @@ def _check() -> bool | None: try: result: bool | None = loop.run_until_complete( asyncio.wait_for( - engine.read_extension_alarm("swKeepalive"), + _poll(), timeout=_DIAGNOSTIC_READ_TIMEOUT_S, ) ) diff --git a/tests/unit/cli/test_real_clearance_composition.py b/tests/unit/cli/test_real_clearance_composition.py index 50abe4e..26ca650 100644 --- a/tests/unit/cli/test_real_clearance_composition.py +++ b/tests/unit/cli/test_real_clearance_composition.py @@ -808,6 +808,72 @@ def test_callable_delegates_to_engine_false(self) -> None: assert result is False + def test_false_is_retried_and_resolves_true(self) -> None: + """False on first read is retried; True on subsequent read returns True.""" + from cli.real_clearance_composition import make_sw_keepalive_checker + + engine = MagicMock() + engine.read_extension_alarm = AsyncMock(side_effect=[False, True]) + loop = asyncio.new_event_loop() + try: + checker = make_sw_keepalive_checker(engine=engine, loop=loop) + result = checker() + finally: + loop.close() + + assert result is True + assert engine.read_extension_alarm.call_count == 2 + + def test_false_repeatedly_returns_false_after_max_attempts(self) -> None: + """Persistent False across all attempts returns False (not indefinite loop).""" + from cli.real_clearance_composition import ( + _SW_KEEPALIVE_MAX_RETRIES, + make_sw_keepalive_checker, + ) + + engine = MagicMock() + engine.read_extension_alarm = AsyncMock(return_value=False) + loop = asyncio.new_event_loop() + try: + checker = make_sw_keepalive_checker(engine=engine, loop=loop) + result = checker() + finally: + loop.close() + + assert result is False + assert engine.read_extension_alarm.call_count == _SW_KEEPALIVE_MAX_RETRIES + 1 + + def test_none_not_retried_returns_none(self) -> None: + """None (SW unreachable) is not retried — returned immediately.""" + from cli.real_clearance_composition import make_sw_keepalive_checker + + engine = MagicMock() + engine.read_extension_alarm = AsyncMock(return_value=None) + loop = asyncio.new_event_loop() + try: + checker = make_sw_keepalive_checker(engine=engine, loop=loop) + result = checker() + finally: + loop.close() + + assert result is None + assert engine.read_extension_alarm.call_count == 1 + + def test_false_then_none_returns_none(self) -> None: + """If alarm disappears mid-retry (False then None), returns None.""" + from cli.real_clearance_composition import make_sw_keepalive_checker + + engine = MagicMock() + engine.read_extension_alarm = AsyncMock(side_effect=[False, None]) + loop = asyncio.new_event_loop() + try: + checker = make_sw_keepalive_checker(engine=engine, loop=loop) + result = checker() + finally: + loop.close() + + assert result is None + def test_injection_domain_consistent_with_navigation_url() -> None: """allowed_clearance_domain must equal the hostname of the navigation target URL.