import asyncio import json from datetime import datetime from typing import Any import pytest import sys import types pymysql_stub = types.ModuleType("pymysql") def _fake_connect(*args, **kwargs): raise RuntimeError("pymysql not available in tests") pymysql_stub.connect = _fake_connect pymysql_stub.cursors = types.ModuleType("pymysql.cursors") pymysql_stub.cursors.DictCursor = object sys.modules.setdefault("pymysql", pymysql_stub) sys.modules.setdefault("pymysql.cursors", pymysql_stub.cursors) import pipelet_ocpp_server as ocpp_server class DummyRequest: def __init__(self, data): self._data = data self.url = "http://test.local/api" self.method = "POST" async def json(self): return self._data @pytest.fixture(autouse=True) def clear_hubject_caches(): ocpp_server.HUBJECT_PENDING_CDRS.clear() ocpp_server.HUBJECT_ACTIVE_CDRS.clear() ocpp_server.PENDING_START_METHODS.clear() yield ocpp_server.HUBJECT_PENDING_CDRS.clear() ocpp_server.HUBJECT_ACTIVE_CDRS.clear() ocpp_server.PENDING_START_METHODS.clear() @pytest.mark.asyncio async def test_hubject_remote_start_success(monkeypatch): calls: list[tuple[str, int, str]] = [] register_calls: list[tuple[str, str | None, bool]] = [] async def dummy_remote_start(station_id, connector_id, id_tag): calls.append((station_id, connector_id, id_tag)) return {"status": "Accepted"} def dummy_register(station_id, id_tag, *, allow_any_rfid=False): register_calls.append((station_id, id_tag, allow_any_rfid)) monkeypatch.setattr(ocpp_server, "remote_start_transaction", dummy_remote_start) monkeypatch.setattr( ocpp_server, "register_remote_api_authorization", dummy_register ) monkeypatch.setattr(ocpp_server, "_find_chargepoint_by_evse", lambda evse: "CP1") request = DummyRequest({"evse_id": "DE*PIP*E1", "connector_id": 2, "token": "ABCD"}) response = await ocpp_server.hubject_remote_start(request) assert response.status == 200 payload = json.loads(response.text) assert payload["status"] == "Accepted" assert payload["chargepoint_id"] == "CP1" assert payload["connector_id"] == 2 assert payload["id_tag"] == "ABCD" assert payload["evse_id"] == "DE*PIP*E1" assert calls == [("CP1", 2, "ABCD")] assert register_calls == [("CP1", "ABCD", True)] @pytest.mark.asyncio async def test_hubject_remote_start_derives_connector(monkeypatch): recorded: list[tuple[str, int, str]] = [] register_calls: list[tuple[str, str | None, bool]] = [] async def dummy_remote_start(station_id, connector_id, id_tag): recorded.append((station_id, connector_id, id_tag)) return {"status": "Accepted"} def dummy_register(station_id, id_tag, *, allow_any_rfid=False): register_calls.append((station_id, id_tag, allow_any_rfid)) monkeypatch.setattr(ocpp_server, "remote_start_transaction", dummy_remote_start) monkeypatch.setattr( ocpp_server, "register_remote_api_authorization", dummy_register ) monkeypatch.setattr(ocpp_server, "_find_chargepoint_by_evse", lambda evse: "CP2") request = DummyRequest({"EVSEId": "DE*PIP*E1#3", "token": "ZXCV"}) response = await ocpp_server.hubject_remote_start(request) assert response.status == 200 data = json.loads(response.text) assert data["connector_id"] == 3 assert recorded == [("CP2", 3, "ZXCV")] assert register_calls == [("CP2", "ZXCV", True)] @pytest.mark.asyncio async def test_hubject_remote_start_remote_identification_token(monkeypatch): recorded: list[tuple[str, int, str]] = [] register_calls: list[tuple[str, str | None, bool]] = [] async def dummy_remote_start(station_id, connector_id, id_tag): recorded.append((station_id, connector_id, id_tag)) return {"status": "Accepted"} def dummy_register(station_id, id_tag, *, allow_any_rfid=False): register_calls.append((station_id, id_tag, allow_any_rfid)) monkeypatch.setattr(ocpp_server, "remote_start_transaction", dummy_remote_start) monkeypatch.setattr( ocpp_server, "register_remote_api_authorization", dummy_register ) monkeypatch.setattr(ocpp_server, "_find_chargepoint_by_evse", lambda evse: "CP3") evco_token = "DE*EMP*123456" request = DummyRequest( { "evse_id": "DE*PIP*E9#4", "connector_id": 4, "token": evco_token, "raw_request": { "Identification": {"RemoteIdentification": {"EvcoID": evco_token}} }, } ) response = await ocpp_server.hubject_remote_start(request) assert response.status == 200 payload = json.loads(response.text) assert payload["id_tag"] == evco_token assert payload["connector_id"] == 4 assert recorded == [("CP3", 4, evco_token)] assert register_calls == [("CP3", evco_token, True)] @pytest.mark.asyncio async def test_hubject_remote_start_records_start_method(monkeypatch): async def dummy_remote_start(station_id, connector_id, id_tag): return {"status": "Accepted"} def dummy_register(station_id, id_tag, *, allow_any_rfid=False): return None monkeypatch.setattr(ocpp_server, "remote_start_transaction", dummy_remote_start) monkeypatch.setattr( ocpp_server, "register_remote_api_authorization", dummy_register ) monkeypatch.setattr(ocpp_server, "_find_chargepoint_by_evse", lambda evse: "CP4") request = DummyRequest({"evse_id": "DE*PIP*E4#1", "token": "TOKEN"}) response = await ocpp_server.hubject_remote_start(request) assert response.status == 200 pending_method = ocpp_server._pop_pending_start_method("CP4", 1) assert pending_method == "remote" @pytest.mark.asyncio async def test_hubject_remote_stop_with_transaction_id(monkeypatch): recorded: list[tuple[str, int]] = [] async def dummy_remote_stop(station_id, transaction_id): recorded.append((station_id, transaction_id)) return {"status": "Accepted"} monkeypatch.setattr(ocpp_server, "remote_stop_transaction", dummy_remote_stop) monkeypatch.setattr(ocpp_server, "_find_chargepoint_by_evse", lambda evse: "CP1") request = DummyRequest( { "evse_id": "DE*PIP*E1#2", "transaction_id": 321, "session_id": "session-stop", } ) response = await ocpp_server.hubject_remote_stop(request) assert response.status == 200 payload = json.loads(response.text) assert payload["transaction_id"] == 321 assert payload["chargepoint_id"] == "CP1" assert recorded == [("CP1", 321)] @pytest.mark.asyncio async def test_hubject_remote_stop_infers_transaction(monkeypatch): recorded: list[tuple[str, int]] = [] async def dummy_remote_stop(station_id, transaction_id): recorded.append((station_id, transaction_id)) return {"status": "Accepted"} monkeypatch.setattr(ocpp_server, "remote_stop_transaction", dummy_remote_stop) monkeypatch.setattr(ocpp_server, "_find_chargepoint_by_evse", lambda evse: "CP2") monkeypatch.setattr( ocpp_server, "current_transactions", {987: {"chargepoint_id": "CP2", "connector_id_int": 3, "connector_id": "3"}}, ) request = DummyRequest({"evse_id": "DE*PIP*E2#3"}) response = await ocpp_server.hubject_remote_stop(request) assert response.status == 200 payload = json.loads(response.text) assert payload["transaction_id"] == 987 assert payload["chargepoint_id"] == "CP2" assert payload["connector_id"] == 3 assert recorded == [("CP2", 987)] @pytest.mark.asyncio async def test_hubject_remote_stop_infers_transaction_by_session(monkeypatch): recorded: list[tuple[str, int]] = [] async def dummy_remote_stop(station_id, transaction_id): recorded.append((station_id, transaction_id)) return {"status": "Accepted"} monkeypatch.setattr(ocpp_server, "remote_stop_transaction", dummy_remote_stop) monkeypatch.setattr(ocpp_server, "_find_chargepoint_by_evse", lambda evse: "CP3") monkeypatch.setattr( ocpp_server, "current_transactions", { 654: { "chargepoint_id": "CP3", "connector_id_int": 1, "connector_id": "1", "hubject": { "metadata": { "session_id": "session-stop", "cpo_partner_session_id": "session-stop", } }, } }, ) request = DummyRequest( { "evse_id": "DE*PIP*E3#1", "session_id": "session-stop", } ) response = await ocpp_server.hubject_remote_stop(request) assert response.status == 200 payload = json.loads(response.text) assert payload["transaction_id"] == 654 assert payload["chargepoint_id"] == "CP3" assert recorded == [("CP3", 654)] @pytest.mark.asyncio async def test_hubject_cdr_flow(monkeypatch): cdr_calls: list[dict[str, Any]] = [] async def dummy_remote_start(station_id, connector_id, id_tag): return {"status": "Accepted"} def dummy_register(station_id, id_tag, *, allow_any_rfid=False): return None monkeypatch.setattr(ocpp_server, "remote_start_transaction", dummy_remote_start) monkeypatch.setattr( ocpp_server, "register_remote_api_authorization", dummy_register ) monkeypatch.setattr(ocpp_server, "_find_chargepoint_by_evse", lambda evse: "CP9") monkeypatch.setattr( ocpp_server, "_resolve_evse_identifier", lambda chargepoint_id, connector_id: f"EVSE-{chargepoint_id}-{connector_id}", ) request_payload = { "evse_id": "DE*PIP*E9#1", "token": "REMOTE123", "session_id": "SESSION-1", "Identification": {"RemoteIdentification": {"EvcoID": "DE*EMP*ABC"}}, } response = await ocpp_server.hubject_remote_start(DummyRequest(request_payload)) assert response.status == 200 pending_key = ("CP9", 1) assert pending_key in ocpp_server.HUBJECT_PENDING_CDRS start_payload = { "connectorId": 1, "meterStart": 12345, "timestamp": "2023-01-01T00:00:00Z", "idTag": "REMOTE123", } tx_id = 42 start_entry = { "chargepoint_id": "CP9", "start_time": datetime.now(), "connector_id": start_payload["connectorId"], "connector_id_int": 1, "oicp_enabled": False, "start_info": dict(start_payload), } start_entry["start_info"]["chargepoint_id"] = "CP9" start_entry["start_info"]["evseId"] = "EVSE-CP9-1" ocpp_server.current_transactions[tx_id] = start_entry ocpp_server._attach_hubject_metadata_to_transaction("CP9", 1, tx_id, start_entry) assert pending_key not in ocpp_server.HUBJECT_PENDING_CDRS assert tx_id in ocpp_server.HUBJECT_ACTIVE_CDRS stop_payload = { "transactionId": tx_id, "meterStop": 12500, "timestamp": "2023-01-01T01:00:00Z", "idTag": "REMOTE123", } info = ocpp_server.current_transactions.pop(tx_id) async def capture_cdr(payload): cdr_calls.append(payload) monkeypatch.setattr(ocpp_server, "_send_hubject_cdr_via_api", capture_cdr) tasks: list[asyncio.Task] = [] real_create_task = asyncio.create_task def capture_task(coro): task = real_create_task(coro) tasks.append(task) return task monkeypatch.setattr(asyncio, "create_task", capture_task) stop_time = datetime.fromisoformat("2023-01-01T01:00:00+00:00") ocpp_server._schedule_hubject_cdr_submission( "CP9", 1, tx_id, info, stop_payload, stop_time, stop_payload["meterStop"], ) assert pending_key not in ocpp_server.HUBJECT_PENDING_CDRS assert tx_id not in ocpp_server.HUBJECT_ACTIVE_CDRS if tasks: await asyncio.gather(*tasks) assert cdr_calls payload = cdr_calls[0]["Cdr"] assert payload["SessionID"] == "SESSION-1" assert payload["EvseID"] == "EVSE-CP9-1" assert payload["MeterStart"] == 12345 assert payload["MeterStop"] == 12500 assert payload["Identification"] == request_payload["Identification"]