import asyncio import json import sys import types import pytest pymysql_stub = types.ModuleType("pymysql") def _fake_connect(*args, **kwargs): # pragma: no cover - simple stub 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) if "aiohttp" not in sys.modules: fake_aiohttp = types.ModuleType("aiohttp") class _DummyClientSession: # pragma: no cover - simple stub async def __aenter__(self): return self async def __aexit__(self, exc_type, exc, tb): return False class _DummyClientTimeout: # pragma: no cover - simple stub def __init__(self, *args, **kwargs): return None class _DummyResponse: # pragma: no cover - simple stub def __init__(self, *args, **kwargs): self.args = args self.kwargs = kwargs def _dummy_json_response(*args, **kwargs): # pragma: no cover - simple stub return _DummyResponse(*args, **kwargs) class _DummyRouteTableDef: # pragma: no cover - simple stub def __init__(self): self._routes = [] def _register(self, *args, **kwargs): def decorator(func): self._routes.append((args, kwargs, func)) return func return decorator def __getattr__(self, name): return self._register class _DummyApplication: # pragma: no cover - simple stub def __init__(self, *args, **kwargs): self._added_routes = [] def add_routes(self, routes): self._added_routes.append(routes) return None class _DummyAppRunner: # pragma: no cover - simple stub def __init__(self, app): self.app = app async def setup(self): return None class _DummyTCPSite: # pragma: no cover - simple stub def __init__(self, runner, host, port): self.runner = runner self.host = host self.port = port async def start(self): return None fake_web = types.SimpleNamespace() fake_web.Request = type("Request", (), {}) fake_web.Response = _DummyResponse fake_web.json_response = _dummy_json_response fake_web.Application = _DummyApplication fake_web.RouteTableDef = _DummyRouteTableDef fake_web.AppRunner = _DummyAppRunner fake_web.TCPSite = _DummyTCPSite fake_aiohttp.ClientSession = _DummyClientSession fake_aiohttp.ClientTimeout = _DummyClientTimeout fake_aiohttp.web = fake_web sys.modules["aiohttp"] = fake_aiohttp if "websockets" not in sys.modules: fake_websockets = types.ModuleType("websockets") fake_websockets.legacy = types.ModuleType("websockets.legacy") legacy_server = types.ModuleType("websockets.legacy.server") class _DummyWebSocketServerProtocol: # pragma: no cover - simple stub pass async def _dummy_serve(*args, **kwargs): # pragma: no cover - simple stub return None legacy_server.WebSocketServerProtocol = _DummyWebSocketServerProtocol legacy_server.serve = _dummy_serve legacy_http = types.ModuleType("websockets.legacy.http") legacy_http.read_request = lambda *args, **kwargs: None exceptions_mod = types.ModuleType("websockets.exceptions") class _DummyWebSocketException(Exception): pass class _DummyInvalidMessage(_DummyWebSocketException): pass class _DummyConnectionClosed(_DummyWebSocketException): pass exceptions_mod.InvalidMessage = _DummyInvalidMessage exceptions_mod.WebSocketException = _DummyWebSocketException exceptions_mod.ConnectionClosed = _DummyConnectionClosed fake_websockets.legacy.server = legacy_server fake_websockets.legacy.http = legacy_http fake_websockets.exceptions = exceptions_mod sys.modules["websockets"] = fake_websockets sys.modules["websockets.legacy"] = fake_websockets.legacy sys.modules["websockets.legacy.server"] = legacy_server sys.modules["websockets.legacy.http"] = legacy_http sys.modules["websockets.exceptions"] = exceptions_mod import pipelet_ocpp_server as server class _FakeCursor: def __init__(self, connection): self._connection = connection self.queries: list[tuple[str, tuple]] = [] self._last_result = None def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def execute(self, query, params=None): params = params or tuple() self.queries.append((query, params)) normalized = query.lstrip().upper() if normalized.startswith("SELECT"): chargepoint_id = params[0] self._last_result = self._connection.select_latest(chargepoint_id) elif normalized.startswith("UPDATE"): payload, record_id = params self._connection.update_configuration(record_id, payload) self._last_result = None elif normalized.startswith("INSERT"): chargepoint_id, payload = params self._connection.insert_configuration(chargepoint_id, payload) self._last_result = None else: # pragma: no cover - defensive branch raise AssertionError(f"Unsupported query: {query}") def fetchone(self): return self._last_result class _FakeConnection: def __init__(self, rows): self.rows = list(rows) self._next_id = (max((row["id"] for row in self.rows), default=0) + 1) self.last_cursor: _FakeCursor | None = None self.commit_calls = 0 self.closed = False def cursor(self, cursorclass=None): # pylint: disable=unused-argument self.last_cursor = _FakeCursor(self) return self.last_cursor def commit(self): self.commit_calls += 1 def close(self): self.closed = True def select_latest(self, chargepoint_id): matches = [ row for row in self.rows if row["chargepoint_id"] == chargepoint_id ] if not matches: return None # Simulate ordering by created_at DESC, id DESC by relying on input order matches.sort(key=lambda row: (row.get("created_at"), row["id"]), reverse=True) return matches[0] def update_configuration(self, record_id, payload): for row in self.rows: if row["id"] == record_id: row["configuration_json"] = payload return raise AssertionError(f"Row with id {record_id} not found") def insert_configuration(self, chargepoint_id, payload): self.rows.append( { "id": self._next_id, "chargepoint_id": chargepoint_id, "configuration_json": payload, } ) self._next_id += 1 def test_store_get_configuration_updates_existing_row(monkeypatch): existing_row = { "id": 17, "chargepoint_id": "WALLBOX-1", "configuration_json": json.dumps({"old": True}), "created_at": "2023-01-01T10:00:00", } connection = _FakeConnection([existing_row]) monkeypatch.setattr(server, "get_db_conn", lambda: connection) new_configuration = {"configurationKey": [{"key": "HeartbeatInterval", "value": "120"}]} asyncio.run(server.store_get_configuration("WALLBOX-1", new_configuration)) assert connection.commit_calls == 1 assert connection.closed is True assert len(connection.rows) == 1 updated_row = connection.rows[0] assert updated_row["id"] == existing_row["id"] assert updated_row["configuration_json"] == json.dumps( new_configuration, ensure_ascii=False )