switch to new websockets asyncio (#1439)

* switch to new websockets asyncio

* add errors raised

* fix error raise

* extra detail on nosub test

* add test for autoconfig

* remove old tests

* update deprecated enums in v16 tests

* extra test for reboot required

* update pre-commit

* fix key

---------

Co-authored-by: lbbrhzn <[email protected]>

.pre-commit-config.yaml

@@ -12,7 +12,7 @@ repos:
       # Run the formatter.
       - id: ruff-format
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.5.0
+    rev: v5.0.0
     hooks:
       - id: check-executables-have-shebangs
         stages: [manual]

custom_components/ocpp/api.py

@@ -8,9 +8,9 @@
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import STATE_OK
 from homeassistant.core import HomeAssistant
-from websockets import Subprotocol
-import websockets.protocol
+from websockets import Subprotocol, NegotiationError
 import websockets.server
+from websockets.asyncio.server import ServerConnection
 
 from .chargepoint import CentralSystemSettings
 from .ocppv16 import ChargePoint as ChargePointv16
@@ -21,7 +21,6 @@
     CONF_CSID,
     CONF_HOST,
     CONF_PORT,
-    CONF_SKIP_SCHEMA_VALIDATION,
     CONF_SSL,
     CONF_SSL_CERTFILE_PATH,
     CONF_SSL_KEYFILE_PATH,
@@ -34,7 +33,6 @@
     DEFAULT_CSID,
     DEFAULT_HOST,
     DEFAULT_PORT,
-    DEFAULT_SKIP_SCHEMA_VALIDATION,
     DEFAULT_SSL,
     DEFAULT_SSL_CERTFILE_PATH,
     DEFAULT_SSL_KEYFILE_PATH,
@@ -112,10 +110,11 @@ async def create(hass: HomeAssistant, entry: ConfigEntry):
         """Create instance and start listening for OCPP connections on given port."""
         self = CentralSystem(hass, entry)
 
-        server = await websockets.server.serve(
+        server = await websockets.serve(
             self.on_connect,
             self.host,
             self.port,
+            select_subprotocol=self.select_subprotocol,
             subprotocols=self.subprotocols,
             ping_interval=None,  # ping interval is not used here, because we send pings mamually in ChargePoint.monitor_connection()
             ping_timeout=None,
@@ -125,27 +124,38 @@ async def create(hass: HomeAssistant, entry: ConfigEntry):
         self._server = server
         return self
 
-    async def on_connect(self, websocket: websockets.server.WebSocketServerProtocol):
+    def select_subprotocol(
+        self, connection: ServerConnection, subprotocols
+    ) -> Subprotocol | None:
+        """Override default subprotocol selection."""
+
+        # Server offers at least one subprotocol but client doesn't offer any.
+        # Default to None
+        if not subprotocols:
+            return None
+
+        # Server and client both offer subprotocols. Look for a shared one.
+        proposed_subprotocols = set(subprotocols)
+        for subprotocol in proposed_subprotocols:
+            if subprotocol in self.subprotocols:
+                return subprotocol
+
+        # No common subprotocol was found.
+        raise NegotiationError(
+            "invalid subprotocol; expected one of " + ", ".join(self.subprotocols)
+        )
+
+    async def on_connect(self, websocket: ServerConnection):
         """Request handler executed for every new OCPP connection."""
-        if self.config.get(CONF_SKIP_SCHEMA_VALIDATION, DEFAULT_SKIP_SCHEMA_VALIDATION):
-            _LOGGER.warning("Skipping websocket subprotocol validation")
+        if websocket.subprotocol is not None:
+            _LOGGER.info("Websocket Subprotocol matched: %s", websocket.subprotocol)
         else:
-            if websocket.subprotocol is not None:
-                _LOGGER.info("Websocket Subprotocol matched: %s", websocket.subprotocol)
-            else:
-                # In the websockets lib if no subprotocols are supported by the
-                # client and the server, it proceeds without a subprotocol,
-                # so we have to manually close the connection.
-                _LOGGER.warning(
-                    "Protocols mismatched | expected Subprotocols: %s,"
-                    " but client supports  %s | Closing connection",
-                    websocket.available_subprotocols,
-                    websocket.request_headers.get("Sec-WebSocket-Protocol", ""),
-                )
-                return await websocket.close()
+            _LOGGER.info(
+                "Websocket Subprotocol not provided by charger: default to ocpp1.6"
+            )
 
-        _LOGGER.info(f"Charger websocket path={websocket.path}")
-        cp_id = websocket.path.strip("/")
+        _LOGGER.info(f"Charger websocket path={websocket.request.path}")
+        cp_id = websocket.request.path.strip("/")
         cp_id = cp_id[cp_id.rfind("/") + 1 :]
         if self.settings.cpid not in self.charge_points:
             _LOGGER.info(f"Charger {cp_id} connected to {self.host}:{self.port}.")

custom_components/ocpp/chargepoint.py

@@ -21,7 +21,9 @@
 from homeassistant.helpers import device_registry, entity_component, entity_registry
 import homeassistant.helpers.config_validation as cv
 import voluptuous as vol
-import websockets.server
+from websockets.asyncio.server import ServerConnection
+from websockets.exceptions import WebSocketException
+from websockets.protocol import State
 
 from ocpp.charge_point import ChargePoint as cp
 from ocpp.v16 import call as callv16
@@ -471,7 +473,7 @@ async def monitor_connection(self):
         self._metrics[cstat.latency_pong.value].unit = "ms"
         connection = self._connection
         timeout_counter = 0
-        while connection.open:
+        while connection.state is State.OPEN:
             try:
                 await asyncio.sleep(self.central.websocket_ping_interval)
                 time0 = time.perf_counter()
@@ -529,7 +531,7 @@ async def run(self, tasks):
             await asyncio.gather(*self.tasks)
         except TimeoutError:
             pass
-        except websockets.exceptions.WebSocketException as websocket_exception:
+        except WebSocketException as websocket_exception:
             _LOGGER.debug(f"Connection closed to '{self.id}': {websocket_exception}")
         except Exception as other_exception:
             _LOGGER.error(
@@ -542,13 +544,13 @@ async def run(self, tasks):
     async def stop(self):
         """Close connection and cancel ongoing tasks."""
         self.status = STATE_UNAVAILABLE
-        if self._connection.open:
+        if self._connection.state is State.OPEN:
             _LOGGER.debug(f"Closing websocket to '{self.id}'")
             await self._connection.close()
         for task in self.tasks:
             task.cancel()
 
-    async def reconnect(self, connection: websockets.server.WebSocketServerProtocol):
+    async def reconnect(self, connection: ServerConnection):
         """Reconnect charge point."""
         _LOGGER.debug(f"Reconnect websocket to {self.id}")
 

custom_components/ocpp/manifest.json

@@ -14,7 +14,7 @@
 	"issue_tracker": "https://github.com/lbbrhzn/ocpp/issues",
 	"requirements": [
 		"ocpp>=1.0.0",
-		"websockets>=12.0"
+		"websockets>=13.1"
 	],
 	"version": "0.6.1"
 }

custom_components/ocpp/ocppv16.py

@@ -9,7 +9,7 @@
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 import voluptuous as vol
-import websockets.server
+from websockets.asyncio.server import ServerConnection
 
 from ocpp.routing import on
 from ocpp.v16 import call, call_result
@@ -74,7 +74,7 @@ class ChargePoint(cp):
     def __init__(
         self,
         id: str,
-        connection: websockets.server.WebSocketServerProtocol,
+        connection: ServerConnection,
         hass: HomeAssistant,
         entry: ConfigEntry,
         central: CentralSystemSettings,

custom_components/ocpp/ocppv201.py

@@ -11,7 +11,7 @@
 from homeassistant.const import UnitOfTime
 from homeassistant.core import HomeAssistant, SupportsResponse, ServiceResponse
 from homeassistant.exceptions import ServiceValidationError, HomeAssistantError
-import websockets.server
+from websockets.asyncio.server import ServerConnection
 
 from ocpp.routing import on
 from ocpp.v201 import call, call_result
@@ -85,7 +85,7 @@ class ChargePoint(cp):
     def __init__(
         self,
         id: str,
-        connection: websockets.server.WebSocketServerProtocol,
+        connection: ServerConnection,
         hass: HomeAssistant,
         entry: ConfigEntry,
         central: CentralSystemSettings,

tests/charge_point_test.py

@@ -23,7 +23,8 @@
 from pytest_homeassistant_custom_component.common import MockConfigEntry
 from typing import Any
 from collections.abc import Callable, Awaitable
-import websockets
+from websockets import connect
+from websockets.asyncio.client import ClientConnection
 
 
 async def set_switch(hass: HomeAssistant, cs: CentralSystem, key: str, on: bool):
@@ -108,12 +109,12 @@ async def run_charge_point_test(
     config_entry: MockConfigEntry,
     identity: str,
     subprotocols: list[str] | None,
-    charge_point: Callable[[websockets.WebSocketClientProtocol], ChargePoint],
+    charge_point: Callable[[ClientConnection], ChargePoint],
     parallel_tests: list[Callable[[ChargePoint], Awaitable]],
 ) -> Any:
     """Connect web socket client to the CSMS and run a number of tests in parallel."""
     completed: list[list[bool]] = [[] for _ in parallel_tests]
-    async with websockets.connect(
+    async with connect(
         f"ws://127.0.0.1:{config_entry.data[CONF_PORT]}/{identity}",
         subprotocols=[Subprotocol(s) for s in subprotocols]
         if subprotocols is not None

tests/conftest.py

@@ -35,11 +35,11 @@ def skip_notifications_fixture():
 def bypass_get_data_fixture():
     """Skip calls to get data from API."""
     future = asyncio.Future()
-    future.set_result(websockets.WebSocketServer)
+    future.set_result(websockets.asyncio.server.Server)
     with (
-        patch("websockets.server.serve", return_value=future),
-        patch("websockets.server.WebSocketServer.close"),
-        patch("websockets.server.WebSocketServer.wait_closed"),
+        patch("websockets.asyncio.server.serve", return_value=future),
+        patch("websockets.asyncio.server.Server.close"),
+        patch("websockets.asyncio.server.Server.wait_closed"),
     ):
         yield
 

tests/const.py

@@ -76,6 +76,7 @@
     CONF_PORT: 9002,
     CONF_CPID: "test_cpid_2",
     CONF_SKIP_SCHEMA_VALIDATION: True,
+    CONF_MONITORED_VARIABLES_AUTOCONFIG: False,
 }
 
 # separate entry for switch so tests can run concurrently