add security event handling (lbbrhzn#634)

matejdro · Oct 12, 2022 · ea5cba0 · ea5cba0
1 parent 62e4d6a
commit ea5cba0
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 0 deletions.
diff --git a/custom_components/ocpp/api.py b/custom_components/ocpp/api.py
@@ -1219,6 +1219,20 @@ def on_diagnostics_status(self, status, **kwargs):
         )
         return call_result.DiagnosticsStatusNotificationPayload()
 
+    @on(Action.SecurityEventNotification)
+    def on_security_event(self, type, timestamp, **kwargs):
+        """Handle security event notification."""
+        _LOGGER.info(
+            "Security event notification received: %s at %s [techinfo: %s]",
+            type,
+            timestamp,
+            kwargs.get(om.tech_info.name, "none"),
+        )
+        self.hass.async_create_task(
+            self.notify_ha(f"Security event notification received: {type}")
+        )
+        return call_result.SecurityEventNotificationPayload()
+
     def get_authorization_status(self, id_tag):
         """Get the authorization status for an id_tag."""
         # get the domain wide configuration

diff --git a/custom_components/ocpp/enums.py b/custom_components/ocpp/enums.py
@@ -106,6 +106,7 @@ class OcppMisc(str, Enum):
     feature_profile_reservation = "Reservation"
     feature_profile_remote = "RemoteTrigger"
     feature_profile_auth = "LocalAuthListManagement"
+    tech_info = "techInfo"
 
     # for use with Smart Charging
     current = "Current"

diff --git a/tests/test_charge_point.py b/tests/test_charge_point.py
@@ -242,6 +242,7 @@ async def test_services(hass, socket_enabled):
                     cp.send_authorize(),
                     cp.send_heartbeat(),
                     cp.send_status_notification(),
+                    cp.send_security_event(),
                     cp.send_firmware_status(),
                     cp.send_data_transfer(),
                     cp.send_start_transaction(),
@@ -877,3 +878,12 @@ async def send_stop_transaction(self, delay: int = 0):
         )
         resp = await self.call(request)
         assert resp.id_tag_info["status"] == AuthorizationStatus.accepted.value
+
+    async def send_security_event(self):
+        """Send a security event notification."""
+        request = call.SecurityEventNotificationPayload(
+            type="SettingSystemTime",
+            timestamp="2022-09-29T20:58:29Z",
+            tech_info="BootNotification",
+        )
+        await self.call(request)