Fix intermittent reconnect failures and missing charge point ID extraction

examples/v16/central_system.py

@@ -13,6 +13,7 @@
 
     sys.exit(1)
 
+from ocpp.charge_point import create_and_start_charge_point
 from ocpp.routing import on
 from ocpp.v16 import ChargePoint as cp
 from ocpp.v16 import call_result
@@ -56,10 +57,7 @@ async def on_connect(websocket):
         )
         return await websocket.close()
 
-    charge_point_id = websocket.request.path.strip("/")
-    cp = ChargePoint(charge_point_id, websocket)
-
-    await cp.start()
+    await create_and_start_charge_point(websocket, ChargePoint)
 
 
 async def main():

examples/v201/central_system.py

@@ -13,6 +13,7 @@
 
     sys.exit(1)
 
+from ocpp.charge_point import create_and_start_charge_point
 from ocpp.routing import on
 from ocpp.v201 import ChargePoint as cp
 from ocpp.v201 import call_result
@@ -61,10 +62,7 @@ async def on_connect(websocket):
         )
         return await websocket.close()
 
-    charge_point_id = websocket.request.path.strip("/")
-    charge_point = ChargePoint(charge_point_id, websocket)
-
-    await charge_point.start()
+    await create_and_start_charge_point(websocket, ChargePoint)
 
 
 async def main():

examples/v21/central_system.py

@@ -13,6 +13,7 @@
 
     sys.exit(1)
 
+from ocpp.charge_point import create_and_start_charge_point
 from ocpp.routing import on
 from ocpp.v21 import ChargePoint as cp
 from ocpp.v21 import call_result
@@ -61,10 +62,7 @@ async def on_connect(websocket):
         )
         return await websocket.close()
 
-    charge_point_id = websocket.request.path.strip("/")
-    charge_point = ChargePoint(charge_point_id, websocket)
-
-    await charge_point.start()
+    await create_and_start_charge_point(websocket, ChargePoint)
 
 
 async def main():

ocpp/charge_point.py

@@ -5,7 +5,8 @@
 import time
 import uuid
 from dataclasses import Field, asdict, is_dataclass
-from typing import Any, Dict, List, Union, get_args, get_origin
+from typing import Any, Dict, List, Optional, Union, get_args, get_origin
+from urllib.parse import urlparse
 
 from ocpp.exceptions import NotImplementedError, NotSupportedError, OCPPError
 from ocpp.messages import Call, MessageType, unpack, validate_payload
@@ -14,6 +15,84 @@
 LOGGER = logging.getLogger("ocpp")
 
 
+def extract_charge_point_id(path: Optional[str]) -> Optional[str]:
+    """Extract the charge point ID from a WebSocket URL path.
+
+    In OCPP, chargers connect to a WebSocket endpoint and include their
+    identity as the last segment of the URL path. For example, a charger
+    with ID "CP001" would connect to ``ws://central-system:9000/CP001``
+    or ``ws://central-system:9000/ocpp/CP001``.
+
+    This function handles various path formats robustly:
+
+    - ``/CP001`` → ``CP001``
+    - ``/ocpp/CP001`` → ``CP001``
+    - ``/`` → ``None``
+    - ``""`` → ``None``
+
+    Args:
+        path: The URL path from the WebSocket request
+            (e.g., ``websocket.request.path``).
+
+    Returns:
+        The charge point ID string, or ``None`` if the path does not
+        contain a valid identifier.
+
+    See Also:
+        :func:`create_and_start_charge_point` for a higher-level helper
+        that extracts the ID, validates it, and starts a ``ChargePoint``.
+    """
+    if not path:
+        return None
+
+    # Strip query strings and fragments, then get the path
+    parsed = urlparse(path)
+    clean_path = parsed.path
+
+    # Take the last non-empty segment as the charge point ID
+    segments = [s for s in clean_path.split("/") if s]
+    if not segments:
+        return None
+
+    charge_point_id = segments[-1]
+
+    # Validate: ID should not be empty after stripping whitespace
+    charge_point_id = charge_point_id.strip()
+    if not charge_point_id:
+        return None
+
+    return charge_point_id
+
+
+async def create_and_start_charge_point(websocket, charge_point_cls):
+    """Extract the charge point ID from a WebSocket and start a ChargePoint.
+
+    This is a convenience coroutine for central system implementations.
+    It extracts the charge point ID from the WebSocket request path,
+    creates an instance of ``charge_point_cls``, and starts it.
+
+    If the path does not contain a valid charge point ID, the WebSocket
+    connection is closed and ``None`` is returned.
+
+    Args:
+        websocket: The WebSocket connection from the ``websockets`` library.
+        charge_point_cls: A ``ChargePoint`` subclass to instantiate.
+
+    Returns:
+        The ``ChargePoint`` instance, or ``None`` if the path was invalid.
+    """
+    charge_point_id = extract_charge_point_id(websocket.request.path)
+    if not charge_point_id:
+        LOGGER.error("No charge point ID in path: %s", websocket.request.path)
+        await websocket.close()
+        return None
+
+    LOGGER.info("Charge point %s connected", charge_point_id)
+    cp = charge_point_cls(charge_point_id, websocket)
+    await cp.start()
+    return cp
+
+
 def camel_to_snake_case(data):
     """
     Convert all keys of all dictionaries inside the given argument from

tests/test_charge_point_connection.py

@@ -0,0 +1,197 @@
+"""Tests for charge point ID extraction and connection handling.
+
+These tests address the issues described in GitHub issue #751:
+- Robust extraction of charge point IDs from WebSocket paths
+- Connection exceptions propagate to consumers for proper handling
+"""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from ocpp.charge_point import create_and_start_charge_point, extract_charge_point_id
+from ocpp.routing import on
+from ocpp.v16 import ChargePoint as cp_16
+from ocpp.v16 import call_result
+from ocpp.v16.enums import Action, RegistrationStatus
+
+# ---------------------------------------------------------------------------
+# Tests for extract_charge_point_id
+# ---------------------------------------------------------------------------
+
+
+class TestExtractChargePointId:
+    """Test the extract_charge_point_id utility function."""
+
+    def test_simple_path(self):
+        """Standard OCPP path with just the charge point ID."""
+        assert extract_charge_point_id("/CP001") == "CP001"
+
+    def test_nested_path(self):
+        """Path with prefix segments (e.g., /ocpp/CP001)."""
+        assert extract_charge_point_id("/ocpp/CP001") == "CP001"
+
+    def test_deeply_nested_path(self):
+        """Deeply nested path still returns the last segment."""
+        assert extract_charge_point_id("/api/v1/ocpp/CP001") == "CP001"
+
+    def test_trailing_slash(self):
+        """Trailing slash should not affect extraction."""
+        assert extract_charge_point_id("/CP001/") == "CP001"
+
+    def test_root_path_returns_none(self):
+        """Root path '/' has no charge point ID."""
+        assert extract_charge_point_id("/") is None
+
+    def test_empty_string_returns_none(self):
+        """Empty string has no charge point ID."""
+        assert extract_charge_point_id("") is None
+
+    def test_none_returns_none(self):
+        """None input returns None."""
+        assert extract_charge_point_id(None) is None
+
+    def test_path_without_leading_slash(self):
+        """Path without leading slash should still work."""
+        assert extract_charge_point_id("CP001") == "CP001"
+
+    def test_path_with_query_string(self):
+        """Query strings should be stripped before extracting ID."""
+        assert extract_charge_point_id("/CP001?token=abc123") == "CP001"
+
+    def test_path_with_fragment(self):
+        """Fragments should be stripped before extracting ID."""
+        assert extract_charge_point_id("/CP001#section") == "CP001"
+
+    def test_whitespace_only_segment(self):
+        """Path with only whitespace segment returns None."""
+        assert extract_charge_point_id("/   ") is None
+
+    def test_charge_point_id_with_special_chars(self):
+        """Charge point IDs can contain hyphens, underscores, etc."""
+        assert extract_charge_point_id("/CP-001_v2") == "CP-001_v2"
+
+    def test_charge_point_id_with_dots(self):
+        """Charge point IDs can contain dots (serial numbers)."""
+        assert extract_charge_point_id("/EVB-P12354.00.01") == "EVB-P12354.00.01"
+
+    def test_multiple_slashes(self):
+        """Multiple consecutive slashes should be handled gracefully."""
+        assert extract_charge_point_id("///CP001") == "CP001"
+
+
+# ---------------------------------------------------------------------------
+# Tests for ChargePoint.start() connection handling
+# ---------------------------------------------------------------------------
+
+
+class TestChargePointStart:
+    """Test that ChargePoint.start() propagates connection exceptions."""
+
+    @pytest.mark.asyncio
+    async def test_start_propagates_exception_on_connection_closed(self, connection):
+        """start() should let connection exceptions propagate to the caller."""
+
+        class MyCP(cp_16):
+            @on(Action.boot_notification)
+            def on_boot_notification(self, **kwargs):
+                return call_result.BootNotification(
+                    current_time="2025-01-01T00:00:00Z",
+                    interval=10,
+                    status=RegistrationStatus.accepted,
+                )
+
+        connection.recv = AsyncMock(side_effect=Exception("Connection closed"))
+
+        cp = MyCP("CP001", connection)
+
+        with pytest.raises(Exception):
+            await cp.start()
+
+    @pytest.mark.asyncio
+    async def test_start_processes_messages_before_exception(self, connection):
+        """start() should process messages until the connection raises."""
+        messages_received = []
+
+        class MyCP(cp_16):
+            @on(Action.boot_notification)
+            def on_boot_notification(self, **kwargs):
+                messages_received.append("boot")
+                return call_result.BootNotification(
+                    current_time="2025-01-01T00:00:00Z",
+                    interval=10,
+                    status=RegistrationStatus.accepted,
+                )
+
+        boot_msg = (
+            '[2,"123","BootNotification",'
+            '{"chargePointVendor":"vendor","chargePointModel":"model"}]'
+        )
+        connection.recv = AsyncMock(
+            side_effect=[boot_msg, Exception("Connection closed")]
+        )
+
+        cp = MyCP("CP001", connection)
+
+        with pytest.raises(Exception):
+            await cp.start()
+
+        assert len(messages_received) == 1
+
+    @pytest.mark.asyncio
+    async def test_reconnection_with_new_instance(self, connection):
+        """Simulates a charger reconnecting by creating a new ChargePoint."""
+        connection.recv = AsyncMock(side_effect=Exception("Connection closed"))
+
+        # First connection
+        cp1 = cp_16("CP001", connection)
+        with pytest.raises(Exception):
+            await cp1.start()
+
+        # Second connection (simulating reconnect with new websocket)
+        connection2 = MagicMock()
+        connection2.send = AsyncMock()
+        connection2.recv = AsyncMock(side_effect=Exception("Connection closed"))
+
+        cp2 = cp_16("CP001", connection2)
+        with pytest.raises(Exception):
+            await cp2.start()
+
+
+# ---------------------------------------------------------------------------
+# Tests for create_and_start_charge_point
+# ---------------------------------------------------------------------------
+
+
+class TestCreateAndStartChargePoint:
+    """Test the create_and_start_charge_point helper."""
+
+    @pytest.mark.asyncio
+    async def test_valid_path_creates_and_starts(self):
+        """Should create a ChargePoint and call start() for a valid path."""
+        ws = MagicMock()
+        ws.request.path = "/CP001"
+        ws.close = AsyncMock()
+
+        started = []
+
+        class MyCP(cp_16):
+            async def start(self):
+                started.append(self.id)
+
+        result = await create_and_start_charge_point(ws, MyCP)
+        assert result is not None
+        assert result.id == "CP001"
+        assert started == ["CP001"]
+        ws.close.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_invalid_path_closes_connection(self):
+        """Should close the websocket and return None for an invalid path."""
+        ws = MagicMock()
+        ws.request.path = "/"
+        ws.close = AsyncMock()
+
+        result = await create_and_start_charge_point(ws, cp_16)
+        assert result is None
+        ws.close.assert_awaited_once()