diff --git a/at_client/connections/atmonitorconnection.py b/at_client/connections/atmonitorconnection.py index 4befdc9..a0d61ca 100644 --- a/at_client/connections/atmonitorconnection.py +++ b/at_client/connections/atmonitorconnection.py @@ -113,7 +113,16 @@ def stop_monitor(self): self._last_heartbeat_sent_time = self._last_heartbeat_ack_time = TimeUtil.current_time_millis() self.disconnect() - + + @staticmethod + def _is_shared_key_notification(atsign, key): + """True if `key` is an incoming shared-key notification for `atsign`. + + e.g. "@alice:shared_key@bob" when we (alice) are the receiver. Uses + to_string() — calling the bound method, not stringifying it. + """ + return key.startswith(atsign.to_string() + ":shared_key@") + def _run(self): what = "" first = True @@ -164,7 +173,7 @@ def _run(self): if uuid == "-1": event_type = AtEventType.STATS_NOTIFICATION elif operation == "update": - if key.startswith(str(self.atsign.to_string) + ":shared_key@"): + if self._is_shared_key_notification(self.atsign, key): event_type = AtEventType.SHARED_KEY_NOTIFICATION else: event_type = AtEventType.UPDATE_NOTIFICATION diff --git a/test/monitor_shared_key_test.py b/test/monitor_shared_key_test.py new file mode 100644 index 0000000..f0bdb48 --- /dev/null +++ b/test/monitor_shared_key_test.py @@ -0,0 +1,27 @@ +import unittest + +from at_client.common import AtSign +from at_client.connections.atmonitorconnection import AtMonitorConnection + + +class MonitorSharedKeyDetectionTest(unittest.TestCase): + """Network-free test for shared-key notification classification. + + Regression for the missing () in `str(self.atsign.to_string)`, which made the + check compare against a bound-method repr and never match, so shared-key + notifications were mis-typed as ordinary UPDATE notifications. + """ + + def test_detects_incoming_shared_key_notification(self): + me = AtSign("@alice") + key = "@alice:shared_key@bob" # bob sharing his key with me (alice) + self.assertTrue(AtMonitorConnection._is_shared_key_notification(me, key)) + + def test_ignores_regular_update_notification(self): + me = AtSign("@alice") + key = "@alice:live_traffic.demo@bob" + self.assertFalse(AtMonitorConnection._is_shared_key_notification(me, key)) + + +if __name__ == "__main__": + unittest.main()