From 616f87d05e2ef8bca921a059430eea720d10db88 Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 18:57:46 -0400 Subject: [PATCH 01/19] Fix SSE field casing and add event handler support The Risco cloud SSE API sends IsOffline and LastStatusUpdate in PascalCase, but the code was reading them as isOffline/lastStatusUpdate, causing the offline check and deduplication logic to silently do nothing. Also adds push-based event handling: when the SSE runtimeUpdate message includes LastEventUpdated, get_events is called and registered event handlers are invoked, so callers no longer need to poll for new events. Co-Authored-By: Claude Sonnet 4.6 --- examples/cloud/cloud_sse_example.py | 7 +++ pyrisco/cloud/risco_cloud.py | 18 ++++++- tests/test_risco_cloud.py | 82 +++++++++++++++++++++++++++-- 3 files changed, 101 insertions(+), 6 deletions(-) diff --git a/examples/cloud/cloud_sse_example.py b/examples/cloud/cloud_sse_example.py index 05258d4..7d0abfb 100644 --- a/examples/cloud/cloud_sse_example.py +++ b/examples/cloud/cloud_sse_example.py @@ -12,6 +12,12 @@ async def on_state(alarm): print(f" Zone {zone_id} ({zone.name}): triggered={zone.triggered}, bypassed={zone.bypassed}") +async def on_event(events): + print(f"Received {len(events)} new event(s)") + for event in events: + print(f" [{event.time}] {event.text}") + + async def on_error(error): print(f"SSE error: {error}, reconnecting in 5s...") await asyncio.sleep(5) @@ -22,6 +28,7 @@ async def on_error(error): async def main(): await risco.login() risco.add_state_handler(on_state) + risco.add_event_handler(on_event) risco.add_error_handler(on_error) await risco.subscribe_states() await asyncio.Future() # run forever diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index fd2138b..d3af50d 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -43,7 +43,9 @@ def __init__(self, username, password, pin, language="en"): self._created_session = False self._state_handlers = [] self._error_handlers = [] + self._event_handlers = [] self._last_status_update = None + self._last_event_update = None self._latest_state = None self._subscription_task = None @@ -167,9 +169,9 @@ async def _sse_loop(self): await handler(e) async def _handle_runtime_update(self, data): - if data.get("isOffline"): + if data.get("IsOffline"): return - ts_str = data.get("lastStatusUpdate") + ts_str = data.get("LastStatusUpdate") if not ts_str: return update_time = datetime.fromisoformat(ts_str) @@ -181,6 +183,14 @@ async def _handle_runtime_update(self, data): self._latest_state = alarm for handler in list(self._state_handlers): await handler(alarm) + event_ts_str = data.get("LastEventUpdated") + if event_ts_str: + event_time = datetime.fromisoformat(event_ts_str) + if self._last_event_update is None or event_time > self._last_event_update: + events = await self.get_events(self._last_event_update) + self._last_event_update = event_time + for handler in list(self._event_handlers): + await handler(events) async def close(self): """Close the connection.""" @@ -215,6 +225,10 @@ def add_error_handler(self, handler): """Register an async callback for SSE errors. Returns a remover callable.""" return RiscoCloud._add_handler(self._error_handlers, handler) + def add_event_handler(self, handler): + """Register an async callback for event log updates via SSE. Returns a remover callable.""" + return RiscoCloud._add_handler(self._event_handlers, handler) + async def subscribe_states(self): """Start listening for push state updates via SSE.""" self._subscription_task = asyncio.create_task(self._sse_loop()) diff --git a/tests/test_risco_cloud.py b/tests/test_risco_cloud.py index b59319a..06f9ff3 100644 --- a/tests/test_risco_cloud.py +++ b/tests/test_risco_cloud.py @@ -214,7 +214,7 @@ async def test_subscribe_states_calls_handler(self, MockClientSession, mock_site mock_session = MockClientSession.return_value mock_resp = MagicMock() mock_resp.content = _make_sse_stream( - ("runtimeUpdate", {"lastStatusUpdate": "2024-01-01T12:00:00Z"}), + ("runtimeUpdate", {"LastStatusUpdate": "2024-01-01T12:00:00Z"}), ) mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) @@ -248,8 +248,8 @@ async def test_subscribe_states_no_update_same_timestamp(self, MockClientSession mock_session = MockClientSession.return_value mock_resp = MagicMock() mock_resp.content = _make_sse_stream( - ("runtimeUpdate", {"lastStatusUpdate": "2024-01-01T12:00:00Z"}), - ("runtimeUpdate", {"lastStatusUpdate": "2024-01-01T12:00:00Z"}), + ("runtimeUpdate", {"LastStatusUpdate": "2024-01-01T12:00:00Z"}), + ("runtimeUpdate", {"LastStatusUpdate": "2024-01-01T12:00:00Z"}), ) mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) @@ -288,7 +288,7 @@ async def test_subscribe_states_skips_fetch_when_offline(self, MockClientSession mock_session = MockClientSession.return_value mock_resp = MagicMock() mock_resp.content = _make_sse_stream( - ("runtimeUpdate", {"lastStatusUpdate": "2024-01-01T12:00:00Z", "isOffline": True}), + ("runtimeUpdate", {"LastStatusUpdate": "2024-01-01T12:00:00Z", "IsOffline": True}), ) mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) @@ -340,5 +340,79 @@ async def on_error(err): self.assertIs(received_errors[0], error) + @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) + @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') + async def test_subscribe_states_calls_event_handler(self, MockClientSession, mock_site_post): + state_payload = {"partitions": [], "zones": []} + event_payload = {"controlPanelEventsList": []} + mock_site_post.side_effect = [ + ({"state": {"status": state_payload}}, False), + (event_payload, False), + ] + + mock_session = MockClientSession.return_value + mock_resp = MagicMock() + mock_resp.content = _make_sse_stream( + ("runtimeUpdate", { + "LastStatusUpdate": "2024-01-01T12:00:00Z", + "LastEventUpdated": "2024-01-01T12:00:00Z", + }), + ) + mock_get_cm = MagicMock() + mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) + mock_get_cm.__aexit__ = AsyncMock(return_value=None) + mock_session.get.return_value = mock_get_cm + + received_events = [] + async def on_event(events): + received_events.append(events) + + risco_cloud = RiscoCloud("username", "password", "pin") + risco_cloud._access_token = "mock_access_token" + risco_cloud._site_id = "mock_site_id" + risco_cloud._session_id = "mock_session_id" + risco_cloud._session = mock_session + risco_cloud.add_event_handler(on_event) + + await risco_cloud.subscribe_states() + await risco_cloud._subscription_task + + self.assertEqual(len(received_events), 1) + self.assertIsInstance(received_events[0], list) + + @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) + @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') + async def test_subscribe_states_no_event_handler_when_no_last_event_updated(self, MockClientSession, mock_site_post): + state_payload = {"partitions": [], "zones": []} + mock_site_post.return_value = ({"state": {"status": state_payload}}, False) + + mock_session = MockClientSession.return_value + mock_resp = MagicMock() + mock_resp.content = _make_sse_stream( + ("runtimeUpdate", {"LastStatusUpdate": "2024-01-01T12:00:00Z"}), + ) + mock_get_cm = MagicMock() + mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) + mock_get_cm.__aexit__ = AsyncMock(return_value=None) + mock_session.get.return_value = mock_get_cm + + received_events = [] + async def on_event(events): + received_events.append(events) + + risco_cloud = RiscoCloud("username", "password", "pin") + risco_cloud._access_token = "mock_access_token" + risco_cloud._site_id = "mock_site_id" + risco_cloud._session_id = "mock_session_id" + risco_cloud._session = mock_session + risco_cloud.add_event_handler(on_event) + + await risco_cloud.subscribe_states() + await risco_cloud._subscription_task + + self.assertEqual(len(received_events), 0) + self.assertEqual(mock_site_post.call_count, 1) # only the state fetch, no event fetch + + if __name__ == '__main__': unittest.main() From 7cc186f6797c7b5ff6226a61106a5257478eb7d0 Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 19:04:57 -0400 Subject: [PATCH 02/19] Handle null response from get_events The API can return a null response body when the event log query returns nothing. Guard against this by returning an empty list instead of crashing. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 2 ++ tests/test_risco_cloud.py | 39 ++++++++++++++++++++++++++++++++++++ 2 files changed, 41 insertions(+) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index d3af50d..6d2fb3b 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -279,6 +279,8 @@ async def get_events(self, newer_than, count=10): "offset": 0, } response, assumed_control_panel_state = await self._site_post(EVENTS_URL, body) + if not response: + return [] return [Event(e) for e in response["controlPanelEventsList"]] async def bypass_zone(self, zone, bypass): diff --git a/tests/test_risco_cloud.py b/tests/test_risco_cloud.py index 06f9ff3..4c17d37 100644 --- a/tests/test_risco_cloud.py +++ b/tests/test_risco_cloud.py @@ -340,6 +340,45 @@ async def on_error(err): self.assertIs(received_errors[0], error) + @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) + @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') + async def test_subscribe_states_calls_event_handler_with_null_response(self, MockClientSession, mock_site_post): + state_payload = {"partitions": [], "zones": []} + mock_site_post.side_effect = [ + ({"state": {"status": state_payload}}, False), + (None, False), + ] + + mock_session = MockClientSession.return_value + mock_resp = MagicMock() + mock_resp.content = _make_sse_stream( + ("runtimeUpdate", { + "LastStatusUpdate": "2024-01-01T12:00:00Z", + "LastEventUpdated": "2024-01-01T12:00:00Z", + }), + ) + mock_get_cm = MagicMock() + mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) + mock_get_cm.__aexit__ = AsyncMock(return_value=None) + mock_session.get.return_value = mock_get_cm + + received_events = [] + async def on_event(events): + received_events.append(events) + + risco_cloud = RiscoCloud("username", "password", "pin") + risco_cloud._access_token = "mock_access_token" + risco_cloud._site_id = "mock_site_id" + risco_cloud._session_id = "mock_session_id" + risco_cloud._session = mock_session + risco_cloud.add_event_handler(on_event) + + await risco_cloud.subscribe_states() + await risco_cloud._subscription_task + + self.assertEqual(len(received_events), 1) + self.assertEqual(received_events[0], []) + @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') async def test_subscribe_states_calls_event_handler(self, MockClientSession, mock_site_post): From d12673c75f4017c48977528138c34b3de6992b2f Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 19:16:59 -0400 Subject: [PATCH 03/19] Fix datetime serialization and skip event fetch without handlers Two issues from code review: - _last_event_update is a datetime; convert to isoformat() before passing as newerThan to avoid JSON serialization error on the second fetch - Skip get_events entirely when no event handlers are registered Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 6d2fb3b..c9e4026 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -184,10 +184,11 @@ async def _handle_runtime_update(self, data): for handler in list(self._state_handlers): await handler(alarm) event_ts_str = data.get("LastEventUpdated") - if event_ts_str: + if event_ts_str and self._event_handlers: event_time = datetime.fromisoformat(event_ts_str) if self._last_event_update is None or event_time > self._last_event_update: - events = await self.get_events(self._last_event_update) + newer_than = self._last_event_update.isoformat() if self._last_event_update else None + events = await self.get_events(newer_than) self._last_event_update = event_time for handler in list(self._event_handlers): await handler(events) From 344465d191b34cee22d8ad5e2f81c67b47ebf79a Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 19:44:29 -0400 Subject: [PATCH 04/19] Decouple state and event update paths; tighten null check - Restructure _handle_runtime_update so LastEventUpdated is processed independently of LastStatusUpdate: a message with only LastEventUpdated (or a non-newer LastStatusUpdate) now correctly triggers event handlers - Replace `if not response` with `if response is None` in get_events so an unexpected empty-dict response surfaces as a KeyError rather than being silently swallowed Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 22 ++++++++++------------ tests/test_risco_cloud.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 12 deletions(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index c9e4026..5a46004 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -172,17 +172,15 @@ async def _handle_runtime_update(self, data): if data.get("IsOffline"): return ts_str = data.get("LastStatusUpdate") - if not ts_str: - return - update_time = datetime.fromisoformat(ts_str) - if self._last_status_update is not None and update_time <= self._last_status_update: - return - resp, assumed = await self._site_post(STATE_URL, {}) - alarm = Alarm(self, resp["state"]["status"], assumed) - self._last_status_update = update_time - self._latest_state = alarm - for handler in list(self._state_handlers): - await handler(alarm) + if ts_str: + update_time = datetime.fromisoformat(ts_str) + if self._last_status_update is None or update_time > self._last_status_update: + resp, assumed = await self._site_post(STATE_URL, {}) + alarm = Alarm(self, resp["state"]["status"], assumed) + self._last_status_update = update_time + self._latest_state = alarm + for handler in list(self._state_handlers): + await handler(alarm) event_ts_str = data.get("LastEventUpdated") if event_ts_str and self._event_handlers: event_time = datetime.fromisoformat(event_ts_str) @@ -280,7 +278,7 @@ async def get_events(self, newer_than, count=10): "offset": 0, } response, assumed_control_panel_state = await self._site_post(EVENTS_URL, body) - if not response: + if response is None: return [] return [Event(e) for e in response["controlPanelEventsList"]] diff --git a/tests/test_risco_cloud.py b/tests/test_risco_cloud.py index 4c17d37..3e14ee4 100644 --- a/tests/test_risco_cloud.py +++ b/tests/test_risco_cloud.py @@ -306,6 +306,40 @@ async def test_subscribe_states_skips_fetch_when_offline(self, MockClientSession mock_site_post.assert_not_called() + @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) + @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') + async def test_subscribe_states_event_handler_fires_without_status_update(self, MockClientSession, mock_site_post): + """LastEventUpdated should trigger the event handler even when LastStatusUpdate is absent.""" + event_payload = {"controlPanelEventsList": []} + mock_site_post.return_value = (event_payload, False) + + mock_session = MockClientSession.return_value + mock_resp = MagicMock() + mock_resp.content = _make_sse_stream( + ("runtimeUpdate", {"LastEventUpdated": "2024-01-01T12:00:00Z"}), + ) + mock_get_cm = MagicMock() + mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) + mock_get_cm.__aexit__ = AsyncMock(return_value=None) + mock_session.get.return_value = mock_get_cm + + received_events = [] + async def on_event(events): + received_events.append(events) + + risco_cloud = RiscoCloud("username", "password", "pin") + risco_cloud._access_token = "mock_access_token" + risco_cloud._site_id = "mock_site_id" + risco_cloud._session_id = "mock_session_id" + risco_cloud._session = mock_session + risco_cloud.add_event_handler(on_event) + + await risco_cloud.subscribe_states() + await risco_cloud._subscription_task + + self.assertEqual(len(received_events), 1) + mock_site_post.call_count == 1 # only event fetch, no state fetch + @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') async def test_subscribe_states_calls_error_handler(self, MockClientSession): mock_session = MockClientSession.return_value From bc71a97240797bc8004e6a04b9c0dbdaadef0067 Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 19:50:31 -0400 Subject: [PATCH 05/19] =?UTF-8?q?Fix=20no-op=20comparison=20in=20test=20?= =?UTF-8?q?=E2=80=94=20use=20assertEqual=20for=20call=5Fcount?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- tests/test_risco_cloud.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_risco_cloud.py b/tests/test_risco_cloud.py index 3e14ee4..823c65c 100644 --- a/tests/test_risco_cloud.py +++ b/tests/test_risco_cloud.py @@ -338,7 +338,7 @@ async def on_event(events): await risco_cloud._subscription_task self.assertEqual(len(received_events), 1) - mock_site_post.call_count == 1 # only event fetch, no state fetch + self.assertEqual(mock_site_post.call_count, 1) # only event fetch, no state fetch @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') async def test_subscribe_states_calls_error_handler(self, MockClientSession): From c74737c532864df547663b9df2c819ddda5e78e7 Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 20:34:45 -0400 Subject: [PATCH 06/19] Fetch initial state immediately after SSE connection opens MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before consuming any SSE messages, do one state fetch and notify handlers. Since the SSE connection is already open at that point, the server is buffering events — so no state changes in the window between the fetch and the first message can be missed. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 7 ++++ tests/test_risco_cloud.py | 73 ++++++++++++++++++++++++++++++------ 2 files changed, 68 insertions(+), 12 deletions(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 5a46004..54483a2 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -150,6 +150,13 @@ async def _sse_loop(self): } params = {"sessionToken": self._session_id} async with self._session.get(url, headers=headers, params=params) as resp: + # SSE connection is now open — fetch initial state before consuming + # messages so no state changes in between can be missed. + initial_resp, assumed = await self._site_post(STATE_URL, {}) + alarm = Alarm(self, initial_resp["state"]["status"], assumed) + self._latest_state = alarm + for handler in list(self._state_handlers): + await handler(alarm) event_type = None data_line = None async for line_bytes in resp.content: diff --git a/tests/test_risco_cloud.py b/tests/test_risco_cloud.py index 823c65c..eb1ed2d 100644 --- a/tests/test_risco_cloud.py +++ b/tests/test_risco_cloud.py @@ -205,10 +205,44 @@ async def test_get_state_with_fallback_to_cloud(self, mock_authenticated_post): self.assertTrue(mock_authenticated_post.call_args_list[0][0][1]['fromControlPanel']) self.assertFalse(mock_authenticated_post.call_args_list[1][0][1]['fromControlPanel']) + @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) + @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') + async def test_subscribe_states_notifies_handler_immediately(self, MockClientSession, mock_site_post): + """Handler should be called right after SSE connects, before any SSE messages.""" + state_payload = {"partitions": [], "zones": []} + mock_site_post.return_value = ({"state": {"status": state_payload}}, False) + + mock_session = MockClientSession.return_value + mock_resp = MagicMock() + mock_resp.content = _make_sse_stream() # no SSE messages + mock_get_cm = MagicMock() + mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) + mock_get_cm.__aexit__ = AsyncMock(return_value=None) + mock_session.get.return_value = mock_get_cm + + received = [] + async def on_state(alarm): + received.append(alarm) + + risco_cloud = RiscoCloud("username", "password", "pin") + risco_cloud._access_token = "mock_access_token" + risco_cloud._site_id = "mock_site_id" + risco_cloud._session_id = "mock_session_id" + risco_cloud._session = mock_session + risco_cloud.add_state_handler(on_state) + + await risco_cloud.subscribe_states() + await risco_cloud._subscription_task + + self.assertEqual(len(received), 1) + self.assertIsInstance(received[0], Alarm) + self.assertEqual(mock_site_post.call_count, 1) + @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') async def test_subscribe_states_calls_handler(self, MockClientSession, mock_site_post): state_payload = {"partitions": [], "zones": []} + # Two calls: initial state fetch + state fetch triggered by SSE message mock_site_post.return_value = ({"state": {"status": state_payload}}, False) mock_session = MockClientSession.return_value @@ -235,9 +269,9 @@ async def on_state(alarm): await risco_cloud.subscribe_states() await risco_cloud._subscription_task - self.assertEqual(len(received), 1) + self.assertEqual(len(received), 2) # initial fetch + SSE-triggered fetch self.assertIsInstance(received[0], Alarm) - self.assertEqual(mock_site_post.call_count, 1) + self.assertEqual(mock_site_post.call_count, 2) @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') @@ -265,7 +299,8 @@ async def test_subscribe_states_no_update_same_timestamp(self, MockClientSession await risco_cloud.subscribe_states() await risco_cloud._subscription_task - self.assertEqual(mock_site_post.call_count, 1) + # initial fetch + first SSE message; second SSE message has same timestamp so skipped + self.assertEqual(mock_site_post.call_count, 2) @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) async def test_get_state_returns_cached(self, mock_site_post): @@ -285,6 +320,9 @@ async def test_get_state_returns_cached(self, mock_site_post): @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') async def test_subscribe_states_skips_fetch_when_offline(self, MockClientSession, mock_site_post): + state_payload = {"partitions": [], "zones": []} + mock_site_post.return_value = ({"state": {"status": state_payload}}, False) + mock_session = MockClientSession.return_value mock_resp = MagicMock() mock_resp.content = _make_sse_stream( @@ -304,14 +342,19 @@ async def test_subscribe_states_skips_fetch_when_offline(self, MockClientSession await risco_cloud.subscribe_states() await risco_cloud._subscription_task - mock_site_post.assert_not_called() + # Only the initial fetch; the SSE message is skipped because IsOffline=True + self.assertEqual(mock_site_post.call_count, 1) @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') async def test_subscribe_states_event_handler_fires_without_status_update(self, MockClientSession, mock_site_post): """LastEventUpdated should trigger the event handler even when LastStatusUpdate is absent.""" + state_payload = {"partitions": [], "zones": []} event_payload = {"controlPanelEventsList": []} - mock_site_post.return_value = (event_payload, False) + mock_site_post.side_effect = [ + ({"state": {"status": state_payload}}, False), # initial state fetch + (event_payload, False), # event fetch from SSE message + ] mock_session = MockClientSession.return_value mock_resp = MagicMock() @@ -338,10 +381,14 @@ async def on_event(events): await risco_cloud._subscription_task self.assertEqual(len(received_events), 1) - self.assertEqual(mock_site_post.call_count, 1) # only event fetch, no state fetch + self.assertEqual(mock_site_post.call_count, 2) # initial state fetch + event fetch + @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') - async def test_subscribe_states_calls_error_handler(self, MockClientSession): + async def test_subscribe_states_calls_error_handler(self, MockClientSession, mock_site_post): + state_payload = {"partitions": [], "zones": []} + mock_site_post.return_value = ({"state": {"status": state_payload}}, False) + mock_session = MockClientSession.return_value error = RuntimeError("stream broken") @@ -379,8 +426,9 @@ async def on_error(err): async def test_subscribe_states_calls_event_handler_with_null_response(self, MockClientSession, mock_site_post): state_payload = {"partitions": [], "zones": []} mock_site_post.side_effect = [ - ({"state": {"status": state_payload}}, False), - (None, False), + ({"state": {"status": state_payload}}, False), # initial state fetch + ({"state": {"status": state_payload}}, False), # state fetch from SSE message + (None, False), # event fetch returns null ] mock_session = MockClientSession.return_value @@ -419,8 +467,9 @@ async def test_subscribe_states_calls_event_handler(self, MockClientSession, moc state_payload = {"partitions": [], "zones": []} event_payload = {"controlPanelEventsList": []} mock_site_post.side_effect = [ - ({"state": {"status": state_payload}}, False), - (event_payload, False), + ({"state": {"status": state_payload}}, False), # initial state fetch + ({"state": {"status": state_payload}}, False), # state fetch from SSE message + (event_payload, False), # event fetch ] mock_session = MockClientSession.return_value @@ -484,7 +533,7 @@ async def on_event(events): await risco_cloud._subscription_task self.assertEqual(len(received_events), 0) - self.assertEqual(mock_site_post.call_count, 1) # only the state fetch, no event fetch + self.assertEqual(mock_site_post.call_count, 2) # initial state fetch + state fetch from SSE, no event fetch if __name__ == '__main__': From 04043b372f12fadd607b9eb0d0290f8d0a357899 Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 20:43:03 -0400 Subject: [PATCH 07/19] Store _last_event_update as raw string to avoid isoformat Z vs +00:00 mismatch Storing the datetime and calling .isoformat() produces +00:00, inconsistent with the Z-suffixed strings the API uses. Store the original SSE string instead and pass it directly to get_events; parse to datetime only locally for comparison. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 54483a2..5355de4 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -191,10 +191,10 @@ async def _handle_runtime_update(self, data): event_ts_str = data.get("LastEventUpdated") if event_ts_str and self._event_handlers: event_time = datetime.fromisoformat(event_ts_str) - if self._last_event_update is None or event_time > self._last_event_update: - newer_than = self._last_event_update.isoformat() if self._last_event_update else None - events = await self.get_events(newer_than) - self._last_event_update = event_time + last_event_time = datetime.fromisoformat(self._last_event_update) if self._last_event_update else None + if last_event_time is None or event_time > last_event_time: + events = await self.get_events(self._last_event_update) + self._last_event_update = event_ts_str for handler in list(self._event_handlers): await handler(events) From 99f2c10157deaeb3323c0e394bcb86fbf5a3524e Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 23:23:49 -0400 Subject: [PATCH 08/19] Auto-reconnect SSE loop on connection errors Network drops (e.g. ClientPayloadError mid-stream) are now handled internally: the loop catches the exception, calls error handlers for observability, sleeps RECONNECT_DELAY (30 s), then reconnects without requiring any action from the caller. The SSE example is updated to reflect that error handlers are now informational-only and no longer need to drive reconnection. Co-Authored-By: Claude Sonnet 4.6 --- examples/cloud/cloud_sse_example.py | 5 +-- pyrisco/cloud/risco_cloud.py | 69 +++++++++++++++-------------- tests/test_risco_cloud.py | 66 +++++++++++++++++++-------- 3 files changed, 84 insertions(+), 56 deletions(-) diff --git a/examples/cloud/cloud_sse_example.py b/examples/cloud/cloud_sse_example.py index 7d0abfb..ddd1949 100644 --- a/examples/cloud/cloud_sse_example.py +++ b/examples/cloud/cloud_sse_example.py @@ -19,10 +19,7 @@ async def on_event(events): async def on_error(error): - print(f"SSE error: {error}, reconnecting in 5s...") - await asyncio.sleep(5) - await risco.login() - await risco.subscribe_states() + print(f"SSE error: {error} (will reconnect automatically)") async def main(): diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 5355de4..955a865 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -23,6 +23,7 @@ NUM_RETRIES = 3 RETRYABLE_RESULT_CODE = 72 +RECONNECT_DELAY = 30 class RiscoCloud: @@ -141,39 +142,41 @@ async def _send_control_command(self, body): return Alarm(self, resp, assumed_control_panel_state) async def _sse_loop(self): - try: - url = SSE_URL % self._site_id - headers = { - "authorization": f"Bearer {self._access_token}", - "User-Agent": "pyrisco/1.0", - "sessionToken": self._session_id, - } - params = {"sessionToken": self._session_id} - async with self._session.get(url, headers=headers, params=params) as resp: - # SSE connection is now open — fetch initial state before consuming - # messages so no state changes in between can be missed. - initial_resp, assumed = await self._site_post(STATE_URL, {}) - alarm = Alarm(self, initial_resp["state"]["status"], assumed) - self._latest_state = alarm - for handler in list(self._state_handlers): - await handler(alarm) - event_type = None - data_line = None - async for line_bytes in resp.content: - line = line_bytes.decode("utf-8").rstrip("\r\n") - if line.startswith("event:"): - event_type = line[6:].strip() - elif line.startswith("data:"): - data_line = line[5:].strip() - elif line == "" and event_type == "runtimeUpdate" and data_line: - await self._handle_runtime_update(json.loads(data_line)) - event_type = None - data_line = None - except asyncio.CancelledError: - raise - except Exception as e: - for handler in list(self._error_handlers): - await handler(e) + while True: + try: + url = SSE_URL % self._site_id + headers = { + "authorization": f"Bearer {self._access_token}", + "User-Agent": "pyrisco/1.0", + "sessionToken": self._session_id, + } + params = {"sessionToken": self._session_id} + async with self._session.get(url, headers=headers, params=params) as resp: + # SSE connection is now open — fetch initial state before consuming + # messages so no state changes in between can be missed. + initial_resp, assumed = await self._site_post(STATE_URL, {}) + alarm = Alarm(self, initial_resp["state"]["status"], assumed) + self._latest_state = alarm + for handler in list(self._state_handlers): + await handler(alarm) + event_type = None + data_line = None + async for line_bytes in resp.content: + line = line_bytes.decode("utf-8").rstrip("\r\n") + if line.startswith("event:"): + event_type = line[6:].strip() + elif line.startswith("data:"): + data_line = line[5:].strip() + elif line == "" and event_type == "runtimeUpdate" and data_line: + await self._handle_runtime_update(json.loads(data_line)) + event_type = None + data_line = None + except asyncio.CancelledError: + raise + except Exception as e: + for handler in list(self._error_handlers): + await handler(e) + await asyncio.sleep(RECONNECT_DELAY) async def _handle_runtime_update(self, data): if data.get("IsOffline"): diff --git a/tests/test_risco_cloud.py b/tests/test_risco_cloud.py index eb1ed2d..adeb6ff 100644 --- a/tests/test_risco_cloud.py +++ b/tests/test_risco_cloud.py @@ -2,12 +2,33 @@ import unittest from datetime import datetime, timezone from unittest.mock import patch, AsyncMock, MagicMock -from pyrisco.cloud.risco_cloud import RiscoCloud, UnauthorizedError, OperationError, RetryableOperationError +from pyrisco.cloud.risco_cloud import RiscoCloud, UnauthorizedError, OperationError, RetryableOperationError, RECONNECT_DELAY from pyrisco.cloud.alarm import Alarm LOGIN_URL = "https://www.riscocloud.com/webapi/api/auth/login" +def _make_cancel_cm(): + """Return a mock async context manager that raises CancelledError on enter. + + Used to stop the SSE reconnect loop cleanly in tests: set the second + session.get() side_effect entry to _make_cancel_cm() so the loop exits + as soon as it tries to reconnect after the test stream ends. + """ + cm = MagicMock() + cm.__aenter__ = AsyncMock(side_effect=asyncio.CancelledError) + cm.__aexit__ = AsyncMock(return_value=None) + return cm + + +async def _run_sse_task(risco_cloud): + """Await the SSE subscription task, absorbing the CancelledError from reconnect.""" + try: + await risco_cloud._subscription_task + except asyncio.CancelledError: + pass + + def _make_sse_stream(*events): """Build an async iterator that yields SSE lines for the given events. @@ -218,7 +239,7 @@ async def test_subscribe_states_notifies_handler_immediately(self, MockClientSes mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) mock_get_cm.__aexit__ = AsyncMock(return_value=None) - mock_session.get.return_value = mock_get_cm + mock_session.get.side_effect = [mock_get_cm, _make_cancel_cm()] received = [] async def on_state(alarm): @@ -232,7 +253,7 @@ async def on_state(alarm): risco_cloud.add_state_handler(on_state) await risco_cloud.subscribe_states() - await risco_cloud._subscription_task + await _run_sse_task(risco_cloud) self.assertEqual(len(received), 1) self.assertIsInstance(received[0], Alarm) @@ -253,7 +274,7 @@ async def test_subscribe_states_calls_handler(self, MockClientSession, mock_site mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) mock_get_cm.__aexit__ = AsyncMock(return_value=None) - mock_session.get.return_value = mock_get_cm + mock_session.get.side_effect = [mock_get_cm, _make_cancel_cm()] received = [] async def on_state(alarm): @@ -267,7 +288,7 @@ async def on_state(alarm): risco_cloud.add_state_handler(on_state) await risco_cloud.subscribe_states() - await risco_cloud._subscription_task + await _run_sse_task(risco_cloud) self.assertEqual(len(received), 2) # initial fetch + SSE-triggered fetch self.assertIsInstance(received[0], Alarm) @@ -288,7 +309,7 @@ async def test_subscribe_states_no_update_same_timestamp(self, MockClientSession mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) mock_get_cm.__aexit__ = AsyncMock(return_value=None) - mock_session.get.return_value = mock_get_cm + mock_session.get.side_effect = [mock_get_cm, _make_cancel_cm()] risco_cloud = RiscoCloud("username", "password", "pin") risco_cloud._access_token = "mock_access_token" @@ -297,7 +318,7 @@ async def test_subscribe_states_no_update_same_timestamp(self, MockClientSession risco_cloud._session = mock_session await risco_cloud.subscribe_states() - await risco_cloud._subscription_task + await _run_sse_task(risco_cloud) # initial fetch + first SSE message; second SSE message has same timestamp so skipped self.assertEqual(mock_site_post.call_count, 2) @@ -331,7 +352,7 @@ async def test_subscribe_states_skips_fetch_when_offline(self, MockClientSession mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) mock_get_cm.__aexit__ = AsyncMock(return_value=None) - mock_session.get.return_value = mock_get_cm + mock_session.get.side_effect = [mock_get_cm, _make_cancel_cm()] risco_cloud = RiscoCloud("username", "password", "pin") risco_cloud._access_token = "mock_access_token" @@ -340,7 +361,7 @@ async def test_subscribe_states_skips_fetch_when_offline(self, MockClientSession risco_cloud._session = mock_session await risco_cloud.subscribe_states() - await risco_cloud._subscription_task + await _run_sse_task(risco_cloud) # Only the initial fetch; the SSE message is skipped because IsOffline=True self.assertEqual(mock_site_post.call_count, 1) @@ -364,7 +385,7 @@ async def test_subscribe_states_event_handler_fires_without_status_update(self, mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) mock_get_cm.__aexit__ = AsyncMock(return_value=None) - mock_session.get.return_value = mock_get_cm + mock_session.get.side_effect = [mock_get_cm, _make_cancel_cm()] received_events = [] async def on_event(events): @@ -378,16 +399,19 @@ async def on_event(events): risco_cloud.add_event_handler(on_event) await risco_cloud.subscribe_states() - await risco_cloud._subscription_task + await _run_sse_task(risco_cloud) self.assertEqual(len(received_events), 1) self.assertEqual(mock_site_post.call_count, 2) # initial state fetch + event fetch + @patch('pyrisco.cloud.risco_cloud.asyncio.sleep', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') - async def test_subscribe_states_calls_error_handler(self, MockClientSession, mock_site_post): + async def test_subscribe_states_calls_error_handler(self, MockClientSession, mock_site_post, mock_sleep): state_payload = {"partitions": [], "zones": []} mock_site_post.return_value = ({"state": {"status": state_payload}}, False) + # Make sleep raise CancelledError so the loop stops cleanly after the first retry delay + mock_sleep.side_effect = asyncio.CancelledError mock_session = MockClientSession.return_value error = RuntimeError("stream broken") @@ -415,10 +439,14 @@ async def on_error(err): risco_cloud.add_error_handler(on_error) await risco_cloud.subscribe_states() - await risco_cloud._subscription_task + try: + await risco_cloud._subscription_task + except asyncio.CancelledError: + pass self.assertEqual(len(received_errors), 1) self.assertIs(received_errors[0], error) + mock_sleep.assert_awaited_once_with(RECONNECT_DELAY) @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @@ -442,7 +470,7 @@ async def test_subscribe_states_calls_event_handler_with_null_response(self, Moc mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) mock_get_cm.__aexit__ = AsyncMock(return_value=None) - mock_session.get.return_value = mock_get_cm + mock_session.get.side_effect = [mock_get_cm, _make_cancel_cm()] received_events = [] async def on_event(events): @@ -456,7 +484,7 @@ async def on_event(events): risco_cloud.add_event_handler(on_event) await risco_cloud.subscribe_states() - await risco_cloud._subscription_task + await _run_sse_task(risco_cloud) self.assertEqual(len(received_events), 1) self.assertEqual(received_events[0], []) @@ -483,7 +511,7 @@ async def test_subscribe_states_calls_event_handler(self, MockClientSession, moc mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) mock_get_cm.__aexit__ = AsyncMock(return_value=None) - mock_session.get.return_value = mock_get_cm + mock_session.get.side_effect = [mock_get_cm, _make_cancel_cm()] received_events = [] async def on_event(events): @@ -497,7 +525,7 @@ async def on_event(events): risco_cloud.add_event_handler(on_event) await risco_cloud.subscribe_states() - await risco_cloud._subscription_task + await _run_sse_task(risco_cloud) self.assertEqual(len(received_events), 1) self.assertIsInstance(received_events[0], list) @@ -516,7 +544,7 @@ async def test_subscribe_states_no_event_handler_when_no_last_event_updated(self mock_get_cm = MagicMock() mock_get_cm.__aenter__ = AsyncMock(return_value=mock_resp) mock_get_cm.__aexit__ = AsyncMock(return_value=None) - mock_session.get.return_value = mock_get_cm + mock_session.get.side_effect = [mock_get_cm, _make_cancel_cm()] received_events = [] async def on_event(events): @@ -530,7 +558,7 @@ async def on_event(events): risco_cloud.add_event_handler(on_event) await risco_cloud.subscribe_states() - await risco_cloud._subscription_task + await _run_sse_task(risco_cloud) self.assertEqual(len(received_events), 0) self.assertEqual(mock_site_post.call_count, 2) # initial state fetch + state fetch from SSE, no event fetch From 685a59f327cf5aedf59f4083d271843ddf127df9 Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 23:29:50 -0400 Subject: [PATCH 09/19] Exponential backoff and MaxRetriesError for SSE reconnect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the flat 30s delay with exponential backoff: 1s, 2s, 4s, 8s… capped at 60s. After RECONNECT_MAX_ATTEMPTS (5) consecutive failures the loop gives up and calls error handlers one final time with MaxRetriesError wrapping the last exception, then exits. A successful connection resets the attempt counter so transient drops don't accumulate toward the limit. MaxRetriesError is exported from the top-level pyrisco package. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/__init__.py | 2 +- pyrisco/cloud/risco_cloud.py | 16 ++++++++++++--- pyrisco/common.py | 8 ++++++++ tests/test_risco_cloud.py | 40 ++++++++++++++++++++++++++++++++++-- 4 files changed, 60 insertions(+), 6 deletions(-) diff --git a/pyrisco/__init__.py b/pyrisco/__init__.py index 0587c40..6fe01ae 100644 --- a/pyrisco/__init__.py +++ b/pyrisco/__init__.py @@ -1,3 +1,3 @@ from pyrisco.local import RiscoLocal from pyrisco.cloud import RiscoCloud -from .common import CannotConnectError, OperationError, UnauthorizedError +from .common import CannotConnectError, OperationError, UnauthorizedError, MaxRetriesError diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 955a865..a121ed9 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -7,7 +7,7 @@ from .alarm import Alarm from .event import Event -from pyrisco.common import UnauthorizedError, CannotConnectError, OperationError, RetryableOperationError, GROUP_ID_TO_NAME +from pyrisco.common import UnauthorizedError, CannotConnectError, OperationError, RetryableOperationError, MaxRetriesError, GROUP_ID_TO_NAME LOGIN_URL = "https://www.riscocloud.com/webapi/api/auth/login" @@ -23,7 +23,9 @@ NUM_RETRIES = 3 RETRYABLE_RESULT_CODE = 72 -RECONNECT_DELAY = 30 +RECONNECT_INITIAL_DELAY = 1 +RECONNECT_MAX_DELAY = 60 +RECONNECT_MAX_ATTEMPTS = 5 class RiscoCloud: @@ -142,6 +144,7 @@ async def _send_control_command(self, body): return Alarm(self, resp, assumed_control_panel_state) async def _sse_loop(self): + attempt = 0 while True: try: url = SSE_URL % self._site_id @@ -171,12 +174,19 @@ async def _sse_loop(self): await self._handle_runtime_update(json.loads(data_line)) event_type = None data_line = None + attempt = 0 # successful connection — reset backoff for next drop except asyncio.CancelledError: raise except Exception as e: + attempt += 1 + if attempt >= RECONNECT_MAX_ATTEMPTS: + for handler in list(self._error_handlers): + await handler(MaxRetriesError(e)) + return for handler in list(self._error_handlers): await handler(e) - await asyncio.sleep(RECONNECT_DELAY) + delay = min(RECONNECT_INITIAL_DELAY * (2 ** (attempt - 1)), RECONNECT_MAX_DELAY) + await asyncio.sleep(delay) async def _handle_runtime_update(self, data): if data.get("IsOffline"): diff --git a/pyrisco/common.py b/pyrisco/common.py index 5fd0901..1627578 100644 --- a/pyrisco/common.py +++ b/pyrisco/common.py @@ -147,3 +147,11 @@ class OperationError(Exception): class RetryableOperationError(OperationError): """Exception to indicate an error in operation that can be retried and might succeed.""" + + +class MaxRetriesError(Exception): + """Raised when the SSE connection fails and all reconnect attempts are exhausted.""" + + def __init__(self, last_error): + self.last_error = last_error + super().__init__(f"SSE reconnect failed after maximum attempts: {last_error}") diff --git a/tests/test_risco_cloud.py b/tests/test_risco_cloud.py index adeb6ff..d285b36 100644 --- a/tests/test_risco_cloud.py +++ b/tests/test_risco_cloud.py @@ -2,7 +2,8 @@ import unittest from datetime import datetime, timezone from unittest.mock import patch, AsyncMock, MagicMock -from pyrisco.cloud.risco_cloud import RiscoCloud, UnauthorizedError, OperationError, RetryableOperationError, RECONNECT_DELAY +from pyrisco.cloud.risco_cloud import RiscoCloud, UnauthorizedError, OperationError, RetryableOperationError, RECONNECT_INITIAL_DELAY, RECONNECT_MAX_ATTEMPTS +from pyrisco.common import MaxRetriesError from pyrisco.cloud.alarm import Alarm LOGIN_URL = "https://www.riscocloud.com/webapi/api/auth/login" @@ -446,7 +447,7 @@ async def on_error(err): self.assertEqual(len(received_errors), 1) self.assertIs(received_errors[0], error) - mock_sleep.assert_awaited_once_with(RECONNECT_DELAY) + mock_sleep.assert_awaited_once_with(RECONNECT_INITIAL_DELAY) # first attempt: 1s @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @@ -564,5 +565,40 @@ async def on_event(events): self.assertEqual(mock_site_post.call_count, 2) # initial state fetch + state fetch from SSE, no event fetch + @patch('pyrisco.cloud.risco_cloud.asyncio.sleep', new_callable=AsyncMock) + @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') + async def test_subscribe_states_max_retries_exceeded(self, MockClientSession, mock_sleep): + """After RECONNECT_MAX_ATTEMPTS failures the loop gives up and notifies via MaxRetriesError.""" + mock_session = MockClientSession.return_value + error = RuntimeError("connection failed") + mock_session.get.side_effect = error + + received_errors = [] + async def on_error(err): + received_errors.append(err) + + risco_cloud = RiscoCloud("username", "password", "pin") + risco_cloud._access_token = "mock_access_token" + risco_cloud._site_id = "mock_site_id" + risco_cloud._session_id = "mock_session_id" + risco_cloud._session = mock_session + risco_cloud.add_error_handler(on_error) + + await risco_cloud.subscribe_states() + await risco_cloud._subscription_task # completes normally after giving up + + # Error handler called once per attempt: first N-1 with the raw error, last with MaxRetriesError + self.assertEqual(len(received_errors), RECONNECT_MAX_ATTEMPTS) + for i in range(RECONNECT_MAX_ATTEMPTS - 1): + self.assertIs(received_errors[i], error) + self.assertIsInstance(received_errors[-1], MaxRetriesError) + self.assertIs(received_errors[-1].last_error, error) + + # Sleep called with exponential backoff (no sleep on final attempt) + self.assertEqual(mock_sleep.await_count, RECONNECT_MAX_ATTEMPTS - 1) + for i, expected_delay in enumerate([1, 2, 4, 8]): + self.assertEqual(mock_sleep.await_args_list[i].args[0], expected_delay) + + if __name__ == '__main__': unittest.main() From d0140df5b675c7554fe988423b59a7af029639d2 Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 23:31:57 -0400 Subject: [PATCH 10/19] Update README and SSE example for new reconnect behavior - Show add_event_handler usage - Remove manual reconnect from on_error (now handled automatically) - Show MaxRetriesError handling pattern for when retries are exhausted - Document exponential backoff and initial-state-on-subscribe behavior Co-Authored-By: Claude Sonnet 4.6 --- README.md | 24 +++++++++++++++++------- examples/cloud/cloud_sse_example.py | 11 +++++++++-- 2 files changed, 26 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index cb87cc7..1682ad3 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ Pyrisco can subscribe to Risco Cloud's Server-Sent Events stream and push state ```python import asyncio -from pyrisco import RiscoCloud +from pyrisco import RiscoCloud, MaxRetriesError r = RiscoCloud("", "", "") @@ -29,15 +29,23 @@ async def on_state(alarm): print(alarm.partitions[0].armed) print(alarm.zones[0].triggered) +async def on_event(events): + for event in events: + print(event.text) + async def on_error(error): - print(f"SSE error: {error}, reconnecting in 5s...") - await asyncio.sleep(5) - await r.login() - await r.subscribe_states() + if isinstance(error, MaxRetriesError): + # All reconnect attempts exhausted — re-login and re-subscribe + await r.login() + await r.subscribe_states() + else: + # Transient error — pyrisco will reconnect automatically with backoff + print(f"SSE error: {error}") async def main(): await r.login() r.add_state_handler(on_state) + r.add_event_handler(on_event) r.add_error_handler(on_error) await r.subscribe_states() await asyncio.Future() # run forever @@ -45,9 +53,11 @@ async def main(): asyncio.run(main()) ``` -Both `add_state_handler` and `add_error_handler` return a callable that removes the handler when called. +`add_state_handler`, `add_event_handler`, and `add_error_handler` each return a callable that removes the handler when called. + +On connection errors pyrisco reconnects automatically with exponential backoff (1 s, 2 s, 4 s, 8 s…). After the maximum number of attempts the error handler is called with a `MaxRetriesError` wrapping the last exception — the recommended response is to re-login and re-subscribe. -After `subscribe_states()` is started, calls to `get_state()` return the latest cached state without making a network request. +After `subscribe_states()` is started, the state handler is called immediately with the current state, and subsequent calls to `get_state()` return the latest cached state without making a network request. #### Polling diff --git a/examples/cloud/cloud_sse_example.py b/examples/cloud/cloud_sse_example.py index ddd1949..6ad07fe 100644 --- a/examples/cloud/cloud_sse_example.py +++ b/examples/cloud/cloud_sse_example.py @@ -1,5 +1,5 @@ import asyncio -from pyrisco import RiscoCloud +from pyrisco import RiscoCloud, MaxRetriesError risco = RiscoCloud("user@example.com", "password", "1234") @@ -19,7 +19,14 @@ async def on_event(events): async def on_error(error): - print(f"SSE error: {error} (will reconnect automatically)") + if isinstance(error, MaxRetriesError): + # All reconnect attempts exhausted — re-login and re-subscribe + print(f"Gave up reconnecting ({error.last_error}), re-logging in...") + await risco.login() + await risco.subscribe_states() + else: + # Transient error — pyrisco will reconnect automatically with backoff + print(f"SSE error: {error} (reconnecting automatically)") async def main(): From 34327fb0fe22fb7b6f1e4407a851c09969fe20b5 Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 23:41:16 -0400 Subject: [PATCH 11/19] Fix self-cancellation deadlock and tight reconnect on clean EOF close() now checks asyncio.current_task() before cancelling/awaiting _subscription_task, so calling close() from within the subscription task (e.g. via UnauthorizedError in _site_post) no longer deadlocks. Add a RECONNECT_INITIAL_DELAY sleep after clean stream EOF before the next connect attempt, preventing a tight reconnect loop when the server cycles connections normally. Mock asyncio.sleep at the TestRiscoCloud class level so reconnect delays never slow the test suite, regardless of which code path is exercised. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 12 +++++++----- tests/test_risco_cloud.py | 22 ++++++++++++++-------- 2 files changed, 21 insertions(+), 13 deletions(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index a121ed9..6728574 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -175,6 +175,7 @@ async def _sse_loop(self): event_type = None data_line = None attempt = 0 # successful connection — reset backoff for next drop + await asyncio.sleep(RECONNECT_INITIAL_DELAY) # small delay before reconnecting on clean EOF except asyncio.CancelledError: raise except Exception as e: @@ -215,11 +216,12 @@ async def close(self): """Close the connection.""" self._session_id = None if self._subscription_task: - self._subscription_task.cancel() - try: - await self._subscription_task - except (asyncio.CancelledError, Exception): - pass + if asyncio.current_task() != self._subscription_task: + self._subscription_task.cancel() + try: + await self._subscription_task + except (asyncio.CancelledError, Exception): + pass self._subscription_task = None if self._created_session and self._session is not None: await self._session.close() diff --git a/tests/test_risco_cloud.py b/tests/test_risco_cloud.py index d285b36..4e55832 100644 --- a/tests/test_risco_cloud.py +++ b/tests/test_risco_cloud.py @@ -51,6 +51,14 @@ async def _iter(): class TestRiscoCloud(unittest.IsolatedAsyncioTestCase): + def setUp(self): + # Mock asyncio.sleep globally for all tests so SSE reconnect delays don't slow the suite. + self._sleep_patcher = patch('pyrisco.cloud.risco_cloud.asyncio.sleep', new_callable=AsyncMock) + self.mock_sleep = self._sleep_patcher.start() + + def tearDown(self): + self._sleep_patcher.stop() + @patch('pyrisco.cloud.risco_cloud.RiscoCloud._authenticated_post', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') async def test_login(self, MockClientSession, mock_authenticated_post): @@ -405,14 +413,13 @@ async def on_event(events): self.assertEqual(len(received_events), 1) self.assertEqual(mock_site_post.call_count, 2) # initial state fetch + event fetch - @patch('pyrisco.cloud.risco_cloud.asyncio.sleep', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') - async def test_subscribe_states_calls_error_handler(self, MockClientSession, mock_site_post, mock_sleep): + async def test_subscribe_states_calls_error_handler(self, MockClientSession, mock_site_post): state_payload = {"partitions": [], "zones": []} mock_site_post.return_value = ({"state": {"status": state_payload}}, False) # Make sleep raise CancelledError so the loop stops cleanly after the first retry delay - mock_sleep.side_effect = asyncio.CancelledError + self.mock_sleep.side_effect = asyncio.CancelledError mock_session = MockClientSession.return_value error = RuntimeError("stream broken") @@ -447,7 +454,7 @@ async def on_error(err): self.assertEqual(len(received_errors), 1) self.assertIs(received_errors[0], error) - mock_sleep.assert_awaited_once_with(RECONNECT_INITIAL_DELAY) # first attempt: 1s + self.mock_sleep.assert_awaited_once_with(RECONNECT_INITIAL_DELAY) # first attempt: 1s @patch('pyrisco.cloud.risco_cloud.RiscoCloud._site_post', new_callable=AsyncMock) @@ -565,9 +572,8 @@ async def on_event(events): self.assertEqual(mock_site_post.call_count, 2) # initial state fetch + state fetch from SSE, no event fetch - @patch('pyrisco.cloud.risco_cloud.asyncio.sleep', new_callable=AsyncMock) @patch('pyrisco.cloud.risco_cloud.aiohttp.ClientSession') - async def test_subscribe_states_max_retries_exceeded(self, MockClientSession, mock_sleep): + async def test_subscribe_states_max_retries_exceeded(self, MockClientSession): """After RECONNECT_MAX_ATTEMPTS failures the loop gives up and notifies via MaxRetriesError.""" mock_session = MockClientSession.return_value error = RuntimeError("connection failed") @@ -595,9 +601,9 @@ async def on_error(err): self.assertIs(received_errors[-1].last_error, error) # Sleep called with exponential backoff (no sleep on final attempt) - self.assertEqual(mock_sleep.await_count, RECONNECT_MAX_ATTEMPTS - 1) + self.assertEqual(self.mock_sleep.await_count, RECONNECT_MAX_ATTEMPTS - 1) for i, expected_delay in enumerate([1, 2, 4, 8]): - self.assertEqual(mock_sleep.await_args_list[i].args[0], expected_delay) + self.assertEqual(self.mock_sleep.await_args_list[i].args[0], expected_delay) if __name__ == '__main__': From f6dfac97f976023ba83ca7bd66ecea1b0c5ca94f Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 23:46:19 -0400 Subject: [PATCH 12/19] Always cancel subscription task in close(), only skip self-await cancel() must always be called so the loop actually stops; previously skipping it when called from within the task meant the loop kept running after close(). Now cancel() is unconditional and only the await is guarded by the current_task() check. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 6728574..8b2d1d1 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -216,8 +216,8 @@ async def close(self): """Close the connection.""" self._session_id = None if self._subscription_task: + self._subscription_task.cancel() if asyncio.current_task() != self._subscription_task: - self._subscription_task.cancel() try: await self._subscription_task except (asyncio.CancelledError, Exception): From 33365cba9ac96abb5c6f60bf925b0548f720796d Mon Sep 17 00:00:00 2001 From: On Freund Date: Wed, 27 May 2026 23:57:19 -0400 Subject: [PATCH 13/19] Fix timestamp parsing, re-login pattern, and close() cancel order MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add _parse_timestamp() helper that normalises trailing 'Z' to '+00:00' before calling fromisoformat(), making the code work on Python < 3.11 and removing the need for callers to know about the format difference - In README and example, schedule re-login via asyncio.create_task() so close() is called from a new task rather than from within the running SSE task — calling close() from within the task schedules CancelledError on itself, which would abort login() before it could complete Co-Authored-By: Claude Sonnet 4.6 --- README.md | 11 ++++++++--- examples/cloud/cloud_sse_example.py | 15 +++++++++++---- pyrisco/cloud/risco_cloud.py | 11 ++++++++--- 3 files changed, 27 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 1682ad3..5d45c44 100644 --- a/README.md +++ b/README.md @@ -33,11 +33,16 @@ async def on_event(events): for event in events: print(event.text) +async def _restart(): + await r.close() + await r.login() + await r.subscribe_states() + async def on_error(error): if isinstance(error, MaxRetriesError): - # All reconnect attempts exhausted — re-login and re-subscribe - await r.login() - await r.subscribe_states() + # Schedule restart outside the SSE task — calling close() from within + # the task would cancel it before login() could complete + asyncio.create_task(_restart()) else: # Transient error — pyrisco will reconnect automatically with backoff print(f"SSE error: {error}") diff --git a/examples/cloud/cloud_sse_example.py b/examples/cloud/cloud_sse_example.py index 6ad07fe..87df36e 100644 --- a/examples/cloud/cloud_sse_example.py +++ b/examples/cloud/cloud_sse_example.py @@ -18,12 +18,19 @@ async def on_event(events): print(f" [{event.time}] {event.text}") +async def _restart(): + """Re-login and re-subscribe after all reconnect attempts are exhausted.""" + await risco.close() + await risco.login() + await risco.subscribe_states() + + async def on_error(error): if isinstance(error, MaxRetriesError): - # All reconnect attempts exhausted — re-login and re-subscribe - print(f"Gave up reconnecting ({error.last_error}), re-logging in...") - await risco.login() - await risco.subscribe_states() + # All reconnect attempts exhausted — schedule a fresh login outside the + # current SSE task (calling close() from within the task would cancel it) + print(f"Gave up reconnecting ({error.last_error}), restarting...") + asyncio.create_task(_restart()) else: # Transient error — pyrisco will reconnect automatically with backoff print(f"SSE error: {error} (reconnecting automatically)") diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 8b2d1d1..28939f0 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -10,6 +10,11 @@ from pyrisco.common import UnauthorizedError, CannotConnectError, OperationError, RetryableOperationError, MaxRetriesError, GROUP_ID_TO_NAME +def _parse_timestamp(ts_str): + """Parse an ISO 8601 timestamp, accepting both 'Z' and '+00:00' UTC suffixes.""" + return datetime.fromisoformat(ts_str.replace("Z", "+00:00")) + + LOGIN_URL = "https://www.riscocloud.com/webapi/api/auth/login" SITE_URL = "https://www.riscocloud.com/webapi/api/wuws/site/GetAll" PIN_URL = "https://www.riscocloud.com/webapi/api/wuws/site/%s/Login" @@ -194,7 +199,7 @@ async def _handle_runtime_update(self, data): return ts_str = data.get("LastStatusUpdate") if ts_str: - update_time = datetime.fromisoformat(ts_str) + update_time = _parse_timestamp(ts_str) if self._last_status_update is None or update_time > self._last_status_update: resp, assumed = await self._site_post(STATE_URL, {}) alarm = Alarm(self, resp["state"]["status"], assumed) @@ -204,8 +209,8 @@ async def _handle_runtime_update(self, data): await handler(alarm) event_ts_str = data.get("LastEventUpdated") if event_ts_str and self._event_handlers: - event_time = datetime.fromisoformat(event_ts_str) - last_event_time = datetime.fromisoformat(self._last_event_update) if self._last_event_update else None + event_time = _parse_timestamp(event_ts_str) + last_event_time = _parse_timestamp(self._last_event_update) if self._last_event_update else None if last_event_time is None or event_time > last_event_time: events = await self.get_events(self._last_event_update) self._last_event_update = event_ts_str From cce5e604639c11924d2ae4eda6c6396ae69c7457 Mon Sep 17 00:00:00 2001 From: On Freund Date: Thu, 28 May 2026 00:04:47 -0400 Subject: [PATCH 14/19] Validate SSE response status and fix close() self-cancel - Call resp.raise_for_status() right after opening the SSE connection so 4xx/5xx responses are treated as errors and go through the backoff/retry path rather than silently appearing as a clean EOF that resets attempt=0 - Revert close() to skip cancel() (not just await) when called from within the subscription task; calling cancel() on the current task schedules CancelledError at the next await, which can abort session.close() and other cleanup steps mid-flight. When close() is called from inside the task (e.g. UnauthorizedError recovery), the loop stops naturally because the session is closed beneath it. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 28939f0..2fdab31 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -160,6 +160,7 @@ async def _sse_loop(self): } params = {"sessionToken": self._session_id} async with self._session.get(url, headers=headers, params=params) as resp: + resp.raise_for_status() # SSE connection is now open — fetch initial state before consuming # messages so no state changes in between can be missed. initial_resp, assumed = await self._site_post(STATE_URL, {}) @@ -221,8 +222,8 @@ async def close(self): """Close the connection.""" self._session_id = None if self._subscription_task: - self._subscription_task.cancel() if asyncio.current_task() != self._subscription_task: + self._subscription_task.cancel() try: await self._subscription_task except (asyncio.CancelledError, Exception): From 91c824f5223092212c0180e5f941f2a9a6cb3d89 Mon Sep 17 00:00:00 2001 From: On Freund Date: Thu, 28 May 2026 07:24:25 -0400 Subject: [PATCH 15/19] Fix close() self-cancel by extracting _relogin() The real root cause of the close()-from-within-task problem was that _site_post's UnauthorizedError recovery called close() + login() from inside the SSE loop. close() cancels the subscription task, which means any subsequent await (including session.close()) could receive CancelledError and abort cleanup. Fix: extract _relogin() which refreshes credentials in-place (clears session_id, re-runs the three login steps) without touching the session or the SSE task. _site_post now calls _relogin() instead of close()+login() on UnauthorizedError. close() is now simple and correct: always cancel+await, no current_task check needed. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 2fdab31..43374c4 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -103,8 +103,7 @@ async def _site_post(self, url, body): if isinstance(e, RetryableOperationError): from_control_panel = False elif isinstance(e, UnauthorizedError): - await self.close() - await self.login() + await self._relogin() async def _login_user_pass(self): headers = {"Content-Type": "application/json", "User-Agent": "pyrisco/1.0"} @@ -144,6 +143,13 @@ async def _init_session(self, session): else: self._session = session + async def _relogin(self): + """Refresh credentials in-place without closing the session or the SSE task.""" + self._session_id = None + await self._login_user_pass() + await self._login_site() + await self._login_session() + async def _send_control_command(self, body): resp, assumed_control_panel_state = await self._site_post(CONTROL_URL, body) return Alarm(self, resp, assumed_control_panel_state) @@ -222,12 +228,11 @@ async def close(self): """Close the connection.""" self._session_id = None if self._subscription_task: - if asyncio.current_task() != self._subscription_task: - self._subscription_task.cancel() - try: - await self._subscription_task - except (asyncio.CancelledError, Exception): - pass + self._subscription_task.cancel() + try: + await self._subscription_task + except (asyncio.CancelledError, Exception): + pass self._subscription_task = None if self._created_session and self._session is not None: await self._session.close() From e690065d034ec3f9f19d93aa24d48c1631411a8b Mon Sep 17 00:00:00 2001 From: On Freund Date: Thu, 28 May 2026 18:52:09 -0400 Subject: [PATCH 16/19] Simplify reconnect: drop _relogin(), fix attempt reset, clarify comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SSE stream drops for network reasons — the session token stays valid, so UnauthorizedError never fires from within _sse_loop in practice. All the _relogin() / close()-self-cancel complexity was solving a problem that doesn't occur. - Remove _relogin(); restore _site_post to close()+login() for the external arm/disarm case where 401 can legitimately happen - Fix attempt reset: move it back to after clean EOF (not after raise_for_status). With the early reset, post-connect failures (e.g. initial state fetch) always saw attempt=0 and MaxRetriesError could never be reached - Fix README and example comment: the real risk of calling close() from within the SSE error handler is a deadlock (close() awaits the running task itself), not that cancel() fires before login() completes Co-Authored-By: Claude Sonnet 4.6 --- README.md | 4 ++-- examples/cloud/cloud_sse_example.py | 5 +++-- pyrisco/cloud/risco_cloud.py | 14 ++++---------- 3 files changed, 9 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 5d45c44..c9fea1b 100644 --- a/README.md +++ b/README.md @@ -40,8 +40,8 @@ async def _restart(): async def on_error(error): if isinstance(error, MaxRetriesError): - # Schedule restart outside the SSE task — calling close() from within - # the task would cancel it before login() could complete + # Schedule restart in a new task — calling close() from within the SSE + # task would deadlock because close() awaits the task itself asyncio.create_task(_restart()) else: # Transient error — pyrisco will reconnect automatically with backoff diff --git a/examples/cloud/cloud_sse_example.py b/examples/cloud/cloud_sse_example.py index 87df36e..10c51af 100644 --- a/examples/cloud/cloud_sse_example.py +++ b/examples/cloud/cloud_sse_example.py @@ -27,8 +27,9 @@ async def _restart(): async def on_error(error): if isinstance(error, MaxRetriesError): - # All reconnect attempts exhausted — schedule a fresh login outside the - # current SSE task (calling close() from within the task would cancel it) + # All reconnect attempts exhausted — schedule restart in a new task; + # calling close() from within the SSE task would deadlock because + # close() awaits the task itself print(f"Gave up reconnecting ({error.last_error}), restarting...") asyncio.create_task(_restart()) else: diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 43374c4..7f4594b 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -103,7 +103,8 @@ async def _site_post(self, url, body): if isinstance(e, RetryableOperationError): from_control_panel = False elif isinstance(e, UnauthorizedError): - await self._relogin() + await self.close() + await self.login() async def _login_user_pass(self): headers = {"Content-Type": "application/json", "User-Agent": "pyrisco/1.0"} @@ -143,13 +144,6 @@ async def _init_session(self, session): else: self._session = session - async def _relogin(self): - """Refresh credentials in-place without closing the session or the SSE task.""" - self._session_id = None - await self._login_user_pass() - await self._login_site() - await self._login_session() - async def _send_control_command(self, body): resp, assumed_control_panel_state = await self._site_post(CONTROL_URL, body) return Alarm(self, resp, assumed_control_panel_state) @@ -186,8 +180,8 @@ async def _sse_loop(self): await self._handle_runtime_update(json.loads(data_line)) event_type = None data_line = None - attempt = 0 # successful connection — reset backoff for next drop - await asyncio.sleep(RECONNECT_INITIAL_DELAY) # small delay before reconnecting on clean EOF + attempt = 0 # clean stream end — reset backoff for next drop + await asyncio.sleep(RECONNECT_INITIAL_DELAY) # small delay before reconnecting except asyncio.CancelledError: raise except Exception as e: From 52a2943538f90134d0d2cff2c9a9273df378e7e4 Mon Sep 17 00:00:00 2001 From: On Freund Date: Thu, 28 May 2026 19:09:31 -0400 Subject: [PATCH 17/19] Fix attempt reset timing and close() self-await deadlock - Move attempt=0 to after initial state fetch succeeds (right before the message loop). Previously reset at clean EOF meant that if TCP connected but the state fetch always failed, attempt never accumulated past 1 and MaxRetriesError could never fire. Now reset requires the connection to reach a usable state (SSE up + initial state fetched). - Add current_task() guard in close() so it always calls cancel() (to stop the loop) but skips await when called from within the subscription task itself, preventing a self-await deadlock. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index 7f4594b..e1d142a 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -168,6 +168,7 @@ async def _sse_loop(self): self._latest_state = alarm for handler in list(self._state_handlers): await handler(alarm) + attempt = 0 # usable connection established — reset backoff counter event_type = None data_line = None async for line_bytes in resp.content: @@ -180,8 +181,7 @@ async def _sse_loop(self): await self._handle_runtime_update(json.loads(data_line)) event_type = None data_line = None - attempt = 0 # clean stream end — reset backoff for next drop - await asyncio.sleep(RECONNECT_INITIAL_DELAY) # small delay before reconnecting + await asyncio.sleep(RECONNECT_INITIAL_DELAY) # small delay before reconnecting on clean EOF except asyncio.CancelledError: raise except Exception as e: @@ -223,10 +223,11 @@ async def close(self): self._session_id = None if self._subscription_task: self._subscription_task.cancel() - try: - await self._subscription_task - except (asyncio.CancelledError, Exception): - pass + if asyncio.current_task() != self._subscription_task: + try: + await self._subscription_task + except (asyncio.CancelledError, Exception): + pass self._subscription_task = None if self._created_session and self._session is not None: await self._session.close() From a754d83f5848ce4ee78048b109b74774e29ab2b4 Mon Sep 17 00:00:00 2001 From: On Freund Date: Thu, 28 May 2026 19:33:22 -0400 Subject: [PATCH 18/19] Isolate handler exceptions from the SSE loop via _call_handlers Add _call_handlers() static method that swallows per-handler exceptions so a buggy or raising callback doesn't propagate into the SSE loop and trigger a spurious reconnect/backoff. Pattern mirrors pyrisco/local's _call_handlers. All state/event handler invocations now go through it. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 18 ++++++++++++------ tests/test_risco_cloud.py | 3 ++- 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index e1d142a..ca81e95 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -64,6 +64,15 @@ def _remove(): handlers.remove(handler) return _remove + @staticmethod + async def _call_handlers(handlers, *params): + """Call handlers, swallowing exceptions so a buggy callback doesn't break the SSE loop.""" + for handler in list(handlers): + try: + await handler(*params) + except Exception: + pass + async def _authenticated_post(self, url, body): headers = { "Content-Type": "application/json", @@ -166,8 +175,7 @@ async def _sse_loop(self): initial_resp, assumed = await self._site_post(STATE_URL, {}) alarm = Alarm(self, initial_resp["state"]["status"], assumed) self._latest_state = alarm - for handler in list(self._state_handlers): - await handler(alarm) + await RiscoCloud._call_handlers(self._state_handlers, alarm) attempt = 0 # usable connection established — reset backoff counter event_type = None data_line = None @@ -206,8 +214,7 @@ async def _handle_runtime_update(self, data): alarm = Alarm(self, resp["state"]["status"], assumed) self._last_status_update = update_time self._latest_state = alarm - for handler in list(self._state_handlers): - await handler(alarm) + await RiscoCloud._call_handlers(self._state_handlers, alarm) event_ts_str = data.get("LastEventUpdated") if event_ts_str and self._event_handlers: event_time = _parse_timestamp(event_ts_str) @@ -215,8 +222,7 @@ async def _handle_runtime_update(self, data): if last_event_time is None or event_time > last_event_time: events = await self.get_events(self._last_event_update) self._last_event_update = event_ts_str - for handler in list(self._event_handlers): - await handler(events) + await RiscoCloud._call_handlers(self._event_handlers, events) async def close(self): """Close the connection.""" diff --git a/tests/test_risco_cloud.py b/tests/test_risco_cloud.py index 4e55832..4eca024 100644 --- a/tests/test_risco_cloud.py +++ b/tests/test_risco_cloud.py @@ -23,11 +23,12 @@ def _make_cancel_cm(): async def _run_sse_task(risco_cloud): - """Await the SSE subscription task, absorbing the CancelledError from reconnect.""" + """Await the SSE subscription task, then yield so handler tasks can run.""" try: await risco_cloud._subscription_task except asyncio.CancelledError: pass + await asyncio.sleep(0) # let any pending tasks settle def _make_sse_stream(*events): From 583c29d2628fcde0b5658ddb0eeee1328c8349e3 Mon Sep 17 00:00:00 2001 From: On Freund Date: Thu, 28 May 2026 23:26:43 -0400 Subject: [PATCH 19/19] Match local _call_handlers pattern: create_task + gather MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Switch _call_handlers to the same non-blocking dispatch as pyrisco/local/risco_local.py: a synchronous method that schedules asyncio.create_task(_gather()) so handler calls don't block the SSE reader. return_exceptions=True on gather means handler exceptions are captured, not propagated. Apply to error handlers too for consistency — a raising error handler no longer risks aborting the reconnect sleep or MaxRetriesError return. Tests drain pending handler tasks after the subscription task ends via asyncio.gather(*all_tasks) so assertions remain deterministic. Co-Authored-By: Claude Sonnet 4.6 --- pyrisco/cloud/risco_cloud.py | 25 +++++++++++-------------- tests/test_risco_cloud.py | 13 +++++++++++-- 2 files changed, 22 insertions(+), 16 deletions(-) diff --git a/pyrisco/cloud/risco_cloud.py b/pyrisco/cloud/risco_cloud.py index ca81e95..4fa8413 100644 --- a/pyrisco/cloud/risco_cloud.py +++ b/pyrisco/cloud/risco_cloud.py @@ -65,13 +65,12 @@ def _remove(): return _remove @staticmethod - async def _call_handlers(handlers, *params): - """Call handlers, swallowing exceptions so a buggy callback doesn't break the SSE loop.""" - for handler in list(handlers): - try: - await handler(*params) - except Exception: - pass + def _call_handlers(handlers, *params): + """Dispatch handlers asynchronously so a slow/buggy callback doesn't block the SSE loop.""" + if handlers: + async def _gather(): + await asyncio.gather(*[h(*params) for h in list(handlers)], return_exceptions=True) + asyncio.create_task(_gather()) async def _authenticated_post(self, url, body): headers = { @@ -175,7 +174,7 @@ async def _sse_loop(self): initial_resp, assumed = await self._site_post(STATE_URL, {}) alarm = Alarm(self, initial_resp["state"]["status"], assumed) self._latest_state = alarm - await RiscoCloud._call_handlers(self._state_handlers, alarm) + RiscoCloud._call_handlers(self._state_handlers, alarm) attempt = 0 # usable connection established — reset backoff counter event_type = None data_line = None @@ -195,11 +194,9 @@ async def _sse_loop(self): except Exception as e: attempt += 1 if attempt >= RECONNECT_MAX_ATTEMPTS: - for handler in list(self._error_handlers): - await handler(MaxRetriesError(e)) + RiscoCloud._call_handlers(self._error_handlers, MaxRetriesError(e)) return - for handler in list(self._error_handlers): - await handler(e) + RiscoCloud._call_handlers(self._error_handlers, e) delay = min(RECONNECT_INITIAL_DELAY * (2 ** (attempt - 1)), RECONNECT_MAX_DELAY) await asyncio.sleep(delay) @@ -214,7 +211,7 @@ async def _handle_runtime_update(self, data): alarm = Alarm(self, resp["state"]["status"], assumed) self._last_status_update = update_time self._latest_state = alarm - await RiscoCloud._call_handlers(self._state_handlers, alarm) + RiscoCloud._call_handlers(self._state_handlers, alarm) event_ts_str = data.get("LastEventUpdated") if event_ts_str and self._event_handlers: event_time = _parse_timestamp(event_ts_str) @@ -222,7 +219,7 @@ async def _handle_runtime_update(self, data): if last_event_time is None or event_time > last_event_time: events = await self.get_events(self._last_event_update) self._last_event_update = event_ts_str - await RiscoCloud._call_handlers(self._event_handlers, events) + RiscoCloud._call_handlers(self._event_handlers, events) async def close(self): """Close the connection.""" diff --git a/tests/test_risco_cloud.py b/tests/test_risco_cloud.py index 4eca024..14d6a47 100644 --- a/tests/test_risco_cloud.py +++ b/tests/test_risco_cloud.py @@ -22,13 +22,20 @@ def _make_cancel_cm(): return cm +async def _drain_handler_tasks(): + """Wait for all handler tasks spawned by _call_handlers to finish.""" + pending = {t for t in asyncio.all_tasks() if t != asyncio.current_task()} + if pending: + await asyncio.gather(*pending, return_exceptions=True) + + async def _run_sse_task(risco_cloud): - """Await the SSE subscription task, then yield so handler tasks can run.""" + """Await the SSE subscription task, then drain any handler tasks.""" try: await risco_cloud._subscription_task except asyncio.CancelledError: pass - await asyncio.sleep(0) # let any pending tasks settle + await _drain_handler_tasks() def _make_sse_stream(*events): @@ -452,6 +459,7 @@ async def on_error(err): await risco_cloud._subscription_task except asyncio.CancelledError: pass + await _drain_handler_tasks() self.assertEqual(len(received_errors), 1) self.assertIs(received_errors[0], error) @@ -593,6 +601,7 @@ async def on_error(err): await risco_cloud.subscribe_states() await risco_cloud._subscription_task # completes normally after giving up + await _drain_handler_tasks() # Error handler called once per attempt: first N-1 with the raw error, last with MaxRetriesError self.assertEqual(len(received_errors), RECONNECT_MAX_ATTEMPTS)