import json import threading from contextlib import contextmanager from datetime import datetime, timezone from queue import Queue from typing import Any, Iterator import pytest from flask import Flask, jsonify, request from werkzeug.serving import make_server import services.hubject_api as hubject_api import pipelet_ocpp_server as ocpp_server class MockOCPPServer: def __init__(self) -> None: self.app = Flask("mock_ocpp") self.queue: Queue[tuple[str, dict, str]] = Queue() @self.app.post("/hubject/authorize-start") @self.app.post("/api/hubject/authorize-start") def authorize_start(): data = request.get_json() or {} self.queue.put(("start", data, request.path)) return jsonify({"status": "Accepted", "SessionID": data.get("session_id")}) @self.app.post("/hubject/authorize-stop") @self.app.post("/api/hubject/authorize-stop") def authorize_stop(): data = request.get_json() or {} self.queue.put(("stop", data, request.path)) return jsonify({"status": "Accepted", "SessionID": data.get("session_id")}) @self.app.post("/hubject/cdr") @self.app.post("/api/hubject/cdr") def cdr(): data = request.get_json() or {} self.queue.put(("cdr", data, request.path)) return jsonify({"stored": True, "meter_end": data.get("MeterValueEnd")}) @self.app.post("/hubject/evse-data") @self.app.post("/api/hubject/evse-data") def evse_data(): data = request.get_json() or {} self.queue.put(("evse_data", data, request.path)) return jsonify( { "OperatorID": data.get("OperatorID"), "ActionType": data.get("ActionType"), "hubject_response": {"StatusCode": {"Code": "000"}}, } ) @self.app.post("/hubject/evse-status") @self.app.post("/api/hubject/evse-status") def evse_status(): data = request.get_json() or {} self.queue.put(("evse_status", data, request.path)) return jsonify( { "OperatorID": data.get("OperatorID"), "ActionType": data.get("ActionType"), "status": data.get("OperatorEvseStatus"), } ) @self.app.post("/hubject/remote-start") @self.app.post("/api/hubject/remote-start") def remote_start(): data = request.get_json() or {} self.queue.put(("remote-start", data, request.path)) return jsonify({"status": "Accepted", "session_id": data.get("session_id")}) @self.app.post("/hubject/remote-stop") @self.app.post("/api/hubject/remote-stop") def remote_stop(): data = request.get_json() or {} self.queue.put(("remote-stop", data, request.path)) return jsonify({"status": "Accepted", "transaction_id": data.get("transaction_id")}) @contextmanager def run(self) -> Iterator[str]: server = make_server("127.0.0.1", 0, self.app) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: port = server.server_port base_url = f"http://127.0.0.1:{port}/" self.base_url = base_url yield base_url finally: server.shutdown() thread.join() @pytest.fixture(autouse=True) def _reset_client(): original_client = hubject_api.OCPP_CLIENT original_hubject_client = getattr(hubject_api, "HUBJECT_API_CLIENT", None) yield hubject_api.OCPP_CLIENT = original_client hubject_api.HUBJECT_API_CLIENT = original_hubject_client @pytest.fixture def client(monkeypatch): mock_server = MockOCPPServer() with mock_server.run() as base_url: monkeypatch.setattr( hubject_api, "OCPP_CLIENT", hubject_api.OCPPApiClient(base_url, timeout=1, verify=False), ) class DummyHubjectApiClient: def __init__(self) -> None: self.calls: list[tuple[str, dict]] = [] self.status_calls: list[tuple[str, dict]] = [] self.cdr_calls: list[dict] = [] self.authorize_start_calls: list[dict] = [] self.authorize_stop_calls: list[dict] = [] def push_evse_data(self, operator_id: str, payload: dict) -> dict: call_payload = dict(payload) self.calls.append((operator_id, call_payload)) return { "StatusCode": {"Code": "000", "Description": "Accepted"}, "OperatorID": operator_id, } def push_evse_status(self, operator_id: str, payload: dict) -> dict: call_payload = dict(payload) self.status_calls.append((operator_id, call_payload)) return { "StatusCode": {"Code": "000", "Description": "Accepted"}, "OperatorID": operator_id, } def send_cdr(self, payload: dict) -> dict: call_payload = dict(payload) self.cdr_calls.append(call_payload) return { "StatusCode": {"Code": "000", "Description": "Accepted"}, "Result": True, } def authorize_start_evse(self, payload: dict) -> dict: call_payload = dict(payload) self.authorize_start_calls.append(call_payload) session_id = ( call_payload.get("CPOPartnerSessionID") or call_payload.get("SessionID") or "session" ) response: dict[str, Any] = { "AuthorizationStatus": "Authorized", "StatusCode": {"Code": "000", "Description": "Success"}, "CPOPartnerSessionID": session_id, "AuthorizationStart": { "SessionID": session_id, "Timestamp": hubject_api._now(), }, "Identification": call_payload.get("Identification"), } operator_id = call_payload.get("OperatorID") if operator_id: response["OperatorID"] = operator_id provider_id = call_payload.get("ProviderID") if provider_id: response["ProviderID"] = provider_id return response def authorize_stop_evse(self, payload: dict) -> dict: call_payload = dict(payload) self.authorize_stop_calls.append(call_payload) session_id = ( call_payload.get("CPOPartnerSessionID") or call_payload.get("SessionID") or "session" ) response: dict[str, Any] = { "AuthorizationStatus": "Authorized", "StatusCode": {"Code": "000", "Description": "Success"}, "CPOPartnerSessionID": session_id, "AuthorizationStop": { "SessionID": session_id, "Timestamp": hubject_api._now(), }, "Identification": call_payload.get("Identification"), } operator_id = call_payload.get("OperatorID") if operator_id: response["OperatorID"] = operator_id provider_id = call_payload.get("ProviderID") if provider_id: response["ProviderID"] = provider_id return response dummy_hubject_client = DummyHubjectApiClient() monkeypatch.setattr(hubject_api, "HUBJECT_API_CLIENT", dummy_hubject_client) with hubject_api.app.test_client() as flask_client: yield flask_client, mock_server, dummy_hubject_client def _verified_environ(): return {"SSL_CLIENT_VERIFY": "SUCCESS"} def test_authorize_start_success(client): flask_client, mock_server, dummy_client = client response = flask_client.post( "/Authorization/AuthorizeStart", data=json.dumps( { "EVSEId": "DE*PIP*E123", "CPOPartnerSessionID": "session-1", "Identification": { "RFIDMifareFamilyIdentification": {"UID": "ABCD1234"} }, } ), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 data = response.get_json() assert data["AuthorizationStatus"] == "Authorized" assert data["StatusCode"]["Code"] == "000" assert "HubjectResponse" in data assert dummy_client.authorize_start_calls call_payload = dummy_client.authorize_start_calls[-1] assert call_payload["EVSEId"] == "DE*PIP*E123" assert call_payload["CPOPartnerSessionID"] == "session-1" identification = call_payload["Identification"] assert identification["RFIDMifareFamilyIdentification"]["UID"] == "ABCD1234" assert mock_server.queue.empty() def test_authorize_start_without_operator_id(client, monkeypatch): flask_client, mock_server, dummy_client = client monkeypatch.setattr(hubject_api, "HUBJECT_OPERATOR_ID", "DE*FALLBACK") response = flask_client.post( "/Authorization/AuthorizeStart", data=json.dumps( { "EVSEId": "DE*PIP*E999", "Identification": { "RFIDMifareFamilyIdentification": {"UID": "NOPERATOR"} }, } ), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 data = response.get_json() assert data["AuthorizationStatus"] == "Authorized" assert mock_server.queue.empty() assert dummy_client.authorize_start_calls call_payload = dummy_client.authorize_start_calls[-1] assert call_payload["OperatorID"] == "DE*FALLBACK" def test_authorize_stop_requires_certificate(client): flask_client, _, __ = client response = flask_client.post( "/Authorization/AuthorizeStop", data=json.dumps( { "EVSEId": "DE*PIP*E999", "Identification": { "RFIDMifareFamilyIdentification": {"UID": "999"} }, } ), headers={"Content-Type": "application/json"}, ) assert response.status_code == 403 def test_authorize_stop_success(client): flask_client, mock_server, dummy_client = client response = flask_client.post( "/Authorization/AuthorizeStop", data=json.dumps( { "EVSEId": "DE*PIP*E123", "Identification": { "RFIDMifareFamilyIdentification": {"UID": "ABCD1234"} }, } ), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 data = response.get_json() assert data["AuthorizationStatus"] == "Authorized" assert "HubjectResponse" in data assert dummy_client.authorize_stop_calls stop_call = dummy_client.authorize_stop_calls[-1] assert stop_call["EVSEId"] == "DE*PIP*E123" identification = stop_call["Identification"] assert identification["RFIDMifareFamilyIdentification"]["UID"] == "ABCD1234" assert mock_server.queue.empty() def test_send_cdr_missing_client_returns_503(client): flask_client, _, dummy_client = client hubject_api.HUBJECT_API_CLIENT = None response = flask_client.post( "/Cdrmgmt/Send", data=json.dumps({"Cdr": {"SessionID": "session-xyz"}}), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 503 data = response.get_json() assert data["Result"] is False assert data["StatusCode"]["Code"] == "301" hubject_api.HUBJECT_API_CLIENT = dummy_client def test_send_cdr_hubject_failure_returns_502(client, monkeypatch): flask_client, _, dummy_client = client def _raise_error(payload): raise hubject_api.HubjectApiError("boom") monkeypatch.setattr(dummy_client, "send_cdr", _raise_error) response = flask_client.post( "/Cdrmgmt/Send", data=json.dumps({"Cdr": {"SessionID": "session-error"}}), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 502 data = response.get_json() assert data["Result"] is False assert data["StatusCode"]["Code"] == "301" assert data["CdrResponse"]["error"] == "boom" def test_authorize_remote_start_success(client): flask_client, mock_server, _ = client payload = { "EvseID": "DE*PIP*E123", "CPOPartnerSessionID": "session-remote", "EMPPartnerSessionID": "emp-remote", "ProviderID": "DE*EMP", "Identification": { "RFIDMifareFamilyIdentification": {"UID": "ABCD1234"} }, "ConnectorID": 2, } response = flask_client.post( "/api/oicp/charging/v21/providers/DE*PIP/authorize-remote/start", data=json.dumps(payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 data = response.get_json() assert data["AuthorizationStatus"] == "Authorized" assert data["ProviderID"] == "DE*EMP" assert data["OperatorID"] == "DE*PIP" assert data["ConnectorID"] == 2 event_type, ocpp_payload, _ = mock_server.queue.get(timeout=1) assert event_type == "remote-start" assert ocpp_payload["evse_id"] == "DE*PIP*E123" assert ocpp_payload["session_id"] == "session-remote" assert ocpp_payload["operator_id"] == "DE*PIP" assert ocpp_payload["token"] == "ABCD1234" assert ocpp_payload["connector_id"] == 2 assert ocpp_payload["raw_request"] == payload def test_authorize_remote_start_uppercase_path(client): flask_client, mock_server, _ = client payload = { "EvseID": "DE*PIP*E123", "CPOPartnerSessionID": "session-remote", } response = flask_client.post( "/api/oicp/charging/v21/providers/DE*PIP/authorize-remote/Start", data=json.dumps(payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 data = response.get_json() assert data["AuthorizationStatus"] == "Authorized" event_type, ocpp_payload, _ = mock_server.queue.get(timeout=1) assert event_type == "remote-start" assert ocpp_payload["session_id"] == "session-remote" assert ocpp_payload["raw_request"] == payload def test_authorize_remote_start_remote_identification(client): flask_client, mock_server, _ = client evco_token = "DE*EMP*REMOTE" payload = { "EVSEId": "DE*PIP*E555", "Identification": {"RemoteIdentification": {"EvcoID": evco_token}}, } response = flask_client.post( "/api/oicp/charging/v21/providers/DE*PIP/authorize-remote/start", data=json.dumps(payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 data = response.get_json() assert data["AuthorizationStatus"] == "Authorized" event_type, ocpp_payload, _ = mock_server.queue.get(timeout=1) assert event_type == "remote-start" assert ocpp_payload["token"] == evco_token assert ocpp_payload["raw_request"]["Identification"]["RemoteIdentification"][ "EvcoID" ] == evco_token def test_authorize_remote_start_oicp_v23(client): flask_client, mock_server, _ = client payload = { "OperatorID": "DE*PIP", "EvseID": "DE*PIP*E999", "CPOPartnerSessionID": "session-v23", "ProviderID": "DE*EMP", } response = flask_client.post( "/api/oicp/2.3/authorize-remote-start-evse", data=json.dumps(payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 data = response.get_json() assert data["AuthorizationStatus"] == "Authorized" assert data["OperatorID"] == "DE*PIP" assert data["ProviderID"] == "DE*EMP" event_type, ocpp_payload, _ = mock_server.queue.get(timeout=1) assert event_type == "remote-start" assert ocpp_payload["evse_id"] == "DE*PIP*E999" assert ocpp_payload["session_id"] == "session-v23" assert ocpp_payload["operator_id"] == "DE*PIP" assert ocpp_payload["provider_id"] == "DE*EMP" def test_authorize_remote_stop_success(client): flask_client, mock_server, _ = client payload = { "EvseID": "DE*PIP*E123#1", "CPOPartnerSessionID": "session-stop", "TransactionID": 42, "ProviderID": "DE*EMP", } response = flask_client.post( "/api/oicp/charging/v21/providers/DE*PIP/authorize-remote/stop", data=json.dumps(payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 data = response.get_json() assert data["AuthorizationStatus"] == "Authorized" assert data["ProviderID"] == "DE*EMP" assert data["OperatorID"] == "DE*PIP" assert data["TransactionID"] == 42 event_type, ocpp_payload, _ = mock_server.queue.get(timeout=1) assert event_type == "remote-stop" assert ocpp_payload["evse_id"] == "DE*PIP*E123#1" assert ocpp_payload["session_id"] == "session-stop" assert ocpp_payload["operator_id"] == "DE*PIP" assert ocpp_payload["transaction_id"] == 42 assert ocpp_payload["raw_request"] == payload def test_authorize_remote_stop_uppercase_path(client): flask_client, mock_server, _ = client payload = { "EvseID": "DE*PIP*E123#1", "CPOPartnerSessionID": "session-stop", } response = flask_client.post( "/api/oicp/charging/v21/providers/DE*PIP/authorize-remote/Stop", data=json.dumps(payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 data = response.get_json() assert data["AuthorizationStatus"] == "Authorized" event_type, ocpp_payload, _ = mock_server.queue.get(timeout=1) assert event_type == "remote-stop" assert ocpp_payload["session_id"] == "session-stop" assert ocpp_payload["raw_request"] == payload def test_authorize_remote_stop_oicp_v23(client): flask_client, mock_server, _ = client payload = { "OperatorID": "DE*PIP", "EvseID": "DE*PIP*E123#1", "CPOPartnerSessionID": "session-stop-v23", "TransactionID": 99, "ProviderID": "DE*EMP", } response = flask_client.post( "/api/oicp/2.3/authorize-remote-stop-evse", data=json.dumps(payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 data = response.get_json() assert data["AuthorizationStatus"] == "Authorized" assert data["OperatorID"] == "DE*PIP" assert data["TransactionID"] == 99 event_type, ocpp_payload, _ = mock_server.queue.get(timeout=1) assert event_type == "remote-stop" assert ocpp_payload["transaction_id"] == 99 assert ocpp_payload["operator_id"] == "DE*PIP" assert ocpp_payload["session_id"] == "session-stop-v23" def test_authorize_remote_stop_invalid_transaction(client): flask_client, _, __ = client payload = { "EvseID": "DE*PIP*E123#1", "CPOPartnerSessionID": "session-stop", "TransactionID": "not-a-number", } response = flask_client.post( "/api/oicp/charging/v21/providers/DE*PIP/authorize-remote/stop", data=json.dumps(payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 400 data = response.get_json() assert data["StatusCode"]["Code"] == "202" assert "Invalid TransactionID" in data["StatusCode"]["Description"] assert data["StatusCode"]["AdditionalInfo"] == "TransactionID must be an integer" def test_remote_commands_respect_ocpp_base_url_prefix(client, monkeypatch): flask_client, mock_server, _ = client prefixed_ocpp_client = hubject_api.OCPPApiClient( f"{mock_server.base_url}api", timeout=1, verify=False, ) monkeypatch.setattr(hubject_api, "OCPP_CLIENT", prefixed_ocpp_client) start_payload = { "EvseID": "DE*PIP*E123", "CPOPartnerSessionID": "session-prefixed", "ProviderID": "DE*EMP", } start_response = flask_client.post( "/api/oicp/charging/v21/providers/DE*PIP/authorize-remote/start", data=json.dumps(start_payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert start_response.status_code == 200 event_type, ocpp_payload, request_path = mock_server.queue.get(timeout=1) assert event_type == "remote-start" assert request_path == "/api/hubject/remote-start" assert ocpp_payload["session_id"] == "session-prefixed" stop_payload = { "EvseID": "DE*PIP*E123", "CPOPartnerSessionID": "session-prefixed", "TransactionID": 7, } stop_response = flask_client.post( "/api/oicp/charging/v21/providers/DE*PIP/authorize-remote/stop", data=json.dumps(stop_payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert stop_response.status_code == 200 event_type, ocpp_payload, request_path = mock_server.queue.get(timeout=1) assert event_type == "remote-stop" assert request_path == "/api/hubject/remote-stop" assert ocpp_payload["transaction_id"] == 7 def test_cdr_send_success(client): flask_client, mock_server, dummy_client = client response = flask_client.post( "/Cdrmgmt/Send", data=json.dumps( { "Cdr": { "SessionID": "sess-77", "EVSEID": "DE*PIP*E123", "MeterValueEnd": 12.5, } } ), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 body = response.get_json() assert body["Result"] is True assert body["CdrResponse"]["StatusCode"]["Code"] == "000" assert dummy_client.cdr_calls[-1]["SessionID"] == "sess-77" assert mock_server.queue.empty() def test_hubject_api_client_send_cdr_normalizes_payload(monkeypatch): client = hubject_api.HubjectApiClient( "https://example.com/api/oicp/", operator_id="DE*WLE", partner_id="DE*EMP", ) captured: dict[str, object] = {} def fake_post(self, path, payload): # type: ignore[override] captured["path"] = path captured["payload"] = payload return {"StatusCode": {"Code": "000"}} monkeypatch.setattr( client, "_post", fake_post.__get__(client, hubject_api.HubjectApiClient), ) response = client.send_cdr( { "SessionID": "sess-1", "OperatorID": "DE*ALT", "EvseID": "DE*ALT*E0001", "Identification": {"RemoteIdentification": {"EvcoID": "EMP1"}}, "StartTimestamp": "2023-01-01T00:00:00Z", "StopTimestamp": "2023-01-01T01:00:00Z", "MeterStart": 1000, "MeterStop": 1250, "StartInfo": {"timestamp": "2023-01-01T00:00:00Z"}, "StopInfo": {"timestamp": "2023-01-01T01:00:00Z"}, } ) assert response["StatusCode"]["Code"] == "000" assert captured["path"] == "cdrmgmt/v22/operators/DE-EMP/charge-detail-record" remote_payload = { "EvseID": "DE*WLE*E0001", "provider_id": "DE*ALT", "CPOPartnerSessionID": "sess-remote", } authorize_calls: list[tuple[str, dict[str, Any]]] = [] def fake_post_remote(self, path, payload): # type: ignore[override] authorize_calls.append((path, payload)) return {"StatusCode": {"Code": "000"}} monkeypatch.setattr( client, "_post", fake_post_remote.__get__(client, hubject_api.HubjectApiClient), ) client.authorize_start_evse(remote_payload) client.authorize_stop_evse({"EVSEId": "DE*WLE*E0002"}) normalized_operator = hubject_api._operator_id_for_path("DE*WLE") assert authorize_calls[0][0] == f"charging/v21/operators/{normalized_operator}/authorize/start" start_payload = authorize_calls[0][1] assert start_payload["OperatorID"] == "DE*WLE" assert start_payload["ProviderID"] == "DE*ALT" assert start_payload["EVSEId"] == "DE*WLE*E0001" assert authorize_calls[1][0] == f"charging/v21/operators/{normalized_operator}/authorize/stop" stop_payload = authorize_calls[1][1] assert stop_payload["OperatorID"] == "DE*WLE" assert stop_payload["ProviderID"] == "DE*EMP" assert stop_payload["EVSEId"] == "DE*WLE*E0002" payload = captured["payload"] assert isinstance(payload, dict) assert payload["SessionID"] == "sess-1" assert payload["HubOperatorID"] == "DE*ALT" assert payload["MeterValueStart"] == 1000.0 assert payload["MeterValueEnd"] == 1250.0 assert payload["ConsumedEnergy"] == 250.0 assert "StartInfo" not in payload assert "StopInfo" not in payload def test_hubject_api_client_authorize_start_falls_back_to_provider(monkeypatch): client = hubject_api.HubjectApiClient( "https://example.com/api/oicp/", partner_id="DE*EMP", ) captured: dict[str, object] = {} def fake_post(self, path, payload): # type: ignore[override] captured["path"] = path captured["payload"] = payload return {"StatusCode": {"Code": "000"}} monkeypatch.setattr( client, "_post", fake_post.__get__(client, hubject_api.HubjectApiClient), ) client.authorize_start_evse({"EVSEId": "TACW2244321T3066", "ProviderID": "FR-EMP"}) assert captured["path"] == "charging/v21/operators/FR-EMP/authorize/start" payload = captured["payload"] assert payload["OperatorID"] == "FR-EMP" assert payload["ProviderID"] == "FR-EMP" def test_hubject_api_client_send_cdr_normalizes_evse_identifier(monkeypatch): client = hubject_api.HubjectApiClient( "https://example.com/api/oicp/", operator_id="DE*WLE", ) captured: dict[str, object] = {} def fake_post(self, path, payload): # type: ignore[override] captured["payload"] = payload return {"StatusCode": {"Code": "000"}} monkeypatch.setattr( client, "_post", fake_post.__get__(client, hubject_api.HubjectApiClient), ) client.send_cdr( { "SessionID": "sess-evse", "EvseID": "DE*WLE*E123#2", "Identification": {"RemoteIdentification": {"EvcoID": "EMP1"}}, "ChargingStart": "2023-01-01T00:00:00Z", "ChargingEnd": "2023-01-01T01:00:00Z", } ) payload = captured.get("payload") assert isinstance(payload, dict) assert payload["EvseID"] == "DE*WLE*E123*2" def test_hubject_api_client_send_cdr_derives_evse_from_operator(monkeypatch): client = hubject_api.HubjectApiClient( "https://example.com/api/oicp/", operator_id="DE*WLE", ) captured: dict[str, object] = {} def fake_post(self, path, payload): # type: ignore[override] captured["payload"] = payload return {"StatusCode": {"Code": "000"}} monkeypatch.setattr( client, "_post", fake_post.__get__(client, hubject_api.HubjectApiClient), ) client.send_cdr( { "SessionID": "sess-derived", "EvseID": "CP9#1", "Identification": {"RemoteIdentification": {"EvcoID": "EMP1"}}, "ChargingStart": "2023-01-01T00:00:00Z", "ChargingEnd": "2023-01-01T01:00:00Z", } ) payload = captured.get("payload") assert isinstance(payload, dict) assert payload["EvseID"] == "DE*WLE*ECP9*1" def test_hubject_api_client_send_cdr_uses_default_partner(monkeypatch): client = hubject_api.HubjectApiClient( "https://example.com/api/oicp/", operator_id="DE*WLE", ) captured: dict[str, object] = {} def fake_post(self, path, payload): # type: ignore[override] captured["path"] = path captured["payload"] = payload return {"StatusCode": {"Code": "000"}} monkeypatch.setattr( client, "_post", fake_post.__get__(client, hubject_api.HubjectApiClient), ) client.send_cdr( { "SessionID": "sess-2", "EvseID": "DE*WLE*E0002", "Identification": {"RemoteIdentification": {"EvcoID": "EMP2"}}, "StartTimestamp": "2023-01-02T00:00:00Z", "StopTimestamp": "2023-01-02T01:00:00Z", "MeterStart": 0, "MeterStop": 0, } ) assert captured["path"] == "cdrmgmt/v22/operators/DE*WLE/charge-detail-record" payload = captured["payload"] assert isinstance(payload, dict) assert payload["HubOperatorID"] == "DE*WLE" def test_hubject_api_client_send_cdr_uses_partner_override(monkeypatch): client = hubject_api.HubjectApiClient( "https://example.com/api/oicp/", operator_id="DE*WLE", partner_id="FR*XYZ", ) captured: dict[str, object] = {} def fake_post(self, path, payload): # type: ignore[override] captured["path"] = path captured["payload"] = payload return {"StatusCode": {"Code": "000"}} monkeypatch.setattr( client, "_post", fake_post.__get__(client, hubject_api.HubjectApiClient), ) client.send_cdr( { "SessionID": "sess-3", "EvseID": "DE*WLE*E0003", "Identification": {"RemoteIdentification": {"EvcoID": "EMP3"}}, "StartTimestamp": "2023-01-03T00:00:00Z", "StopTimestamp": "2023-01-03T01:00:00Z", "MeterStart": 10, "MeterStop": 20, } ) assert captured["path"] == "cdrmgmt/v22/operators/FR-XYZ/charge-detail-record" payload = captured["payload"] assert isinstance(payload, dict) assert payload["HubOperatorID"] == "DE*WLE" def test_push_evse_data_success(client): flask_client, mock_server, dummy_hubject_client = client response = flask_client.post( "/api/oicp/evsepush/v23/operators/DE*PIP/data-records", data=json.dumps( { "ActionType": "update", "OperatorEvseData": [ { "OperatorName": {"En": "Pipelet"}, "EvseData": [ { "EvseID": "DE*PIP*E123", "EvseStatus": "Available", } ], } ], } ), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 body = response.get_json() assert body["Result"] is True assert body["OperatorID"] == "DE*PIP" assert "HubjectResponse" in body assert body["HubjectResponse"]["OperatorID"] == "DE*PIP" assert mock_server.queue.qsize() == 0 assert dummy_hubject_client.calls == [ ( "DE*PIP", { "ActionType": "update", "OperatorEvseData": { "OperatorName": "Pipelet", "EvseDataRecord": [ { "EvseID": "DE*PIP*E123", "EvseStatus": "Available", } ], "OperatorID": "DE*PIP", }, "OperatorID": "DE*PIP", }, ) ] def test_push_evse_data_merges_operator_entries(client): flask_client, _, dummy_hubject_client = client response = flask_client.post( "/api/oicp/evsepush/v23/operators/DE*PIP/data-records", data=json.dumps( { "ActionType": "insert", "OperatorEvseData": [ { "OperatorName": {"En": "Pipelet"}, "EvseData": [ {"EvseID": "DE*PIP*E1", "EvseStatus": "Available"} ], }, { "OperatorName": {"En": "Pipelet"}, "EvseDataRecord": [ {"EvseID": "DE*PIP*E2", "EvseStatus": "Occupied"} ], }, ], } ), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 normalized_payload = dummy_hubject_client.calls[-1][1]["OperatorEvseData"] assert normalized_payload["EvseDataRecord"] == [ {"EvseID": "DE*PIP*E1", "EvseStatus": "Available"}, {"EvseID": "DE*PIP*E2", "EvseStatus": "Occupied"}, ] assert normalized_payload["OperatorID"] == "DE*PIP" assert normalized_payload["OperatorName"] == "Pipelet" def test_push_evse_data_operator_name_localization(client): flask_client, _, dummy_hubject_client = client response = flask_client.post( "/api/oicp/evsepush/v23/operators/DE*LOC/data-records", data=json.dumps( { "ActionType": "update", "OperatorEvseData": { "OperatorName": {"De": "Betreiber", "En": "Operator"}, "EvseDataRecord": [ {"EvseID": "DE*LOC*E1", "EvseStatus": "Available"} ], }, } ), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 normalized_payload = dummy_hubject_client.calls[-1][1]["OperatorEvseData"] assert normalized_payload["OperatorName"] == "Operator" def test_push_evse_data_normalizes_calibration_law_availability(client): flask_client, _, dummy_hubject_client = client response = flask_client.post( "/api/oicp/evsepush/v23/operators/DE*CAL/data-records", data=json.dumps( { "ActionType": "update", "OperatorEvseData": { "EvseDataRecord": [ { "EvseID": "DE*CAL*E1", "EvseStatus": "Available", "CalibrationLawDataAvailability": "NotAvailable", } ], }, } ), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 evse_records = dummy_hubject_client.calls[-1][1]["OperatorEvseData"]["EvseDataRecord"] assert evse_records == [ { "EvseID": "DE*CAL*E1", "EvseStatus": "Available", "CalibrationLawDataAvailability": "Not Available", } ] def test_push_evse_status_validation_and_forwarding(client, monkeypatch): flask_client, mock_server, dummy_hubject_client = client invalid_response = flask_client.post( "/api/oicp/evsepush/v21/operators/DE*PIP/status-records", data=json.dumps( { "ActionType": "invalid", "OperatorEvseStatus": { "EvseStatusRecords": [ {"EvseID": "DE*PIP*E123", "EvseStatus": "Available"} ] }, } ), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert invalid_response.status_code == 400 invalid_body = invalid_response.get_json() assert invalid_body["Result"] is False assert dummy_hubject_client.status_calls == [] assert mock_server.queue.qsize() == 0 status_payload = { "ActionType": "update", "OperatorEvseStatus": [ { "OperatorName": {"En": "Pipelet"}, "EvseStatusRecords": [ {"EvseID": "DE*PIP*E123", "EvseStatus": "Available"} ], } ], } response = flask_client.post( "/api/oicp/evsepush/v21/operators/DE*PIP/status-records", data=json.dumps(status_payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert response.status_code == 200 body = response.get_json() assert body["Result"] is True assert body["OperatorID"] == "DE*PIP" assert "HubjectResponse" in body assert dummy_hubject_client.status_calls == [ ( "DE*PIP", { "ActionType": "update", "OperatorEvseStatus": { "OperatorName": "Pipelet", "EvseStatusRecords": [ {"EvseID": "DE*PIP*E123", "EvseStatus": "Available"} ], "OperatorID": "DE*PIP", }, "OperatorID": "DE*PIP", }, ) ] assert mock_server.queue.qsize() == 0 monkeypatch.setattr(hubject_api, "HUBJECT_API_CLIENT", None) fallback_response = flask_client.post( "/api/oicp/evsepush/v21/operators/DE*PIP/status-records", data=json.dumps(status_payload), headers={"Content-Type": "application/json"}, environ_overrides=_verified_environ(), ) assert fallback_response.status_code == 200 fallback_body = fallback_response.get_json() assert "OcppResponse" in fallback_body event_type, forwarded_payload, _ = mock_server.queue.get(timeout=1) assert event_type == "evse_status" assert forwarded_payload["OperatorID"] == "DE*PIP" assert forwarded_payload["ActionType"] == "update" assert forwarded_payload["OperatorEvseStatus"]["EvseStatusRecords"][0]["EvseID"] == "DE*PIP*E123" def test_robots_txt_accessible_without_client_cert(client): flask_client, _, __ = client response = flask_client.get("/robots.txt") assert response.status_code == 200 assert response.headers["Content-Type"].startswith("text/plain") assert response.get_data(as_text=True) == "User-agent: *\nDisallow: /\n" @pytest.mark.asyncio async def test_hubject_client_called_for_status_changes(monkeypatch): calls: list[tuple[str, str]] = [] class DummyHubjectClient: async def set_evse_available(self, evse_id, *, timestamp): calls.append(("available", evse_id)) return {}, 200, "{}" async def set_evse_occupied(self, evse_id, *, timestamp): calls.append(("occupied", evse_id)) return {}, 200, "{}" async def set_evse_status(self, evse_id, status, *, timestamp): calls.append((status.lower(), evse_id)) return {}, 200, "{}" monkeypatch.setattr(ocpp_server, "HUBJECT_CLIENT", DummyHubjectClient()) monkeypatch.setattr(ocpp_server, "_evse_status_cache", {}) monkeypatch.setattr(ocpp_server, "_evse_id_cache", {"CP1": "DE*PIP*E1"}) timestamp = datetime(2023, 7, 5, 12, 0, tzinfo=timezone.utc) await ocpp_server.process_status_notification_for_hubject( "CP1", 1, "Available", timestamp, ) assert calls == [("available", "DE*PIP*E1#1")] # Same status again should not trigger another call await ocpp_server.process_status_notification_for_hubject( "CP1", 1, "Available", timestamp, ) assert calls == [("available", "DE*PIP*E1#1")] await ocpp_server.process_status_notification_for_hubject( "CP1", 1, "Charging", timestamp, ) assert calls[-1] == ("occupied", "DE*PIP*E1#1") assert len(calls) == 2