# list of additional paths to ignore in this project.
# Same syntax as gitignore, so you can use * and **.
+# Important: quote patterns that start with `*`, otherwise YAML treats them as aliases.
+# Example:
+# ignored_paths:
+# - "examples/**"
+# - ".worktrees/**"
+# - "**/bin/**"
+# - "**/obj/**"
# Note: global ignored_paths from serena_config.yml are also applied additively.
ignored_paths: []
# Find the list of tools here: https://oraios.github.io/serena/01-about/035_tools.html
fixed_tools: []
-# list of languages for which language servers are started (LSP backend only); choose from:
-# ada al angular ansible bash
-# bsl clojure cpp cpp_ccls crystal
-# csharp csharp_omnisharp cue dart elixir
-# elm erlang fortran fsharp gdscript
-# go groovy haskell haxe hlsl
-# html java json julia kotlin
-# latex lean4 lua luau markdown
-# matlab msl nix ocaml pascal
-# perl php php_phpactor php_phpantom powershell
-# python python_jedi python_pyrefly python_ty r
-# rego ruby ruby_solargraph rust scala
-# scss solidity svelte swift systemverilog
-# terraform toml typescript typescript_vts vue
-# yaml zig
-# (This list may be outdated; generated with scripts/print_language_list.py;
-# For the current list, see values of Language enum here:
-# https://github.com/oraios/serena/blob/main/src/solidlsp/ls_config.py)
-# For some languages, there are alternative language servers, e.g. csharp_omnisharp, ruby_solargraph.)
-# Note:
-# - For C, use cpp
-# - For JavaScript, use typescript
-# - For Angular projects, use angular (subsumes typescript+html; requires `npm install` in the project root)
-# - For Svelte projects, use svelte (subsumes typescript/javascript for .svelte projects; requires npm)
-# - For SCSS / Sass / plain CSS, use scss (some-sass-language-server handles all three)
-# - For Free Pascal/Lazarus, use pascal
-# Special requirements:
-# Some languages require additional setup/installations.
-# See here for details: https://oraios.github.io/serena/01-about/020_programming-languages.html#language-servers
-# When using multiple languages, the first language server that supports a given file will be used for that file.
-# The first language is the default language and the respective language server will be used as a fallback.
-# Note that when using the JetBrains backend, language servers are not used and this list is correspondingly ignored.
-languages:
- - typescript
-
# time budget (seconds) per tool call for the retrieval of additional symbol information
# such as docstrings or parameter information.
# This overrides the corresponding setting in the global configuration; see the documentation there.
# maximum time in seconds to wait for activation_command to complete before killing it (default 180s).
# must be a positive number.
activation_command_timeout: 180.0
+
+# list of language servers to start when using the LSP backend; choose from:
+# ada al angular ansible bash
+# bsl clojure cpp cpp_ccls crystal
+# csharp csharp_omnisharp cue dart elixir
+# elm erlang fortran fsharp gdscript
+# go groovy haskell haxe hlsl
+# html java json julia kotlin
+# latex lean4 lua luau markdown
+# matlab msl nix ocaml pascal
+# perl php php_phpactor php_phpantom powershell
+# python python_basedpyright python_jedi python_pyrefly python_ty
+# qml r rego ruby ruby_solargraph
+# rust scala scss solidity svelte
+# swift systemverilog terraform toml typescript
+# typescript_vts vue yaml zig
+# (This list may be outdated; generated with scripts/print_language_list.py;
+# For the current list, see values of the LanguageServerId enum here:
+# https://github.com/oraios/serena/blob/main/src/solidlsp/ls_config.py)
+# For some languages, there are several alternative language servers, e.g. csharp_omnisharp, ruby_solargraph.)
+# Note:
+# - For C, use cpp
+# - For JavaScript, use typescript
+# - For Angular projects, use angular (subsumes typescript+html; requires `npm install` in the project root)
+# - For Svelte projects, use svelte (subsumes typescript/javascript for .svelte projects; requires npm)
+# - For SCSS / Sass / plain CSS, use scss (some-sass-language-server handles all three)
+# - For Free Pascal/Lazarus, use pascal
+# Special requirements:
+# Some language servers require additional setup/installations.
+# See here for details: https://oraios.github.io/serena/01-about/020_programming-languages.html#language-servers
+# When using multiple language servers, the first language server that supports a given file will be used for that file.
+# The first language server is the default language and the respective language server will be used as a fallback.
+# Note that when using the JetBrains backend, language servers are not used and this list is correspondingly ignored.
+language_servers:
+ - typescript
})
await describe('all template files round-trip', async () => {
- await it('should validate all 15 station template files through the pipeline', async t => {
+ await it('should validate all station template files through the pipeline', async t => {
mockLoggerWarnDebug(t, logger)
const fs = await import('node:fs')
const path = await import('node:path')
const templateDir = path.join(import.meta.dirname, '../../src/assets/station-templates')
const files = fs.readdirSync(templateDir).filter(f => f.endsWith('.json'))
- assert.strictEqual(files.length, 15)
+ assert.ok(
+ files.length >= 15,
+ `expected at least 15 template files, found ${String(files.length)}`
+ )
for (const file of files) {
const content = fs.readFileSync(path.join(templateDir, file), 'utf8')
-# OCPP2 Mock Server
+# OCPP Mock Servers
-An OCPP 2.0.1 mock CSMS (Central System Management System) for end-to-end testing of charging station simulators.
+Mock CSMS (Central System Management System) servers for end-to-end testing of charging station simulators:
+
+- **`server.py`** — OCPP 2.0.1 mock CSMS
+- **`server16.py`** — OCPP 1.6 mock CSMS
## Prerequisites
poetry install --no-root
```
-## Running the Server
+## Running the Servers
+
+OCPP 2.0.1:
```shell
poetry run python server.py
```
-The server listens on `127.0.0.1:9000` by default.
+OCPP 1.6:
+
+```shell
+poetry run python server16.py
+```
+
+Both servers listen on `127.0.0.1:9000` by default.
## Configuration
These flags customize the payload of specific commands:
- `--trigger-message <TYPE>`: TriggerMessage requested message type (default: `StatusNotification`)
- - `StatusNotification`, `BootNotification`, `Heartbeat`, `MeterValues`, `FirmwareStatusNotification`, `LogStatusNotification`, `SignCertificate`
+ - `StatusNotification`, `BootNotification`, `Heartbeat`, `MeterValues`, `FirmwareStatusNotification`, `LogStatusNotification`, `PublishFirmwareStatusNotification`, `TransactionEvent`, `SignChargingStationCertificate`, `SignV2GCertificate`, `SignCombinedCertificate`
- `--reset-type <TYPE>`: Reset type (default: `Immediate`)
- `Immediate` — Reset now
- `OnIdle` — Reset when no transaction is active
- `--set-variables <SPECS>`: SetVariables data as `Component.Variable=Value,...` (values must not contain commas)
- `--get-variables <SPECS>`: GetVariables data as `Component.Variable,...`
- `--local-list-tokens TOKEN ...`: Tokens to include in SendLocalList (default: test token)
+- `--reservation-id <ID>`: ReserveNow/CancelReservation reservation id (default: `1`)
+- `--reserve-id-token <TOKEN>`: ReserveNow id_token value the reservation is bound to (default: `reserved_token`)
+- `--reserve-evse-id <ID>`: ReserveNow target EVSE id (default: `1`)
```shell
poetry run python server.py --command TriggerMessage --trigger-message BootNotification --delay 5
--set-variables "OCPPCommCtrlr.HeartbeatInterval=30,TxCtrlr.EVConnectionTimeOut=60"
poetry run python server.py --command GetLocalListVersion --delay 5
poetry run python server.py --command SendLocalList --delay 5 --local-list-tokens token1 token2
+poetry run python server.py --command ReserveNow --reserve-evse-id 1 --reserve-id-token mytag --reservation-id 42 --delay 5
```
## Supported OCPP 2.0.1 Messages
### Outgoing Commands (CSMS → CS)
+- `CancelReservation` — Cancel a reservation
- `CertificateSigned` — Send a signed certificate to the charging station
- `ChangeAvailability` — Change connector availability
- `ClearCache` — Clear the charging station cache
- `InstallCertificate` — Install a CA certificate
- `RequestStartTransaction` — Remote start a transaction
- `RequestStopTransaction` — Remote stop a transaction
+- `ReserveNow` — Reserve an EVSE
- `Reset` — Reset the charging station
- `SendLocalList` — Send a local authorization list update
- `SetNetworkProfile` — Set the network connection profile
The server tracks active transaction IDs from `TransactionEvent.Started` and uses real IDs in `RequestStopTransaction` and `GetTransactionStatus`. Falls back to a test ID when no transaction is active.
+## OCPP 1.6 Server (`server16.py`)
+
+An OCPP 1.6 mock CSMS. Runs independently of the 2.0.1 server; both share the same CLI conventions.
+
+### Server
+
+- `--host <HOST>`: Bind address (default: `127.0.0.1`)
+- `--port <PORT>`: Listening port (default: `9000`)
+- `--connector-id <ID>`: Connector id used in CSMS→CS commands (default: `1`)
+
+### Boot Behavior
+
+- `--boot-status <STATUS>`: Fixed BootNotification response status (default: `Accepted`)
+ - `Accepted` — Station registered
+ - `Pending` — Station not yet registered, must retry
+ - `Rejected` — Station rejected, must retry
+- `--boot-status-sequence <SEQ>`: Comma-separated status sequence (e.g., `Pending,Pending,Accepted`). Returns the next status on each BootNotification, stays on the last value once exhausted.
+
+`--boot-status` and `--boot-status-sequence` are mutually exclusive. `--boot-status X` is shorthand for `--boot-status-sequence X`.
+
+```shell
+poetry run python server16.py --boot-status Rejected
+poetry run python server16.py --boot-status-sequence Pending,Pending,Accepted
+```
+
+### Authorization
+
+- `--auth-mode <MODE>`: Authorization mode (default: `normal`)
+ - `normal` — Accept all id tags
+ - `whitelist` — Only accept id tags in the whitelist
+ - `blacklist` — Block id tags in the blacklist, accept all others
+- `--whitelist TAG ...`: Authorized id tags (default: `valid_token test_token authorized_user`)
+- `--blacklist TAG ...`: Blocked id tags (default: `blocked_token invalid_user`)
+- `--offline`: Simulate network failure (raises `InternalError` on Authorize)
+- `--auth-parent-id-tag <TAG>`: Include `parentIdTag` in Authorize and StartTransaction responses (default: none)
+
+```shell
+poetry run python server16.py --auth-mode whitelist --whitelist valid_token test_token
+poetry run python server16.py --auth-mode blacklist --blacklist blocked_token
+poetry run python server16.py --offline
+poetry run python server16.py --auth-parent-id-tag ParentTag
+```
+
+### OCPP Commands
+
+Send CSMS-initiated commands to connected charging stations.
+
+#### Single command
+
+- `--command <NAME>`: OCPP command to send (see supported commands below)
+- `--delay <SEC>`: One-shot delay before sending (mutually exclusive with `--period`)
+- `--period <SEC>`: Repeat interval in seconds (mutually exclusive with `--delay`)
+
+`--delay` or `--period` is required when `--command` is specified.
+
+```shell
+poetry run python server16.py --command Reset --delay 5
+poetry run python server16.py --command TriggerMessage --period 10
+```
+
+#### Command sequence
+
+- `--commands <SEQ>`: Comma-separated `CMD:DELAY` pairs, executed sequentially (e.g., `RemoteStartTransaction:5,RemoteStopTransaction:30`)
+
+Mutually exclusive with `--command`.
+
+```shell
+poetry run python server16.py --commands "RemoteStartTransaction:5,RemoteStopTransaction:30"
+```
+
+#### Command-specific options
+
+- `--trigger-message <TYPE>`: TriggerMessage requested message type (default: `StatusNotification`)
+ - `StatusNotification`, `BootNotification`, `Heartbeat`, `MeterValues`, `FirmwareStatusNotification`, `DiagnosticsStatusNotification`, `LogStatusNotification`, `SignChargePointCertificate`
+- `--reset-type <TYPE>`: Reset type (default: `Hard`)
+ - `Hard` — Reset now
+ - `Soft` — Graceful reset
+- `--availability-type <TYPE>`: ChangeAvailability type (default: `Operative`)
+ - `Operative` — Connector available
+ - `Inoperative` — Connector unavailable
+- `--reserve-connector-id <ID>`: ReserveNow connector id (default: `1`)
+- `--reserve-id-tag <TAG>`: ReserveNow id tag reserving the connector (default: `reserved_tag`)
+- `--reservation-id <ID>`: ReserveNow/CancelReservation reservation id (default: `1`)
+
+```shell
+poetry run python server16.py --command TriggerMessage --trigger-message BootNotification --delay 5
+poetry run python server16.py --command Reset --reset-type Soft --delay 5
+poetry run python server16.py --command ChangeAvailability --availability-type Inoperative --delay 5
+poetry run python server16.py --command ReserveNow --reserve-connector-id 1 --reserve-id-tag mytag --reservation-id 42 --delay 5
+```
+
+### Supported OCPP 1.6 Messages
+
+#### Outgoing Commands (CSMS → CS)
+
+- `TriggerMessage` — Trigger a specific message from the station
+- `RemoteStartTransaction` — Remote start a transaction
+- `RemoteStopTransaction` — Remote stop a transaction
+- `Reset` — Reset the charging station
+- `ChangeAvailability` — Change connector availability
+- `UpdateFirmware` — Request firmware update
+- `GetDiagnostics` — Request diagnostics upload
+- `ReserveNow` — Reserve a connector
+- `CancelReservation` — Cancel a reservation
+
+#### Incoming Handlers (CS → CSMS)
+
+- `Authorize` — Handle authorization requests (configurable auth modes)
+- `BootNotification` — Handle boot notification (configurable status sequence)
+- `Heartbeat` — Handle heartbeat messages
+- `StartTransaction` — Handle transaction start
+- `StopTransaction` — Handle transaction stop (logs signed transaction data)
+- `MeterValues` — Handle meter value reports (logs `SignedData`-format samples)
+- `StatusNotification` — Handle connector status notifications
+- `DataTransfer` — Handle vendor-specific data transfer
+- `FirmwareStatusNotification` — Handle firmware update status
+- `DiagnosticsStatusNotification` — Handle diagnostics upload status
+
+### Transaction Tracking
+
+The server tracks active transaction IDs from `StartTransaction` and uses real IDs in `RemoteStopTransaction`. Falls back to transaction ID `1` when no transaction is active.
+
## Development
### Code formatting
--- /dev/null
+"""OCPP-version-agnostic helpers shared by the 1.6 and 2.0.1 mock servers."""
+
+import argparse
+import asyncio
+import logging
+import math
+import signal
+import sys
+from enum import StrEnum
+from typing import TypeVar
+
+ActionT = TypeVar("ActionT", bound=StrEnum)
+
+DEFAULT_WHITELIST: tuple[str, ...] = ("valid_token", "test_token", "authorized_user")
+DEFAULT_BLACKLIST: tuple[str, ...] = ("blocked_token", "invalid_user")
+
+
+def check_positive_number(value: str) -> float:
+ try:
+ number = float(value)
+ except ValueError:
+ raise argparse.ArgumentTypeError("must be a number") from None
+ if not math.isfinite(number):
+ raise argparse.ArgumentTypeError("must be a finite number")
+ if number <= 0:
+ raise argparse.ArgumentTypeError("must be a positive number")
+ return number
+
+
+def parse_commands(
+ commands_str: str, action_type: type[ActionT]
+) -> list[tuple[ActionT, float]]:
+ result: list[tuple[ActionT, float]] = []
+ for raw_entry in commands_str.split(","):
+ entry = raw_entry.strip()
+ if not entry:
+ continue
+ if ":" not in entry:
+ raise argparse.ArgumentTypeError(
+ f"Invalid command entry '{entry}': expected 'CMD:DELAY' format"
+ )
+ cmd_str, delay_str = entry.split(":", 1)
+ try:
+ cmd = action_type(cmd_str.strip())
+ except ValueError:
+ raise argparse.ArgumentTypeError(
+ f"Unknown action: '{cmd_str.strip()}'"
+ ) from None
+ try:
+ delay = float(delay_str.strip())
+ except ValueError:
+ raise argparse.ArgumentTypeError(
+ f"Invalid delay '{delay_str.strip()}': must be a number"
+ ) from None
+ if not math.isfinite(delay) or delay <= 0:
+ raise argparse.ArgumentTypeError(
+ f"Delay must be a finite positive number, got {delay}"
+ )
+ result.append((cmd, delay))
+ return result
+
+
+async def negotiate_subprotocol(websocket, logger: logging.Logger) -> bool:
+ try:
+ requested_protocols = websocket.request.headers["Sec-WebSocket-Protocol"]
+ except KeyError:
+ logger.info("Client hasn't requested any Subprotocol. Closing Connection")
+ await websocket.close()
+ return False
+
+ if websocket.subprotocol:
+ logger.info("Protocols Matched: %s", websocket.subprotocol)
+ return True
+
+ logger.warning(
+ "Protocols Mismatched | Expected Subprotocols: %s,"
+ " but client supports %s | Closing connection",
+ websocket.available_subprotocols,
+ requested_protocols,
+ )
+ await websocket.close()
+ return False
+
+
+async def install_signal_handlers_and_wait(
+ loop: asyncio.AbstractEventLoop,
+ server,
+ shutdown_event: asyncio.Event,
+ shutdown_timeout: float,
+ logger: logging.Logger,
+) -> None:
+ shutdown_count = 0
+
+ def _on_signal(sig: signal.Signals) -> None:
+ nonlocal shutdown_count
+ shutdown_count += 1
+ if shutdown_count == 1:
+ logger.info("Received %s, initiating graceful shutdown...", sig.name)
+ server.close()
+ shutdown_event.set()
+ else:
+ logger.warning("Received %s again, forcing exit", sig.name)
+ # Unix convention: fatal-signal exit status is 128 + signal number.
+ sys.exit(128 + sig.value)
+
+ for sig in (signal.SIGINT, signal.SIGTERM):
+ try:
+ loop.add_signal_handler(sig, _on_signal, sig)
+ except NotImplementedError:
+ # Windows: ProactorEventLoop doesn't support add_signal_handler.
+ # signal.signal() fires outside the event loop, so schedule
+ # _on_signal into the loop via call_soon_threadsafe.
+ def _signal_handler(
+ _signum: int,
+ _frame: object,
+ s: signal.Signals = sig,
+ ) -> None:
+ loop.call_soon_threadsafe(_on_signal, s)
+
+ signal.signal(sig, _signal_handler)
+
+ await shutdown_event.wait()
+
+ try:
+ async with asyncio.timeout(shutdown_timeout):
+ await server.wait_closed()
+ except TimeoutError:
+ logger.warning(
+ "Shutdown timed out after %.0fs — connections may not have closed cleanly",
+ shutdown_timeout,
+ )
[project]
name = "ocpp-server"
version = "4.10.1"
-description = "OCPP2 mock server"
+description = "OCPP 1.6 and 2.0.1 mock CSMS servers for charging station simulator testing"
authors = [{ name = "Jérôme Benoit", email = "jerome.benoit@sap.com" }]
readme = "README.md"
requires-python = ">=3.12,<4.0"
[tool.taskipy.tasks]
server = "poetry run python server.py"
+server16 = "poetry run python server16.py"
test = "poetry run pytest -v"
test_coverage = "poetry run pytest -v --cov --cov-report=term-missing"
-typecheck = "poetry run mypy server.py timer.py test_server.py test_timer.py"
+typecheck = "poetry run mypy server.py server16.py timer.py _common.py test_server.py test_server16.py test_timer.py"
format = "ruff check --fix . && ruff format ."
lint = "ruff check --diff . && ruff format --check --diff ."
asyncio_mode = "auto"
[tool.mypy]
-python_version = "3.14"
+python_version = "3.12"
warn_return_any = true
warn_unused_configs = true
warn_redundant_casts = true
ignore_missing_imports = true
[[tool.mypy.overrides]]
-module = ["test_server", "test_timer"]
+module = ["test_server", "test_server16", "test_timer"]
disallow_untyped_defs = false
[tool.coverage.run]
branch = true
-source = ["server", "timer"]
+source = ["server", "server16", "timer", "_common"]
[tool.coverage.report]
show_missing = true
import argparse
import asyncio
+import contextlib
import logging
-import math
-import signal
import sys
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from ocpp.v201.enums import (
Action,
AuthorizationStatusEnumType,
+ CancelReservationStatusEnumType,
CertificateSignedStatusEnumType,
CertificateSigningUseEnumType,
ChangeAvailabilityStatusEnumType,
OperationalStatusEnumType,
RegistrationStatusEnumType,
ReportBaseEnumType,
+ ReserveNowStatusEnumType,
ResetEnumType,
ResetStatusEnumType,
SendLocalListStatusEnumType,
)
from websockets import ConnectionClosed
+from _common import (
+ DEFAULT_BLACKLIST,
+ DEFAULT_WHITELIST,
+ check_positive_number,
+ install_signal_handlers_and_wait,
+ negotiate_subprotocol,
+ parse_commands,
+)
from timer import Timer
logger = logging.getLogger(__name__)
DEFAULT_LOG_URL = "https://example.com/logs"
DEFAULT_LOCAL_LIST_VERSION = 1
DEFAULT_CUSTOMER_ID = "test_customer_001"
+DEFAULT_RESERVATION_ID = 1
+DEFAULT_RESERVE_ID_TOKEN = "reserved_token" # noqa: S105
+DEFAULT_RESERVATION_EXPIRY_SECONDS = 3600
FALLBACK_TRANSACTION_ID = "test_transaction_123"
MAX_REQUEST_ID = 2**31 - 1
SHUTDOWN_TIMEOUT_SECONDS = 30.0
whitelist = "whitelist"
blacklist = "blacklist"
rate_limit = "rate_limit"
- offline = "offline"
@dataclass(frozen=True)
total_cost: float
# Intentionally mutable despite frozen dataclass
charge_points: set["ChargePoint"]
- # Shared mutable counter so boot_sequence advances across reconnections
+ # Shared mutable counter so boot_sequence advances across reconnections.
+ # NOTE: on_connect passes this same list to every ChargePoint, so the boot
+ # index is shared across ALL stations on this server, not per-station: the
+ # Nth BootNotification server-wide gets the Nth boot_sequence status.
boot_index: list[int] = field(default_factory=lambda: [0])
commands: list[tuple[Action, float]] | None = None
trigger_message_type: MessageTriggerEnumType = (
set_variables_data: list[dict] | None = None
get_variables_data: list[dict] | None = None
local_list_tokens: list[str] | None = None
+ reservation_id: int = DEFAULT_RESERVATION_ID
+ reserve_id_token: str = DEFAULT_RESERVE_ID_TOKEN
+ reserve_evse_id: int = DEFAULT_EVSE_ID
class ChargePoint(ocpp.v201.ChargePoint):
_set_variables_data: list[dict] | None
_get_variables_data: list[dict] | None
_local_list_tokens: list[str] | None
+ _reservation_id: int
+ _reserve_id_token: str
+ _reserve_evse_id: int
def __init__(
self,
set_variables_data: list[dict] | None = None,
get_variables_data: list[dict] | None = None,
local_list_tokens: list[str] | None = None,
+ reservation_id: int = DEFAULT_RESERVATION_ID,
+ reserve_id_token: str = DEFAULT_RESERVE_ID_TOKEN,
+ reserve_evse_id: int = DEFAULT_EVSE_ID,
):
# Extract CP ID from last URL segment (OCPP 2.0.1 Part 4)
cp_id = connection.request.path.strip("/").split("/")[-1]
self._set_variables_data = set_variables_data
self._get_variables_data = get_variables_data
self._local_list_tokens = local_list_tokens
+ self._reservation_id = reservation_id
+ self._reserve_id_token = reserve_id_token
+ self._reserve_evse_id = reserve_evse_id
self._charge_points.add(self)
self._active_transactions: dict[str, int] = {}
if auth_config is None:
self._auth_config = AuthConfig(
mode=AuthMode.normal,
- whitelist=("valid_token", "test_token", "authorized_user"),
- blacklist=("blocked_token", "invalid_user"),
+ whitelist=DEFAULT_WHITELIST,
+ blacklist=DEFAULT_BLACKLIST,
offline=False,
default_status=AuthorizationStatusEnumType.accepted,
)
)
case AuthMode.rate_limit:
return AuthorizationStatusEnumType.not_at_this_time
- case _:
- return self._auth_config.default_status
+ return self._auth_config.default_status
def _build_id_token_info(self, token_id: str) -> dict:
"""Build id_token_info dict with optional groupIdToken and cacheExpiry."""
logger.info("Received %s Started", Action.transaction_event)
transaction_id = transaction_info.get("transaction_id", "")
- evse_id = kwargs.get("evse", {}).get("id", 0)
+ evse_id = (kwargs.get("evse") or {}).get("id", 0)
if transaction_id:
self._active_transactions[transaction_id] = evse_id
else:
# --- Outgoing commands (CSMS → CS) ---
async def _call_and_log(self, request, action: Action, success_status) -> None:
- """Send an OCPP request and log success or failure."""
+ """Send an OCPP request and log success or failure based on its status."""
response = await self.call(request, suppress=False)
if response.status == success_status:
logger.info("%s successful", action)
else:
- logger.info("%s failed", action)
+ logger.info("%s failed: %s", action, response.status)
async def _send_clear_cache(self):
request = ocpp.v201.call.ClearCache()
request, Action.update_firmware, UpdateFirmwareStatusEnumType.accepted
)
+ async def _send_reserve_now(self):
+ expiry_date_time = datetime.now(timezone.utc) + timedelta(
+ seconds=DEFAULT_RESERVATION_EXPIRY_SECONDS
+ )
+ request = ocpp.v201.call.ReserveNow(
+ id=self._reservation_id,
+ expiry_date_time=expiry_date_time.isoformat(),
+ id_token={"id_token": self._reserve_id_token, "type": DEFAULT_TOKEN_TYPE},
+ evse_id=self._reserve_evse_id,
+ )
+ await self._call_and_log(
+ request, Action.reserve_now, ReserveNowStatusEnumType.accepted
+ )
+
+ async def _send_cancel_reservation(self):
+ request = ocpp.v201.call.CancelReservation(reservation_id=self._reservation_id)
+ await self._call_and_log(
+ request,
+ Action.cancel_reservation,
+ CancelReservationStatusEnumType.accepted,
+ )
+
# --- Command dispatch ---
+ # Intentional subset of CSMS→CS commands supported by this mock.
+ # Any Action absent from this map is handled by _send_command via
+ # logger.warning only — no request is sent and no error is raised.
_COMMAND_HANDLERS: ClassVar[dict[Action, str]] = {
Action.clear_cache: "_send_clear_cache",
Action.get_base_report: "_send_get_base_report",
Action.install_certificate: "_send_install_certificate",
Action.set_network_profile: "_send_set_network_profile",
Action.update_firmware: "_send_update_firmware",
+ Action.reserve_now: "_send_reserve_now",
+ Action.cancel_reservation: "_send_cancel_reservation",
}
async def _send_command(self, command_name: Action):
await asyncio.sleep(delay)
await self._send_command(command_name)
- def handle_connection_closed(self):
+ def handle_connection_closed(self) -> None:
logger.info("ChargePoint %s closed connection", self.id)
if self._command_timer:
self._command_timer.cancel()
config: ServerConfig,
):
"""Handle new WebSocket connections from charge points."""
- try:
- requested_protocols = websocket.request.headers["Sec-WebSocket-Protocol"]
- except KeyError:
- logger.info("Client hasn't requested any Subprotocol. Closing Connection")
- return await websocket.close()
-
- if websocket.subprotocol:
- logger.info("Protocols Matched: %s", websocket.subprotocol)
- else:
- logger.warning(
- "Protocols Mismatched | Expected Subprotocols: %s,"
- " but client supports %s | Closing connection",
- websocket.available_subprotocols,
- requested_protocols,
- )
- return await websocket.close()
+ if not await negotiate_subprotocol(websocket, logger):
+ return
charge_points: set[ChargePoint] = config.charge_points
cp = ChargePoint(
set_variables_data=config.set_variables_data,
get_variables_data=config.get_variables_data,
local_list_tokens=config.local_list_tokens,
+ reservation_id=config.reservation_id,
+ reserve_id_token=config.reserve_id_token,
+ reserve_evse_id=config.reserve_evse_id,
)
if config.command_name:
await cp.send_command(config.command_name, config.delay, config.period)
cp.handle_connection_closed()
-def check_positive_number(value):
- try:
- value = float(value)
- except ValueError:
- raise argparse.ArgumentTypeError("must be a number") from None
- if not math.isfinite(value):
- raise argparse.ArgumentTypeError("must be a finite number")
- if value <= 0:
- raise argparse.ArgumentTypeError("must be a positive number")
- return value
-
-
def _parse_commands(commands_str: str) -> list[tuple[Action, float]]:
- result: list[tuple[Action, float]] = []
- for entry in commands_str.split(","):
- entry = entry.strip()
- if not entry:
- continue
- if ":" not in entry:
- raise argparse.ArgumentTypeError(
- f"Invalid command entry '{entry}': expected 'CMD:DELAY' format"
- )
- cmd_str, delay_str = entry.split(":", 1)
- try:
- cmd = Action(cmd_str.strip())
- except ValueError:
- raise argparse.ArgumentTypeError(
- f"Unknown action: '{cmd_str.strip()}'"
- ) from None
- try:
- delay = float(delay_str.strip())
- except ValueError:
- raise argparse.ArgumentTypeError(
- f"Invalid delay '{delay_str.strip()}': must be a number"
- ) from None
- if not math.isfinite(delay) or delay <= 0:
- raise argparse.ArgumentTypeError(
- f"Delay must be a finite positive number, got {delay}"
- )
- result.append((cmd, delay))
- return result
+ return parse_commands(commands_str, Action)
def _parse_variable_specs(specs_str: str, require_value: bool = False) -> list[dict]:
async def main():
- parser = argparse.ArgumentParser(description="OCPP2 Server")
+ parser = argparse.ArgumentParser(description="OCPP 2.0.1 Server")
command_group = parser.add_mutually_exclusive_group()
command_group.add_argument("-c", "--command", type=Action, help="command name")
command_group.add_argument(
default=OperationalStatusEnumType.operative,
help="ChangeAvailability status: Operative, Inoperative (default: Operative)",
)
+ parser.add_argument(
+ "--reservation-id",
+ type=int,
+ default=DEFAULT_RESERVATION_ID,
+ help=(
+ "ReserveNow/CancelReservation reservation id"
+ f" (default: {DEFAULT_RESERVATION_ID})"
+ ),
+ )
+ parser.add_argument(
+ "--reserve-id-token",
+ type=str,
+ default=DEFAULT_RESERVE_ID_TOKEN,
+ help=(
+ "ReserveNow id_token value the reservation is bound to"
+ f" (default: {DEFAULT_RESERVE_ID_TOKEN})"
+ ),
+ )
+ parser.add_argument(
+ "--reserve-evse-id",
+ type=int,
+ default=DEFAULT_EVSE_ID,
+ help=f"ReserveNow target EVSE id (default: {DEFAULT_EVSE_ID})",
+ )
# Auth configuration
parser.add_argument(
"--auth-mode",
- type=str,
- choices=["normal", "whitelist", "blacklist", "rate_limit"],
- default="normal",
- help="Authorization mode (default: normal)",
+ type=AuthMode,
+ default=AuthMode.normal,
+ help=(
+ "Authorization mode: normal, whitelist, blacklist, rate_limit"
+ " (default: normal)"
+ ),
)
parser.add_argument(
"--whitelist",
type=str,
nargs="+",
- default=["valid_token", "test_token", "authorized_user"],
+ default=list(DEFAULT_WHITELIST),
help="Whitelist of authorized tokens (space-separated)",
)
parser.add_argument(
"--blacklist",
type=str,
nargs="+",
- default=["blocked_token", "invalid_user"],
+ default=list(DEFAULT_BLACKLIST),
help="Blacklist of blocked tokens (space-separated)",
)
parser.add_argument(
)
boot_sequence_items.append(status)
boot_sequence = tuple(boot_sequence_items)
- if not boot_sequence:
- parser.error("--boot-status-sequence must contain at least one status")
elif args.boot_status is not None:
boot_sequence = (args.boot_status,)
else:
boot_sequence = (RegistrationStatusEnumType.accepted,)
auth_config = AuthConfig(
- mode=AuthMode(args.auth_mode),
+ mode=args.auth_mode,
whitelist=tuple(args.whitelist),
blacklist=tuple(args.blacklist),
offline=args.offline,
set_variables_data=parsed_set_variables,
get_variables_data=parsed_get_variables,
local_list_tokens=args.local_list_tokens,
+ reservation_id=args.reservation_id,
+ reserve_id_token=args.reserve_id_token,
+ reserve_evse_id=args.reserve_evse_id,
)
logger.info(
)
loop = asyncio.get_running_loop()
- shutdown_count = 0
shutdown_event = asyncio.Event()
async with websockets.serve(
subprotocols=SUBPROTOCOLS,
) as server:
logger.info("WebSocket Server Started on %s:%d", args.host, args.port)
-
- def _on_signal(sig: signal.Signals) -> None:
- nonlocal shutdown_count
- shutdown_count += 1
- if shutdown_count == 1:
- logger.info("Received %s, initiating graceful shutdown...", sig.name)
- server.close()
- shutdown_event.set()
- else:
- logger.warning("Received %s again, forcing exit", sig.name)
- sys.exit(128 + sig.value)
-
- for sig in (signal.SIGINT, signal.SIGTERM):
- try:
- loop.add_signal_handler(sig, _on_signal, sig)
- except NotImplementedError:
- # Windows: ProactorEventLoop doesn't support add_signal_handler.
- # signal.signal() fires outside the event loop, so schedule
- # _on_signal into the loop via call_soon_threadsafe.
- def _signal_handler(
- _signum: int,
- _frame: object,
- s: signal.Signals = sig,
- ) -> None:
- loop.call_soon_threadsafe(_on_signal, s)
-
- signal.signal(sig, _signal_handler)
-
- await shutdown_event.wait()
-
- try:
- async with asyncio.timeout(SHUTDOWN_TIMEOUT_SECONDS):
- await server.wait_closed()
- except TimeoutError:
- logger.warning(
- "Shutdown timed out after %.0fs"
- " — connections may not have closed cleanly",
- SHUTDOWN_TIMEOUT_SECONDS,
- )
+ await install_signal_handlers_and_wait(
+ loop, server, shutdown_event, SHUTDOWN_TIMEOUT_SECONDS, logger
+ )
logger.info("Server shutdown complete")
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
- try:
+ with contextlib.suppress(KeyboardInterrupt):
asyncio.run(main())
- except KeyboardInterrupt:
- pass
sys.exit(0)
--- /dev/null
+"""OCPP 1.6 mock server for e-mobility charging station simulator testing."""
+
+import argparse
+import asyncio
+import contextlib
+import logging
+import sys
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from enum import StrEnum
+from functools import partial
+from random import randint
+from typing import ClassVar
+
+import ocpp.v16
+import websockets
+from ocpp.exceptions import InternalError, OCPPError
+from ocpp.routing import on
+from ocpp.v16.enums import (
+ Action,
+ AuthorizationStatus,
+ AvailabilityStatus,
+ AvailabilityType,
+ CancelReservationStatus,
+ DataTransferStatus,
+ MessageTrigger,
+ RegistrationStatus,
+ RemoteStartStopStatus,
+ ReservationStatus,
+ ResetStatus,
+ ResetType,
+ TriggerMessageStatus,
+ ValueFormat,
+)
+from websockets import ConnectionClosed
+
+from _common import (
+ DEFAULT_BLACKLIST,
+ DEFAULT_WHITELIST,
+ check_positive_number,
+ install_signal_handlers_and_wait,
+ negotiate_subprotocol,
+ parse_commands,
+)
+from timer import Timer
+
+logger = logging.getLogger(__name__)
+
+# Server defaults
+DEFAULT_HOST = "127.0.0.1"
+DEFAULT_PORT = 9000
+DEFAULT_HEARTBEAT_INTERVAL_SECONDS = 60
+DEFAULT_CONNECTOR_ID = 1
+DEFAULT_TEST_TOKEN = "test_token" # noqa: S105
+DEFAULT_FIRMWARE_URL = "https://example.com/firmware/v1.6.bin"
+DEFAULT_DIAGNOSTICS_URL = "https://example.com/diagnostics"
+DEFAULT_RESERVE_CONNECTOR_ID = 1
+DEFAULT_RESERVE_ID_TAG = "reserved_tag"
+DEFAULT_RESERVATION_ID = 1
+DEFAULT_RESERVATION_EXPIRY_SECONDS = 3600
+FALLBACK_TRANSACTION_ID = 1
+MAX_TRANSACTION_ID = 2**31 - 1
+SHUTDOWN_TIMEOUT_SECONDS = 30.0
+SUBPROTOCOLS: list[websockets.Subprotocol] = [
+ websockets.Subprotocol("ocpp1.6"),
+]
+
+
+def _log_signed_meter_values(meter_value: list) -> None:
+ """Log signed meter value details from a list of OCPP 1.6 meter value dicts.
+
+ In OCPP 1.6 a signed reading is carried by a SampledValue whose ``format``
+ is ``SignedData`` (the ``value`` field then holds the signed blob). This is
+ used to exercise the P6C.5 signed meter value test case.
+ """
+ for mv in meter_value:
+ for sv in mv.get("sampled_value", []):
+ if sv.get("format") == ValueFormat.signed_data.value:
+ logger.info(
+ "Received signed meter value (SignedData):"
+ " measurand=%s, context=%s, value=%s",
+ sv.get("measurand"),
+ sv.get("context"),
+ sv.get("value"),
+ )
+
+
+def _random_transaction_id() -> int:
+ """Generate a random OCPP 1.6 transaction ID within the valid range."""
+ return randint(1, MAX_TRANSACTION_ID) # noqa: S311
+
+
+class AuthMode(StrEnum):
+ """Authorization modes for testing different authentication scenarios."""
+
+ normal = "normal"
+ whitelist = "whitelist"
+ blacklist = "blacklist"
+
+
+@dataclass(frozen=True)
+class AuthConfig:
+ """Authorization configuration for a charge point."""
+
+ mode: AuthMode
+ whitelist: tuple[str, ...]
+ blacklist: tuple[str, ...]
+ offline: bool
+ default_status: AuthorizationStatus
+ parent_id_tag: str | None = None
+
+
+@dataclass(frozen=True)
+class ServerConfig:
+ """Server-level configuration passed to each connection handler."""
+
+ command_name: Action | None
+ delay: float | None
+ period: float | None
+ auth_config: AuthConfig
+ boot_sequence: tuple[RegistrationStatus, ...]
+ connector_id: int
+ # Intentionally mutable despite frozen dataclass
+ charge_points: set["ChargePoint"]
+ # Shared mutable counter so boot_sequence advances across reconnections.
+ # NOTE: on_connect passes this same list to every ChargePoint, so the boot
+ # index is shared across ALL stations on this server, not per-station: the
+ # Nth BootNotification server-wide gets the Nth boot_sequence status.
+ boot_index: list[int] = field(default_factory=lambda: [0])
+ commands: list[tuple[Action, float]] | None = None
+ trigger_message_type: MessageTrigger = MessageTrigger.status_notification
+ reset_type: ResetType = ResetType.hard
+ availability_type: AvailabilityType = AvailabilityType.operative
+ reserve_connector_id: int = DEFAULT_RESERVE_CONNECTOR_ID
+ reserve_id_tag: str = DEFAULT_RESERVE_ID_TAG
+ reservation_id: int = DEFAULT_RESERVATION_ID
+
+
+class ChargePoint(ocpp.v16.ChargePoint):
+ """OCPP 1.6 charge point handler with configurable behavior for testing."""
+
+ _command_timer: Timer | None
+ _commands_task: asyncio.Task[None] | None
+ _auth_config: AuthConfig
+ _boot_sequence: tuple[RegistrationStatus, ...]
+ _boot_index: list[int]
+ _connector_id: int
+ _trigger_message_type: MessageTrigger
+ _reset_type: ResetType
+ _availability_type: AvailabilityType
+ _reserve_connector_id: int
+ _reserve_id_tag: str
+ _reservation_id: int
+ _charge_points: set["ChargePoint"]
+ _active_transactions: dict[int, int]
+
+ def __init__(
+ self,
+ connection,
+ auth_config: AuthConfig | None = None,
+ boot_sequence: tuple[RegistrationStatus, ...] = (RegistrationStatus.accepted,),
+ boot_index: list[int] | None = None,
+ connector_id: int = DEFAULT_CONNECTOR_ID,
+ trigger_message_type: MessageTrigger = MessageTrigger.status_notification,
+ reset_type: ResetType = ResetType.hard,
+ availability_type: AvailabilityType = AvailabilityType.operative,
+ reserve_connector_id: int = DEFAULT_RESERVE_CONNECTOR_ID,
+ reserve_id_tag: str = DEFAULT_RESERVE_ID_TAG,
+ reservation_id: int = DEFAULT_RESERVATION_ID,
+ charge_points: set["ChargePoint"] | None = None,
+ ):
+ # Extract CP ID from last URL segment (OCPP 1.6 uses ws://host/<cpId>)
+ cp_id = connection.request.path.strip("/").split("/")[-1]
+ if cp_id == "":
+ logger.warning(
+ "Empty CP ID extracted from path: %s", connection.request.path
+ )
+ super().__init__(cp_id, connection)
+ self._charge_points = charge_points if charge_points is not None else set()
+ self._command_timer = None
+ self._commands_task = None
+ self._boot_sequence = boot_sequence
+ if not self._boot_sequence:
+ raise ValueError("boot_sequence must contain at least one status")
+ self._boot_index = boot_index if boot_index is not None else [0]
+ self._connector_id = connector_id
+ self._trigger_message_type = trigger_message_type
+ self._reset_type = reset_type
+ self._availability_type = availability_type
+ self._reserve_connector_id = reserve_connector_id
+ self._reserve_id_tag = reserve_id_tag
+ self._reservation_id = reservation_id
+ self._charge_points.add(self)
+ self._active_transactions = {}
+ if auth_config is None:
+ self._auth_config = AuthConfig(
+ mode=AuthMode.normal,
+ whitelist=DEFAULT_WHITELIST,
+ blacklist=DEFAULT_BLACKLIST,
+ offline=False,
+ default_status=AuthorizationStatus.accepted,
+ )
+ else:
+ self._auth_config = auth_config
+
+ def _resolve_auth_status(self, id_tag: str) -> AuthorizationStatus:
+ """Resolve authorization status based on auth mode and id tag."""
+ match self._auth_config.mode:
+ case AuthMode.whitelist:
+ return (
+ AuthorizationStatus.accepted
+ if id_tag in self._auth_config.whitelist
+ else AuthorizationStatus.blocked
+ )
+ case AuthMode.blacklist:
+ return (
+ AuthorizationStatus.blocked
+ if id_tag in self._auth_config.blacklist
+ else AuthorizationStatus.accepted
+ )
+ return self._auth_config.default_status
+
+ def _build_id_tag_info(self, id_tag: str) -> dict:
+ """Build id_tag_info dict with optional parentIdTag."""
+ id_tag_info: dict = {"status": self._resolve_auth_status(id_tag)}
+ if self._auth_config.parent_id_tag is not None:
+ id_tag_info["parent_id_tag"] = self._auth_config.parent_id_tag
+ return id_tag_info
+
+ def _get_active_or_fallback_transaction_id(self) -> int:
+ """Return the first active transaction ID, or fall back to a test ID."""
+ transaction_id = next(iter(self._active_transactions), None)
+ if transaction_id is None:
+ logger.warning("No active transaction found, using fallback ID")
+ transaction_id = FALLBACK_TRANSACTION_ID
+ return transaction_id
+
+ # --- Incoming message handlers (CS → CSMS) ---
+
+ @on(Action.boot_notification)
+ async def on_boot_notification(
+ self, charge_point_model, charge_point_vendor, **kwargs
+ ):
+ logger.info(
+ "Received %s from model=%s vendor=%s",
+ Action.boot_notification,
+ charge_point_model,
+ charge_point_vendor,
+ )
+ idx = self._boot_index[0]
+ status = self._boot_sequence[min(idx, len(self._boot_sequence) - 1)]
+ self._boot_index[0] = idx + 1
+ return ocpp.v16.call_result.BootNotification(
+ current_time=datetime.now(timezone.utc).isoformat(),
+ interval=DEFAULT_HEARTBEAT_INTERVAL_SECONDS,
+ status=status,
+ )
+
+ @on(Action.heartbeat)
+ async def on_heartbeat(self, **kwargs):
+ logger.info("Received %s", Action.heartbeat)
+ return ocpp.v16.call_result.Heartbeat(
+ current_time=datetime.now(timezone.utc).isoformat()
+ )
+
+ @on(Action.authorize)
+ async def on_authorize(self, id_tag, **kwargs):
+ logger.info("Received %s for id_tag: %s", Action.authorize, id_tag)
+ if self._auth_config.offline:
+ logger.warning("Offline mode - simulating network failure")
+ raise InternalError(description="Simulated network failure")
+ id_tag_info = self._build_id_tag_info(id_tag)
+ logger.info("Authorization status for %s: %s", id_tag, id_tag_info["status"])
+ return ocpp.v16.call_result.Authorize(id_tag_info=id_tag_info)
+
+ @on(Action.start_transaction)
+ async def on_start_transaction(
+ self, connector_id: int, id_tag: str, meter_start: int, timestamp, **kwargs
+ ):
+ logger.info(
+ "Received %s on connector %s for id_tag %s (meter_start=%s)",
+ Action.start_transaction,
+ connector_id,
+ id_tag,
+ meter_start,
+ )
+ id_tag_info = self._build_id_tag_info(id_tag)
+ transaction_id = _random_transaction_id()
+ self._active_transactions[transaction_id] = connector_id
+ logger.info(
+ "Started transaction %s (auth status %s)",
+ transaction_id,
+ id_tag_info["status"],
+ )
+ return ocpp.v16.call_result.StartTransaction(
+ transaction_id=transaction_id, id_tag_info=id_tag_info
+ )
+
+ @on(Action.stop_transaction)
+ async def on_stop_transaction(
+ self, meter_stop: int, timestamp, transaction_id: int, **kwargs
+ ):
+ logger.info(
+ "Received %s for transaction %s (meter_stop=%s, reason=%s)",
+ Action.stop_transaction,
+ transaction_id,
+ meter_stop,
+ kwargs.get("reason"),
+ )
+ self._active_transactions.pop(transaction_id, None)
+ transaction_data = kwargs.get("transaction_data")
+ if transaction_data is not None:
+ _log_signed_meter_values(transaction_data)
+ id_tag = kwargs.get("id_tag")
+ if id_tag is not None:
+ return ocpp.v16.call_result.StopTransaction(
+ id_tag_info=self._build_id_tag_info(id_tag)
+ )
+ return ocpp.v16.call_result.StopTransaction()
+
+ @on(Action.meter_values)
+ async def on_meter_values(self, connector_id, meter_value, **kwargs):
+ logger.info(
+ "Received %s on connector %s (transaction_id=%s)",
+ Action.meter_values,
+ connector_id,
+ kwargs.get("transaction_id"),
+ )
+ _log_signed_meter_values(meter_value)
+ return ocpp.v16.call_result.MeterValues()
+
+ @on(Action.status_notification)
+ async def on_status_notification(
+ self, connector_id: int, error_code: str, status: str, **kwargs
+ ):
+ logger.info(
+ "Received %s on connector %s: status=%s, error_code=%s",
+ Action.status_notification,
+ connector_id,
+ status,
+ error_code,
+ )
+ return ocpp.v16.call_result.StatusNotification()
+
+ @on(Action.data_transfer)
+ async def on_data_transfer(self, vendor_id, **kwargs):
+ logger.info(
+ "Received %s from vendor %s (message_id=%s)",
+ Action.data_transfer,
+ vendor_id,
+ kwargs.get("message_id"),
+ )
+ return ocpp.v16.call_result.DataTransfer(status=DataTransferStatus.accepted)
+
+ @on(Action.firmware_status_notification)
+ async def on_firmware_status_notification(self, status, **kwargs):
+ logger.info(
+ "Received %s: status=%s", Action.firmware_status_notification, status
+ )
+ return ocpp.v16.call_result.FirmwareStatusNotification()
+
+ @on(Action.diagnostics_status_notification)
+ async def on_diagnostics_status_notification(self, status, **kwargs):
+ logger.info(
+ "Received %s: status=%s", Action.diagnostics_status_notification, status
+ )
+ return ocpp.v16.call_result.DiagnosticsStatusNotification()
+
+ # --- Outgoing commands (CSMS → CS) ---
+
+ async def _call_and_log(self, request, action: Action, success_status) -> None:
+ """Send an OCPP request and log success or failure based on its status."""
+ response = await self.call(request, suppress=False)
+ if response.status == success_status:
+ logger.info("%s successful", action)
+ else:
+ logger.info("%s failed: %s", action, response.status)
+
+ async def _send_trigger_message(self):
+ request = ocpp.v16.call.TriggerMessage(
+ requested_message=self._trigger_message_type,
+ connector_id=self._connector_id,
+ )
+ await self._call_and_log(
+ request, Action.trigger_message, TriggerMessageStatus.accepted
+ )
+
+ async def _send_remote_start_transaction(self):
+ request = ocpp.v16.call.RemoteStartTransaction(
+ id_tag=DEFAULT_TEST_TOKEN, connector_id=self._connector_id
+ )
+ await self._call_and_log(
+ request, Action.remote_start_transaction, RemoteStartStopStatus.accepted
+ )
+
+ async def _send_remote_stop_transaction(self):
+ request = ocpp.v16.call.RemoteStopTransaction(
+ transaction_id=self._get_active_or_fallback_transaction_id()
+ )
+ await self._call_and_log(
+ request, Action.remote_stop_transaction, RemoteStartStopStatus.accepted
+ )
+
+ async def _send_reset(self):
+ request = ocpp.v16.call.Reset(type=self._reset_type)
+ await self._call_and_log(request, Action.reset, ResetStatus.accepted)
+
+ async def _send_change_availability(self):
+ request = ocpp.v16.call.ChangeAvailability(
+ connector_id=self._connector_id, type=self._availability_type
+ )
+ await self._call_and_log(
+ request, Action.change_availability, AvailabilityStatus.accepted
+ )
+
+ async def _send_update_firmware(self):
+ request = ocpp.v16.call.UpdateFirmware(
+ location=DEFAULT_FIRMWARE_URL,
+ retrieve_date=datetime.now(timezone.utc).isoformat(),
+ )
+ await self.call(request, suppress=False)
+ logger.info("%s response received", Action.update_firmware)
+
+ async def _send_get_diagnostics(self):
+ request = ocpp.v16.call.GetDiagnostics(location=DEFAULT_DIAGNOSTICS_URL)
+ response = await self.call(request, suppress=False)
+ logger.info(
+ "%s response received: file_name=%s",
+ Action.get_diagnostics,
+ response.file_name,
+ )
+
+ async def _send_reserve_now(self):
+ expiry_date = (
+ datetime.now(timezone.utc)
+ + timedelta(seconds=DEFAULT_RESERVATION_EXPIRY_SECONDS)
+ ).isoformat()
+ request = ocpp.v16.call.ReserveNow(
+ connector_id=self._reserve_connector_id,
+ expiry_date=expiry_date,
+ id_tag=self._reserve_id_tag,
+ reservation_id=self._reservation_id,
+ )
+ await self._call_and_log(
+ request, Action.reserve_now, ReservationStatus.accepted
+ )
+
+ async def _send_cancel_reservation(self):
+ request = ocpp.v16.call.CancelReservation(reservation_id=self._reservation_id)
+ await self._call_and_log(
+ request, Action.cancel_reservation, CancelReservationStatus.accepted
+ )
+
+ # --- Command dispatch ---
+
+ # Intentional subset of CSMS→CS commands supported by this mock.
+ # Any Action absent from this map is handled by _send_command via
+ # logger.warning only — no request is sent and no error is raised.
+ _COMMAND_HANDLERS: ClassVar[dict[Action, str]] = {
+ Action.trigger_message: "_send_trigger_message",
+ Action.remote_start_transaction: "_send_remote_start_transaction",
+ Action.remote_stop_transaction: "_send_remote_stop_transaction",
+ Action.reset: "_send_reset",
+ Action.change_availability: "_send_change_availability",
+ Action.update_firmware: "_send_update_firmware",
+ Action.get_diagnostics: "_send_get_diagnostics",
+ Action.reserve_now: "_send_reserve_now",
+ Action.cancel_reservation: "_send_cancel_reservation",
+ }
+
+ async def _send_command(self, command_name: Action):
+ logger.debug("Sending OCPP command %s", command_name)
+ try:
+ handler_name = self._COMMAND_HANDLERS.get(command_name)
+ if handler_name is not None:
+ await getattr(self, handler_name)()
+ else:
+ logger.warning("Not supported command %s", command_name)
+ except TimeoutError:
+ logger.error("Timeout waiting for %s response", command_name)
+ except OCPPError as e:
+ logger.error(
+ "OCPP error sending %s: [%s] %s",
+ command_name,
+ type(e).__name__,
+ e.description,
+ )
+ except ConnectionClosed:
+ logger.warning("Connection closed while sending %s", command_name)
+ self.handle_connection_closed()
+ except Exception:
+ logger.exception("Unexpected error sending %s", command_name)
+
+ async def send_command(
+ self, command_name: Action, delay: float | None, period: float | None
+ ):
+ try:
+ if delay and not self._command_timer:
+ self._command_timer = Timer(
+ delay,
+ False,
+ self._send_command,
+ (command_name,),
+ )
+ if period and not self._command_timer:
+ self._command_timer = Timer(
+ period,
+ True,
+ self._send_command,
+ (command_name,),
+ )
+ except ConnectionClosed:
+ self.handle_connection_closed()
+
+ async def send_commands(self, commands: list[tuple[Action, float]]) -> None:
+ for command_name, delay in commands:
+ await asyncio.sleep(delay)
+ await self._send_command(command_name)
+
+ def handle_connection_closed(self) -> None:
+ logger.info("ChargePoint %s closed connection", self.id)
+ if self._command_timer:
+ self._command_timer.cancel()
+ if self._commands_task:
+ self._commands_task.cancel()
+ self._charge_points.discard(self)
+ logger.debug("Connected ChargePoint(s): %d", len(self._charge_points))
+
+
+async def on_connect(
+ websocket,
+ config: ServerConfig,
+):
+ """Handle new WebSocket connections from charge points."""
+ if not await negotiate_subprotocol(websocket, logger):
+ return
+
+ charge_points: set[ChargePoint] = config.charge_points
+ cp = ChargePoint(
+ websocket,
+ auth_config=config.auth_config,
+ boot_sequence=config.boot_sequence,
+ boot_index=config.boot_index,
+ connector_id=config.connector_id,
+ trigger_message_type=config.trigger_message_type,
+ reset_type=config.reset_type,
+ availability_type=config.availability_type,
+ reserve_connector_id=config.reserve_connector_id,
+ reserve_id_tag=config.reserve_id_tag,
+ reservation_id=config.reservation_id,
+ charge_points=charge_points,
+ )
+ if config.command_name:
+ await cp.send_command(config.command_name, config.delay, config.period)
+ elif config.commands:
+ # send_commands() begins with asyncio.sleep(delay) which yields to
+ # cp.start() below. All delays are validated > 0 by _parse_commands.
+ cp._commands_task = asyncio.create_task(cp.send_commands(config.commands))
+
+ try:
+ await cp.start()
+ except ConnectionClosed:
+ cp.handle_connection_closed()
+
+
+def _parse_commands(commands_str: str) -> list[tuple[Action, float]]:
+ return parse_commands(commands_str, Action)
+
+
+async def main():
+ parser = argparse.ArgumentParser(description="OCPP 1.6 Server")
+ command_group = parser.add_mutually_exclusive_group()
+ command_group.add_argument("-c", "--command", type=Action, help="command name")
+ command_group.add_argument(
+ "--commands",
+ type=str,
+ default=None,
+ help=(
+ 'comma-separated command sequence: "CMD1:DELAY1,CMD2:DELAY2,..."'
+ ' (e.g., "RemoteStartTransaction:5,RemoteStopTransaction:30")'
+ ),
+ )
+ group = parser.add_mutually_exclusive_group()
+ group.add_argument(
+ "-d",
+ "--delay",
+ type=check_positive_number,
+ help="delay in seconds",
+ )
+ group.add_argument(
+ "-p",
+ "--period",
+ type=check_positive_number,
+ help="period in seconds",
+ )
+
+ # Server configuration
+ parser.add_argument(
+ "--host",
+ type=str,
+ default=DEFAULT_HOST,
+ help=f"server host (default: {DEFAULT_HOST})",
+ )
+ parser.add_argument(
+ "--port",
+ type=int,
+ default=DEFAULT_PORT,
+ help=f"server port (default: {DEFAULT_PORT})",
+ )
+ parser.add_argument(
+ "--connector-id",
+ type=int,
+ default=DEFAULT_CONNECTOR_ID,
+ help=(
+ f"connector id used in CSMS→CS commands (default: {DEFAULT_CONNECTOR_ID})"
+ ),
+ )
+
+ # Boot behavior
+ boot_group = parser.add_mutually_exclusive_group()
+ boot_group.add_argument(
+ "--boot-status",
+ type=RegistrationStatus,
+ default=None,
+ help=(
+ "boot notification response status"
+ " (Accepted, Pending, Rejected; default: Accepted)"
+ ),
+ )
+ boot_group.add_argument(
+ "--boot-status-sequence",
+ type=str,
+ default=None,
+ help=(
+ "comma-separated boot notification status sequence"
+ " (e.g. Pending,Pending,Accepted)"
+ ),
+ )
+
+ # Command-specific options
+ parser.add_argument(
+ "--trigger-message",
+ type=MessageTrigger,
+ default=MessageTrigger.status_notification,
+ help="TriggerMessage requested_message type (default: StatusNotification)",
+ )
+ parser.add_argument(
+ "--reset-type",
+ type=ResetType,
+ default=ResetType.hard,
+ help="Reset type: Hard, Soft (default: Hard)",
+ )
+ parser.add_argument(
+ "--availability-type",
+ type=AvailabilityType,
+ default=AvailabilityType.operative,
+ help="ChangeAvailability type: Operative, Inoperative (default: Operative)",
+ )
+ parser.add_argument(
+ "--reserve-connector-id",
+ type=int,
+ default=DEFAULT_RESERVE_CONNECTOR_ID,
+ help=(f"ReserveNow connector id (default: {DEFAULT_RESERVE_CONNECTOR_ID})"),
+ )
+ parser.add_argument(
+ "--reserve-id-tag",
+ type=str,
+ default=DEFAULT_RESERVE_ID_TAG,
+ help=(
+ "ReserveNow id tag reserving the connector"
+ f" (default: {DEFAULT_RESERVE_ID_TAG})"
+ ),
+ )
+ parser.add_argument(
+ "--reservation-id",
+ type=int,
+ default=DEFAULT_RESERVATION_ID,
+ help=(
+ "ReserveNow/CancelReservation reservation id"
+ f" (default: {DEFAULT_RESERVATION_ID})"
+ ),
+ )
+
+ # Auth configuration
+ parser.add_argument(
+ "--auth-mode",
+ type=AuthMode,
+ default=AuthMode.normal,
+ help="Authorization mode: normal, whitelist, blacklist (default: normal)",
+ )
+ parser.add_argument(
+ "--whitelist",
+ type=str,
+ nargs="+",
+ default=list(DEFAULT_WHITELIST),
+ help="Whitelist of authorized id tags (space-separated)",
+ )
+ parser.add_argument(
+ "--blacklist",
+ type=str,
+ nargs="+",
+ default=list(DEFAULT_BLACKLIST),
+ help="Blacklist of blocked id tags (space-separated)",
+ )
+ parser.add_argument(
+ "--offline",
+ action="store_true",
+ help="Simulate offline/network failure mode",
+ )
+ parser.add_argument(
+ "--auth-parent-id-tag",
+ type=str,
+ default=None,
+ help="parentIdTag value to include in Authorize/StartTransaction responses",
+ )
+
+ args, _ = parser.parse_known_args()
+ group.required = args.command is not None
+
+ args = parser.parse_args()
+
+ try:
+ parsed_commands = _parse_commands(args.commands) if args.commands else None
+ if parsed_commands is not None and not parsed_commands:
+ parser.error("--commands must contain at least one CMD:DELAY entry")
+ except argparse.ArgumentTypeError as e:
+ parser.error(str(e))
+
+ if args.boot_status_sequence is not None:
+ boot_sequence_items: list[RegistrationStatus] = []
+ for raw_value in args.boot_status_sequence.split(","):
+ value = raw_value.strip()
+ try:
+ status = RegistrationStatus(value)
+ except ValueError:
+ valid = ", ".join(e.value for e in RegistrationStatus)
+ parser.error(
+ f"invalid value for --boot-status-sequence: {value!r}."
+ f" Valid values are: {valid}"
+ )
+ boot_sequence_items.append(status)
+ boot_sequence = tuple(boot_sequence_items)
+ elif args.boot_status is not None:
+ boot_sequence = (args.boot_status,)
+ else:
+ boot_sequence = (RegistrationStatus.accepted,)
+
+ auth_config = AuthConfig(
+ mode=args.auth_mode,
+ whitelist=tuple(args.whitelist),
+ blacklist=tuple(args.blacklist),
+ offline=args.offline,
+ default_status=AuthorizationStatus.accepted,
+ parent_id_tag=args.auth_parent_id_tag,
+ )
+
+ config = ServerConfig(
+ command_name=args.command,
+ delay=args.delay,
+ period=args.period,
+ auth_config=auth_config,
+ boot_sequence=boot_sequence,
+ connector_id=args.connector_id,
+ boot_index=[0],
+ charge_points=set(),
+ commands=parsed_commands,
+ trigger_message_type=args.trigger_message,
+ reset_type=args.reset_type,
+ availability_type=args.availability_type,
+ reserve_connector_id=args.reserve_connector_id,
+ reserve_id_tag=args.reserve_id_tag,
+ reservation_id=args.reservation_id,
+ )
+
+ logger.info(
+ "Auth configuration: mode=%s, offline=%s",
+ auth_config.mode,
+ auth_config.offline,
+ )
+
+ loop = asyncio.get_running_loop()
+ shutdown_event = asyncio.Event()
+
+ async with websockets.serve(
+ partial(
+ on_connect,
+ config=config,
+ ),
+ args.host,
+ args.port,
+ subprotocols=SUBPROTOCOLS,
+ ) as server:
+ logger.info("WebSocket Server Started on %s:%d", args.host, args.port)
+ await install_signal_handlers_and_wait(
+ loop, server, shutdown_event, SHUTDOWN_TIMEOUT_SECONDS, logger
+ )
+
+ logger.info("Server shutdown complete")
+
+
+if __name__ == "__main__":
+ logging.basicConfig(level=logging.DEBUG)
+ with contextlib.suppress(KeyboardInterrupt):
+ asyncio.run(main())
+ sys.exit(0)
import contextlib
import logging
import signal
+from datetime import datetime, timezone
from typing import Any, ClassVar
from unittest.mock import AsyncMock, MagicMock, patch
from ocpp.v201.enums import (
Action,
AuthorizationStatusEnumType,
+ CancelReservationStatusEnumType,
CertificateSignedStatusEnumType,
ChangeAvailabilityStatusEnumType,
ClearCacheStatusEnumType,
RegistrationStatusEnumType,
ReportBaseEnumType,
RequestStartStopStatusEnumType,
+ ReserveNowStatusEnumType,
ResetEnumType,
ResetStatusEnumType,
SendLocalListStatusEnumType,
)
from server import (
+ DEFAULT_EVSE_ID,
DEFAULT_HEARTBEAT_INTERVAL_SECONDS,
DEFAULT_LOCAL_LIST_VERSION,
+ DEFAULT_RESERVATION_ID,
+ DEFAULT_RESERVE_ID_TOKEN,
DEFAULT_TOTAL_COST,
FALLBACK_TRANSACTION_ID,
MAX_REQUEST_ID,
AuthMode,
ChargePoint,
ServerConfig,
+ _log_signed_meter_values,
_parse_commands,
_parse_variable_specs,
_random_request_id,
@contextlib.contextmanager
-def _patch_main(mock_loop, mock_server, mock_event, extra_patches=None):
+def _patch_main(
+ mock_loop, mock_server, mock_event, extra_patches=None, **args_overrides
+):
args = argparse.Namespace(
command=None,
commands=None,
set_variables=None,
get_variables=None,
local_list_tokens=None,
+ reservation_id=DEFAULT_RESERVATION_ID,
+ reserve_id_token=DEFAULT_RESERVE_ID_TOKEN,
+ reserve_evse_id=DEFAULT_EVSE_ID,
)
+ for key, value in args_overrides.items():
+ setattr(args, key, value)
mock_serve_cm = AsyncMock()
mock_serve_cm.__aenter__ = AsyncMock(return_value=mock_server)
mock_serve_cm.__aexit__ = AsyncMock(return_value=False)
"_send_send_local_list",
"_send_set_network_profile",
"_send_update_firmware",
+ "_send_reserve_now",
+ "_send_cancel_reservation",
]
@pytest.mark.parametrize("handler_name", EXPECTED_INCOMING_HANDLERS)
f"Missing incoming handler: {handler_name}"
)
assert callable(getattr(ChargePoint, handler_name))
+ on_action = getattr(getattr(ChargePoint, handler_name), "_on_action", None)
+ assert on_action is not None, (
+ f"Handler {handler_name} is not registered with an @on decorator"
+ )
+ expected_action = Action[handler_name.removeprefix("on_")]
+ assert on_action == expected_action, (
+ f"Handler {handler_name} routes {on_action}, expected {expected_action}"
+ )
@pytest.mark.parametrize("method_name", EXPECTED_OUTGOING_COMMANDS)
def test_outgoing_command_exists(self, method_name):
assert response.status == RegistrationStatusEnumType.accepted
assert response.interval == DEFAULT_HEARTBEAT_INTERVAL_SECONDS
assert isinstance(response.current_time, str)
- assert "T" in response.current_time
+ assert datetime.fromisoformat(response.current_time).tzinfo is not None
async def test_configurable_boot_status(self, mock_connection):
cp = ChargePoint(
async def test_returns_current_time(self, charge_point):
response = await charge_point.on_heartbeat()
assert isinstance(response.current_time, str)
- assert "T" in response.current_time
+ # Real contract: current_time must be a parseable ISO-8601 timestamp (UTC),
+ # not merely a string containing "T". datetime.fromisoformat raises on garbage.
+ parsed = datetime.fromisoformat(response.current_time)
+ assert parsed.tzinfo is not None
+ # emitted "now" — within a generous window of the assertion time
+ delta = abs((datetime.now(timezone.utc) - parsed).total_seconds())
+ assert delta < 60
class TestStatusNotificationHandler:
connector_status="Available",
)
assert isinstance(response, ocpp.v201.call_result.StatusNotification)
+ assert response == ocpp.v201.call_result.StatusNotification()
class TestAuthorizeHandler:
)
assert "cache_expiry_date_time" in result.id_token_info
expiry = result.id_token_info["cache_expiry_date_time"]
- assert isinstance(expiry, str) and "T" in expiry
+ assert isinstance(expiry, str)
+ assert datetime.fromisoformat(expiry).tzinfo is not None
async def test_authorize_no_enrichment_by_default(self, charge_point):
result = await charge_point.on_authorize(
assert response.total_cost is None
assert response.id_token_info is None
- async def test_started_with_signed_meter_value(self, charge_point):
+ async def test_started_with_signed_meter_value(self, charge_point, caplog):
+ caplog.set_level(logging.INFO)
signed_mv = {
"signed_meter_data": "T0NNRnx7fXxmYWtlc2lnbmF0dXJl",
"signing_method": "ECDSA-secp256r1-SHA256",
)
assert isinstance(response, ocpp.v201.call_result.TransactionEvent)
assert response.id_token_info["status"] == AuthorizationStatusEnumType.accepted
+ assert any("signed meter value" in r.message.lower() for r in caplog.records)
- async def test_updated_with_signed_meter_value(self, charge_point):
+ async def test_updated_with_signed_meter_value(self, charge_point, caplog):
+ caplog.set_level(logging.INFO)
signed_mv = {
"signed_meter_data": "T0NNRnx7fXxmYWtlc2lnbmF0dXJl",
"signing_method": "ECDSA-secp256r1-SHA256",
)
assert isinstance(response, ocpp.v201.call_result.TransactionEvent)
assert response.total_cost == DEFAULT_TOTAL_COST
+ assert any("signed meter value" in r.message.lower() for r in caplog.records)
- async def test_ended_with_signed_meter_value(self, charge_point):
+ async def test_ended_with_signed_meter_value(self, charge_point, caplog):
+ caplog.set_level(logging.INFO)
signed_mv = {
"signed_meter_data": "T0NNRnx7fXxmYWtlc2lnbmF0dXJl",
"signing_method": "ECDSA-secp256r1-SHA256",
meter_value=[begin_mv, end_mv],
)
assert isinstance(response, ocpp.v201.call_result.TransactionEvent)
+ assert any("signed meter value" in r.message.lower() for r in caplog.records)
class TestDataTransferHandler:
)
assert "" not in charge_point._active_transactions
+ async def test_transaction_event_started_null_evse_defaults_to_zero(
+ self, charge_point
+ ):
+ await charge_point.on_transaction_event(
+ event_type=TransactionEventEnumType.started,
+ timestamp=TEST_TIMESTAMP,
+ trigger_reason="Authorized",
+ seq_no=0,
+ transaction_info={"transaction_id": TEST_TRANSACTION_ID},
+ id_token={"id_token": TEST_TOKEN, "type": "ISO14443"},
+ evse=None,
+ )
+ assert charge_point._active_transactions[TEST_TRANSACTION_ID] == 0
+
class TestCertificateHandlers:
"""Tests for certificate-related incoming handlers."""
meter_value=[{"timestamp": TEST_TIMESTAMP}],
)
assert isinstance(response, ocpp.v201.call_result.MeterValues)
+ assert response == ocpp.v201.call_result.MeterValues()
- async def test_meter_values_with_signed_meter_value(self, charge_point):
+ async def test_meter_values_with_signed_meter_value(self, charge_point, caplog):
+ caplog.set_level(logging.INFO)
signed_mv = {
"signed_meter_data": "T0NNRnx7fXxmYWtlc2lnbmF0dXJl",
"signing_method": "ECDSA-secp256r1-SHA256",
evse_id=TEST_EVSE_ID, meter_value=[mv]
)
assert isinstance(response, ocpp.v201.call_result.MeterValues)
+ assert any("signed meter value" in r.message.lower() for r in caplog.records)
async def test_notify_report(self, charge_point):
response = await charge_point.on_notify_report(
seq_no=0,
)
assert isinstance(response, ocpp.v201.call_result.NotifyReport)
+ assert response == ocpp.v201.call_result.NotifyReport()
async def test_firmware_status_notification(self, charge_point):
response = await charge_point.on_firmware_status_notification(
status="Installed"
)
assert isinstance(response, ocpp.v201.call_result.FirmwareStatusNotification)
+ assert response == ocpp.v201.call_result.FirmwareStatusNotification()
async def test_log_status_notification(self, charge_point):
response = await charge_point.on_log_status_notification(
status="Uploaded", request_id=1
)
assert isinstance(response, ocpp.v201.call_result.LogStatusNotification)
+ assert response == ocpp.v201.call_result.LogStatusNotification()
async def test_security_event_notification(self, charge_point):
response = await charge_point.on_security_event_notification(
event_type="FirmwareUpdated", timestamp=TEST_TIMESTAMP
)
assert isinstance(response, ocpp.v201.call_result.SecurityEventNotification)
+ assert response == ocpp.v201.call_result.SecurityEventNotification()
async def test_notify_customer_information(self, charge_point):
response = await charge_point.on_notify_customer_information(
request_id=1,
)
assert isinstance(response, ocpp.v201.call_result.NotifyCustomerInformation)
+ assert response == ocpp.v201.call_result.NotifyCustomerInformation()
+
+
+class TestLogSignedMeterValues:
+ """Tests for the _log_signed_meter_values helper (OCPP 2.0.1 semantics)."""
+
+ def test_logs_signed_data(self, caplog):
+ caplog.set_level(logging.INFO)
+ signed_mv = {
+ "signed_meter_data": "T0NNRnx7fXxmYWtlc2lnbmF0dXJl",
+ "signing_method": "ECDSA-secp256r1-SHA256",
+ "encoding_method": "OCMF",
+ "public_key": "b2NhOmJhc2UxNjphc24xOmZha2VrZXk=",
+ }
+ meter_value = [
+ {
+ "timestamp": TEST_TIMESTAMP,
+ "sampled_value": [
+ {
+ "value": 0.0,
+ "measurand": "Energy.Active.Import.Register",
+ "signed_meter_value": signed_mv,
+ }
+ ],
+ }
+ ]
+ _log_signed_meter_values(meter_value)
+ assert any("signed meter value" in r.message.lower() for r in caplog.records)
+
+ def test_ignores_unsigned_data(self, caplog):
+ caplog.set_level(logging.INFO)
+ meter_value = [
+ {
+ "timestamp": TEST_TIMESTAMP,
+ "sampled_value": [
+ {
+ "value": 10.5,
+ "measurand": "Energy.Active.Import.Register",
+ }
+ ],
+ }
+ ]
+ _log_signed_meter_values(meter_value)
+ assert not any(
+ "signed meter value" in r.message.lower() for r in caplog.records
+ )
+
+ def test_empty_list_no_error(self, caplog):
+ caplog.set_level(logging.INFO)
+ _log_signed_meter_values([])
+ assert not any(
+ "signed meter value" in r.message.lower() for r in caplog.records
+ )
class TestSendCommandErrorHandling:
"""Tests for error handling in the command dispatch layer."""
- async def test_timeout_is_caught(self, charge_point):
+ async def test_timeout_is_caught(self, charge_point, caplog):
+ caplog.set_level(logging.ERROR)
with patch.object(
charge_point, "_send_clear_cache", side_effect=TimeoutError("timed out")
):
await charge_point._send_command(command_name=Action.clear_cache)
+ assert any(
+ r.levelno == logging.ERROR and "timeout waiting for" in r.message.lower()
+ for r in caplog.records
+ )
- async def test_ocpp_error_is_caught(self, charge_point):
+ async def test_ocpp_error_is_caught(self, charge_point, caplog):
from ocpp.exceptions import InternalError as OCPPInternalError
+ caplog.set_level(logging.ERROR)
with patch.object(
charge_point,
"_send_clear_cache",
side_effect=OCPPInternalError(description="test error"),
):
await charge_point._send_command(command_name=Action.clear_cache)
+ assert any(
+ r.levelno == logging.ERROR
+ and "ocpp error sending" in r.message.lower()
+ and "test error" in r.message
+ for r in caplog.records
+ )
async def test_connection_closed_is_caught(self, charge_point):
from websockets.exceptions import ConnectionClosedOK
await charge_point._send_command(command_name=Action.clear_cache)
charge_point.handle_connection_closed.assert_called_once()
- async def test_unsupported_command_logs_warning(self, charge_point):
+ async def test_unsupported_command_logs_warning(self, charge_point, caplog):
+ caplog.set_level(logging.WARNING)
unsupported = MagicMock(value="Unsupported")
await charge_point._send_command(command_name=unsupported)
+ assert any("not supported" in r.message.lower() for r in caplog.records)
+
+ async def test_unexpected_error_is_caught(self, charge_point, caplog):
+ caplog.set_level(logging.ERROR)
+ with patch.object(
+ charge_point, "_send_clear_cache", side_effect=RuntimeError("boom")
+ ):
+ await charge_point._send_command(command_name=Action.clear_cache)
+ assert any(
+ r.levelno == logging.ERROR
+ and "unexpected error sending" in r.message.lower()
+ for r in caplog.records
+ )
class TestOutgoingCommands:
- """Behavioral tests for all 20 _send_* outgoing command methods."""
+ """Behavioral tests for all 24 _send_* outgoing command methods."""
# --- Success path tests ---
assert request.request_id > 0
assert isinstance(request.firmware, dict)
- # --- Failure path tests (rejected/failed status → correct logging) ---
+ async def test_send_reserve_now(self, command_charge_point):
+ command_charge_point.call.return_value = ocpp.v201.call_result.ReserveNow(
+ status=ReserveNowStatusEnumType.accepted
+ )
+ await command_charge_point._send_reserve_now()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v201.call.ReserveNow)
+ assert request.id == command_charge_point._reservation_id
+ assert request.evse_id == command_charge_point._reserve_evse_id
+ assert request.id_token["id_token"] == command_charge_point._reserve_id_token
+ assert isinstance(request.expiry_date_time, str)
+
+ async def test_send_cancel_reservation(self, command_charge_point):
+ command_charge_point.call.return_value = (
+ ocpp.v201.call_result.CancelReservation(
+ status=CancelReservationStatusEnumType.accepted
+ )
+ )
+ await command_charge_point._send_cancel_reservation()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v201.call.CancelReservation)
+ assert request.reservation_id == command_charge_point._reservation_id
FAILURE_PATH_CASES: ClassVar[list[tuple[str, type, object]]] = [
(
ocpp.v201.call_result.UpdateFirmware,
UpdateFirmwareStatusEnumType.rejected,
),
+ (
+ "_send_reserve_now",
+ ocpp.v201.call_result.ReserveNow,
+ ReserveNowStatusEnumType.rejected,
+ ),
+ (
+ "_send_cancel_reservation",
+ ocpp.v201.call_result.CancelReservation,
+ CancelReservationStatusEnumType.rejected,
+ ),
]
@pytest.mark.parametrize(
await on_connect(mock_ws, config=config)
mock_ws.close.assert_called_once()
+ assert len(config.charge_points) == 0
async def test_protocol_mismatch_closes_connection(self):
mock_ws = MagicMock()
await on_connect(mock_ws, config=config)
mock_ws.close.assert_called_once()
+ assert len(config.charge_points) == 0
async def test_successful_connection_creates_charge_point(self, mock_valid_ws):
config = self._make_config()
await on_connect(mock_valid_ws, config=config)
mock_cp.send_command.assert_called_once_with(Action.clear_cache, 1.0, None)
+ async def test_commands_sequence_scheduled_on_connect(self, mock_valid_ws):
+ config = self._make_config(commands=[(Action.clear_cache, 1.0)])
+
+ with (
+ patch("server.ChargePoint") as MockCP,
+ patch("server.asyncio.create_task") as mock_create_task,
+ ):
+ mock_cp = AsyncMock()
+ mock_cp.send_commands = MagicMock()
+ MockCP.return_value = mock_cp
+ await on_connect(mock_valid_ws, config=config)
+ mock_create_task.assert_called_once()
+
class TestHandleConnectionClosed:
"""Tests for the handle_connection_closed cleanup method."""
mock_loop,
mock_server,
mock_event,
- extra_patches=[patch("server.signal.signal", mock_signal_fn)],
+ extra_patches=[patch("_common.signal.signal", mock_signal_fn)],
):
await main()
mock_loop,
mock_server,
mock_event,
- extra_patches=[patch("server.signal.signal", side_effect=_capture_signal)],
+ extra_patches=[patch("_common.signal.signal", side_effect=_capture_signal)],
):
await main()
sigint_handler(signal.SIGINT.value, None)
mock_loop.call_soon_threadsafe.assert_called_once()
+ async def test_boot_status_sequence_parsed(self, main_mocks):
+ mock_loop, mock_server, mock_event, signal_handlers = main_mocks
+
+ async def _fire_sigint():
+ handler, args = signal_handlers[signal.SIGINT]
+ handler(*args)
+
+ mock_event.wait = AsyncMock(side_effect=_fire_sigint)
+
+ captured: dict[str, ServerConfig] = {}
+ real_server_config = ServerConfig
+
+ def _capture_server_config(*args, **kwargs):
+ config = real_server_config(*args, **kwargs)
+ captured["config"] = config
+ return config
+
+ with _patch_main(
+ mock_loop,
+ mock_server,
+ mock_event,
+ extra_patches=[
+ patch("server.ServerConfig", side_effect=_capture_server_config)
+ ],
+ boot_status_sequence="Pending,Accepted",
+ ):
+ await main()
+
+ mock_server.close.assert_called_once()
+ assert captured["config"].boot_sequence == (
+ RegistrationStatusEnumType.pending,
+ RegistrationStatusEnumType.accepted,
+ )
+
+ async def test_boot_status_single_value(self, main_mocks):
+ mock_loop, mock_server, mock_event, signal_handlers = main_mocks
+
+ async def _fire_sigint():
+ handler, args = signal_handlers[signal.SIGINT]
+ handler(*args)
+
+ mock_event.wait = AsyncMock(side_effect=_fire_sigint)
+
+ captured: dict[str, ServerConfig] = {}
+ real_server_config = ServerConfig
+
+ def _capture_server_config(*args, **kwargs):
+ config = real_server_config(*args, **kwargs)
+ captured["config"] = config
+ return config
+
+ with _patch_main(
+ mock_loop,
+ mock_server,
+ mock_event,
+ extra_patches=[
+ patch("server.ServerConfig", side_effect=_capture_server_config)
+ ],
+ boot_status=RegistrationStatusEnumType.rejected,
+ ):
+ await main()
+
+ mock_server.close.assert_called_once()
+ assert captured["config"].boot_sequence == (
+ RegistrationStatusEnumType.rejected,
+ )
+
class TestTriggerMessageType:
"""Tests for configurable TriggerMessage type."""
with pytest.raises(argparse.ArgumentTypeError, match="finite positive"):
_parse_commands("Reset:nan")
+ def test_parse_commands_non_numeric_delay(self):
+ with pytest.raises(argparse.ArgumentTypeError, match="must be a number"):
+ _parse_commands("Reset:abc")
+
+ def test_parse_commands_skips_empty_entries(self):
+ result = _parse_commands("Reset:5,,ClearCache:10")
+ assert result == [(Action.reset, 5.0), (Action.clear_cache, 10.0)]
+
class TestMultiVariableCommands:
"""Tests for multi-variable SetVariables/GetVariables CLI support."""
):
_parse_variable_specs("NoComponentVariable=30", require_value=True)
+ def test_parse_set_variable_specs_skips_empty_entries(self):
+ result = _parse_variable_specs(
+ "OCPPCommCtrlr.HeartbeatInterval=30,,TxCtrlr.EVConnectionTimeOut=60",
+ require_value=True,
+ )
+ assert len(result) == 2
+ assert result[0]["variable"]["name"] == "HeartbeatInterval"
+ assert result[1]["variable"]["name"] == "EVConnectionTimeOut"
+
+ def test_parse_get_variable_specs_invalid_no_dot(self):
+ with pytest.raises(
+ argparse.ArgumentTypeError,
+ match=r"expected 'Component\.Variable'",
+ ):
+ _parse_variable_specs("NoDot", require_value=False)
+
async def test_send_set_variables_uses_custom_data(self, command_charge_point):
custom_data = [
{
--- /dev/null
+"""Tests for the OCPP 1.6 mock server."""
+
+import argparse
+import contextlib
+import logging
+import signal
+from datetime import datetime, timezone
+from typing import Any, ClassVar
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import ocpp.v16.call
+import ocpp.v16.call_result
+import pytest
+from ocpp.v16.enums import (
+ Action,
+ AuthorizationStatus,
+ AvailabilityStatus,
+ AvailabilityType,
+ CancelReservationStatus,
+ DataTransferStatus,
+ MessageTrigger,
+ RegistrationStatus,
+ RemoteStartStopStatus,
+ ReservationStatus,
+ ResetStatus,
+ ResetType,
+ TriggerMessageStatus,
+ ValueFormat,
+)
+
+from server16 import (
+ DEFAULT_DIAGNOSTICS_URL,
+ DEFAULT_FIRMWARE_URL,
+ DEFAULT_HEARTBEAT_INTERVAL_SECONDS,
+ DEFAULT_RESERVE_ID_TAG,
+ DEFAULT_TEST_TOKEN,
+ FALLBACK_TRANSACTION_ID,
+ MAX_TRANSACTION_ID,
+ AuthConfig,
+ AuthMode,
+ ChargePoint,
+ ServerConfig,
+ _log_signed_meter_values,
+ _parse_commands,
+ _random_transaction_id,
+ check_positive_number,
+ main,
+ on_connect,
+)
+
+# --- Test constants ---
+TEST_CHARGE_POINT_PATH = "/TestChargePoint"
+TEST_VALID_TOKEN = "valid_token" # noqa: S105
+TEST_TOKEN = "test_token" # noqa: S105
+TEST_BLOCKED_TOKEN = "blocked_token" # noqa: S105
+TEST_TIMESTAMP = "2026-01-01T00:00:00Z"
+TEST_TRANSACTION_ID = 12345
+TEST_CONNECTOR_ID = 1
+TEST_VENDOR_ID = "TestVendor"
+TEST_MODEL = "Test"
+TEST_METER_START = 0
+TEST_METER_STOP = 15000
+
+
+@pytest.fixture
+def mock_connection():
+ """Create a mock WebSocket connection for ChargePoint instantiation."""
+ conn = MagicMock()
+ conn.request = MagicMock()
+ conn.request.path = TEST_CHARGE_POINT_PATH
+ return conn
+
+
+@pytest.fixture
+def charge_point(mock_connection):
+ """Create a ChargePoint instance with default auth config."""
+ return ChargePoint(mock_connection)
+
+
+@pytest.fixture
+def whitelist_charge_point(mock_connection):
+ """Create a ChargePoint with whitelist auth mode."""
+ return ChargePoint(
+ mock_connection,
+ auth_config=AuthConfig(
+ mode=AuthMode.whitelist,
+ whitelist=(TEST_VALID_TOKEN, TEST_TOKEN),
+ blacklist=(),
+ offline=False,
+ default_status=AuthorizationStatus.accepted,
+ ),
+ )
+
+
+@pytest.fixture
+def blacklist_charge_point(mock_connection):
+ """Create a ChargePoint with blacklist auth mode."""
+ return ChargePoint(
+ mock_connection,
+ auth_config=AuthConfig(
+ mode=AuthMode.blacklist,
+ whitelist=(),
+ blacklist=(TEST_BLOCKED_TOKEN,),
+ offline=False,
+ default_status=AuthorizationStatus.accepted,
+ ),
+ )
+
+
+@pytest.fixture
+def offline_charge_point(mock_connection):
+ """Create a ChargePoint with offline mode enabled."""
+ return ChargePoint(
+ mock_connection,
+ auth_config=AuthConfig(
+ mode=AuthMode.normal,
+ whitelist=(),
+ blacklist=(),
+ offline=True,
+ default_status=AuthorizationStatus.accepted,
+ ),
+ )
+
+
+@pytest.fixture
+def command_charge_point(mock_connection):
+ """Create a ChargePoint with self.call mocked as AsyncMock."""
+ cp = ChargePoint(mock_connection)
+ cp.call = AsyncMock()
+ return cp
+
+
+@pytest.fixture
+def mock_valid_ws():
+ """Create a mock WebSocket with valid OCPP 1.6 subprotocol."""
+ ws = MagicMock()
+ ws.request = MagicMock()
+ ws.request.headers = {"Sec-WebSocket-Protocol": "ocpp1.6"}
+ ws.subprotocol = "ocpp1.6"
+ ws.request.path = TEST_CHARGE_POINT_PATH
+ ws.close = AsyncMock()
+ return ws
+
+
+@pytest.fixture
+def main_mocks():
+ """Provide mock loop, server, shutdown event, and signal capture."""
+ mock_loop = MagicMock()
+ signal_handlers: dict[int, tuple] = {}
+
+ def _capture_handler(sig, callback, *args):
+ signal_handlers[sig] = (callback, args)
+
+ mock_loop.add_signal_handler = MagicMock(side_effect=_capture_handler)
+
+ mock_server = AsyncMock()
+ mock_server.close = MagicMock()
+ mock_server.wait_closed = AsyncMock()
+
+ mock_event = MagicMock()
+ mock_event.set = MagicMock()
+
+ return mock_loop, mock_server, mock_event, signal_handlers
+
+
+@contextlib.contextmanager
+def _patch_main(
+ mock_loop, mock_server, mock_event, extra_patches=None, **args_overrides
+):
+ args = argparse.Namespace(
+ command=None,
+ commands=None,
+ delay=None,
+ period=None,
+ host="127.0.0.1",
+ port=9000,
+ connector_id=1,
+ boot_status=None,
+ boot_status_sequence=None,
+ trigger_message=MessageTrigger.status_notification,
+ reset_type=ResetType.hard,
+ availability_type=AvailabilityType.operative,
+ reserve_connector_id=1,
+ reserve_id_tag=DEFAULT_RESERVE_ID_TAG,
+ reservation_id=1,
+ auth_mode="normal",
+ whitelist=["valid_token"],
+ blacklist=["blocked_token"],
+ offline=False,
+ auth_parent_id_tag=None,
+ )
+ for key, value in args_overrides.items():
+ setattr(args, key, value)
+ mock_serve_cm = AsyncMock()
+ mock_serve_cm.__aenter__ = AsyncMock(return_value=mock_server)
+ mock_serve_cm.__aexit__ = AsyncMock(return_value=False)
+
+ patches = [
+ patch(
+ "server16.argparse.ArgumentParser.parse_known_args",
+ return_value=(MagicMock(command=args.command), []),
+ ),
+ patch("server16.argparse.ArgumentParser.parse_args", return_value=args),
+ patch("server16.websockets.serve", return_value=mock_serve_cm),
+ patch("server16.asyncio.get_running_loop", return_value=mock_loop),
+ patch("server16.asyncio.Event", return_value=mock_event),
+ *(extra_patches or []),
+ ]
+ with contextlib.ExitStack() as stack:
+ for p in patches:
+ stack.enter_context(p)
+ yield
+
+
+class TestCheckPositiveNumber:
+ """Tests for the check_positive_number argument validator."""
+
+ def test_positive_integer(self):
+ assert check_positive_number("5") == 5.0
+
+ def test_positive_float(self):
+ assert check_positive_number("3.14") == 3.14
+
+ def test_zero_raises(self):
+ with pytest.raises(
+ argparse.ArgumentTypeError, match="must be a positive number"
+ ):
+ check_positive_number("0")
+
+ def test_negative_raises(self):
+ with pytest.raises(
+ argparse.ArgumentTypeError, match="must be a positive number"
+ ):
+ check_positive_number("-1")
+
+ def test_non_numeric_raises(self):
+ with pytest.raises(argparse.ArgumentTypeError, match="must be a number"):
+ check_positive_number("abc")
+
+ @pytest.mark.parametrize("value", ["inf", "-inf"])
+ def test_infinity_raises(self, value):
+ with pytest.raises(argparse.ArgumentTypeError, match="must be a finite number"):
+ check_positive_number(value)
+
+ def test_nan_raises(self):
+ with pytest.raises(argparse.ArgumentTypeError, match="must be a finite number"):
+ check_positive_number("nan")
+
+
+class TestRandomTransactionId:
+ """Tests for MAX_TRANSACTION_ID constant and _random_transaction_id helper."""
+
+ def test_max_transaction_id_value(self):
+ assert MAX_TRANSACTION_ID == 2**31 - 1
+
+ def test_random_transaction_id_in_range(self):
+ for _ in range(100):
+ tid = _random_transaction_id()
+ assert 1 <= tid <= MAX_TRANSACTION_ID
+
+ def test_random_transaction_id_returns_int(self):
+ assert isinstance(_random_transaction_id(), int)
+
+
+class TestCommandSequencing:
+ """Tests for command sequencing (send_commands and _parse_commands)."""
+
+ async def test_send_commands_executes_in_order(self, mock_connection):
+ cp = ChargePoint(mock_connection)
+ mock_send = AsyncMock()
+ with patch.object(cp, "_send_command", mock_send):
+ commands = [(Action.trigger_message, 0.001), (Action.reset, 0.001)]
+ await cp.send_commands(commands)
+ assert mock_send.call_count == 2
+ assert mock_send.call_args_list[0][0][0] == Action.trigger_message
+ assert mock_send.call_args_list[1][0][0] == Action.reset
+
+ def test_parse_commands_valid(self):
+ result = _parse_commands("Reset:5,TriggerMessage:10")
+ assert result == [(Action.reset, 5.0), (Action.trigger_message, 10.0)]
+
+ def test_parse_commands_invalid_format(self):
+ with pytest.raises(argparse.ArgumentTypeError, match="expected 'CMD:DELAY'"):
+ _parse_commands("ResetOnly")
+
+ def test_parse_commands_unknown_action(self):
+ with pytest.raises(argparse.ArgumentTypeError, match="Unknown action"):
+ _parse_commands("UnknownAction:5")
+
+ def test_parse_commands_case_sensitive(self):
+ with pytest.raises(argparse.ArgumentTypeError, match="Unknown action"):
+ _parse_commands("reset:5")
+
+ def test_parse_commands_infinite_delay(self):
+ with pytest.raises(argparse.ArgumentTypeError, match="finite positive"):
+ _parse_commands("Reset:inf")
+
+ def test_parse_commands_nan_delay(self):
+ with pytest.raises(argparse.ArgumentTypeError, match="finite positive"):
+ _parse_commands("Reset:nan")
+
+ def test_parse_commands_non_numeric_delay(self):
+ with pytest.raises(argparse.ArgumentTypeError, match="must be a number"):
+ _parse_commands("Reset:abc")
+
+ def test_parse_commands_skips_empty_entries(self):
+ result = _parse_commands("Reset:5,,ClearCache:10")
+ assert result == [(Action.reset, 5.0), (Action.clear_cache, 10.0)]
+
+
+class TestHandlerCoverage:
+ """Tests verifying all expected OCPP 1.6 handlers and commands are implemented."""
+
+ EXPECTED_INCOMING_HANDLERS: ClassVar[list[str]] = [
+ "on_boot_notification",
+ "on_authorize",
+ "on_start_transaction",
+ "on_stop_transaction",
+ "on_meter_values",
+ "on_status_notification",
+ "on_heartbeat",
+ "on_data_transfer",
+ "on_firmware_status_notification",
+ "on_diagnostics_status_notification",
+ ]
+
+ EXPECTED_OUTGOING_COMMANDS: ClassVar[list[str]] = [
+ "_send_trigger_message",
+ "_send_remote_start_transaction",
+ "_send_remote_stop_transaction",
+ "_send_reset",
+ "_send_change_availability",
+ "_send_update_firmware",
+ "_send_get_diagnostics",
+ "_send_reserve_now",
+ "_send_cancel_reservation",
+ ]
+
+ @pytest.mark.parametrize("handler_name", EXPECTED_INCOMING_HANDLERS)
+ def test_incoming_handler_exists(self, handler_name):
+ assert hasattr(ChargePoint, handler_name), (
+ f"Missing incoming handler: {handler_name}"
+ )
+ assert callable(getattr(ChargePoint, handler_name))
+ on_action = getattr(getattr(ChargePoint, handler_name), "_on_action", None)
+ assert on_action is not None, (
+ f"Handler {handler_name} is not registered with an @on decorator"
+ )
+ expected_action = Action[handler_name.removeprefix("on_")]
+ assert on_action == expected_action, (
+ f"Handler {handler_name} routes {on_action}, expected {expected_action}"
+ )
+
+ @pytest.mark.parametrize("method_name", EXPECTED_OUTGOING_COMMANDS)
+ def test_outgoing_command_exists(self, method_name):
+ assert hasattr(ChargePoint, method_name), (
+ f"Missing outgoing command method: {method_name}"
+ )
+ assert callable(getattr(ChargePoint, method_name))
+
+
+class TestChargePointDefaultConfig:
+ """Tests for ChargePoint default configuration."""
+
+ def test_default_auth_config(self, charge_point):
+ assert charge_point._auth_config.mode == AuthMode.normal
+ assert charge_point._auth_config.offline is False
+ assert TEST_VALID_TOKEN in charge_point._auth_config.whitelist
+ assert TEST_BLOCKED_TOKEN in charge_point._auth_config.blacklist
+
+ def test_custom_auth_config(self, mock_connection):
+ config = AuthConfig(
+ mode=AuthMode.whitelist,
+ whitelist=("token1",),
+ blacklist=(),
+ offline=False,
+ default_status=AuthorizationStatus.accepted,
+ )
+ cp = ChargePoint(mock_connection, auth_config=config)
+ assert cp._auth_config.mode == AuthMode.whitelist
+ assert cp._auth_config.whitelist == ("token1",)
+
+ def test_command_timer_initially_none(self, charge_point):
+ assert charge_point._command_timer is None
+
+ def test_default_boot_sequence(self, charge_point):
+ assert charge_point._boot_sequence == (RegistrationStatus.accepted,)
+
+ def test_custom_boot_sequence(self, mock_connection):
+ cp = ChargePoint(
+ mock_connection,
+ boot_sequence=(RegistrationStatus.rejected,),
+ )
+ assert cp._boot_sequence == (RegistrationStatus.rejected,)
+
+ def test_empty_boot_sequence_raises(self, mock_connection):
+ with pytest.raises(ValueError, match="at least one status"):
+ ChargePoint(mock_connection, boot_sequence=())
+
+
+class TestResolveAuthStatus:
+ """Tests for the _resolve_auth_status method."""
+
+ def test_normal_mode_accepts(self, charge_point):
+ status = charge_point._resolve_auth_status("any_token")
+ assert status == AuthorizationStatus.accepted
+
+ def test_whitelist_mode_accepts_valid_token(self, whitelist_charge_point):
+ status = whitelist_charge_point._resolve_auth_status(TEST_VALID_TOKEN)
+ assert status == AuthorizationStatus.accepted
+
+ def test_whitelist_mode_blocks_unknown_token(self, whitelist_charge_point):
+ status = whitelist_charge_point._resolve_auth_status("unknown_token")
+ assert status == AuthorizationStatus.blocked
+
+ def test_blacklist_mode_blocks_blacklisted_token(self, blacklist_charge_point):
+ status = blacklist_charge_point._resolve_auth_status(TEST_BLOCKED_TOKEN)
+ assert status == AuthorizationStatus.blocked
+
+ def test_blacklist_mode_accepts_valid_token(self, blacklist_charge_point):
+ status = blacklist_charge_point._resolve_auth_status("good_token")
+ assert status == AuthorizationStatus.accepted
+
+ def test_whitelist_blocks_empty_token(self, whitelist_charge_point):
+ status = whitelist_charge_point._resolve_auth_status("")
+ assert status == AuthorizationStatus.blocked
+
+ def test_blacklist_accepts_empty_token(self, blacklist_charge_point):
+ status = blacklist_charge_point._resolve_auth_status("")
+ assert status == AuthorizationStatus.accepted
+
+
+# --- Async handler tests ---
+
+
+class TestBootNotificationHandler:
+ """Tests for the BootNotification incoming handler."""
+
+ async def test_returns_accepted_by_default(self, charge_point):
+ response = await charge_point.on_boot_notification(
+ charge_point_model=TEST_MODEL,
+ charge_point_vendor=TEST_VENDOR_ID,
+ )
+ assert response.status == RegistrationStatus.accepted
+ assert response.interval == DEFAULT_HEARTBEAT_INTERVAL_SECONDS
+ assert isinstance(response.current_time, str)
+ # current_time must be a parseable ISO-8601 UTC timestamp (consistent with the
+ # Heartbeat handler contract), not merely a string containing "T".
+ parsed = datetime.fromisoformat(response.current_time)
+ assert parsed.tzinfo is not None
+
+ async def test_configurable_boot_status(self, mock_connection):
+ cp = ChargePoint(
+ mock_connection,
+ boot_sequence=(RegistrationStatus.rejected,),
+ )
+ response = await cp.on_boot_notification(
+ charge_point_model=TEST_MODEL,
+ charge_point_vendor=TEST_VENDOR_ID,
+ )
+ assert response.status == RegistrationStatus.rejected
+
+ async def test_pending_boot_status(self, mock_connection):
+ cp = ChargePoint(
+ mock_connection,
+ boot_sequence=(RegistrationStatus.pending,),
+ )
+ response = await cp.on_boot_notification(
+ charge_point_model=TEST_MODEL,
+ charge_point_vendor=TEST_VENDOR_ID,
+ )
+ assert response.status == RegistrationStatus.pending
+
+ async def test_boot_notification_sequence_iterates(self, mock_connection):
+ cp = ChargePoint(
+ mock_connection,
+ boot_sequence=(
+ RegistrationStatus.pending,
+ RegistrationStatus.pending,
+ RegistrationStatus.accepted,
+ ),
+ )
+ r1 = await cp.on_boot_notification(
+ charge_point_model=TEST_MODEL, charge_point_vendor=TEST_VENDOR_ID
+ )
+ r2 = await cp.on_boot_notification(
+ charge_point_model=TEST_MODEL, charge_point_vendor=TEST_VENDOR_ID
+ )
+ r3 = await cp.on_boot_notification(
+ charge_point_model=TEST_MODEL, charge_point_vendor=TEST_VENDOR_ID
+ )
+ assert r1.status == RegistrationStatus.pending
+ assert r2.status == RegistrationStatus.pending
+ assert r3.status == RegistrationStatus.accepted
+
+ async def test_boot_notification_sequence_clamps_to_last(self, mock_connection):
+ cp = ChargePoint(
+ mock_connection,
+ boot_sequence=(
+ RegistrationStatus.pending,
+ RegistrationStatus.accepted,
+ ),
+ )
+ for _ in range(3):
+ response = await cp.on_boot_notification(
+ charge_point_model=TEST_MODEL, charge_point_vendor=TEST_VENDOR_ID
+ )
+ assert response.status == RegistrationStatus.accepted
+
+
+class TestHeartbeatHandler:
+ """Tests for the Heartbeat incoming handler."""
+
+ async def test_returns_current_time(self, charge_point):
+ response = await charge_point.on_heartbeat()
+ assert isinstance(response.current_time, str)
+ # Real contract: current_time must be a parseable ISO-8601 timestamp (UTC),
+ # not merely a string containing "T". datetime.fromisoformat raises on garbage.
+ parsed = datetime.fromisoformat(response.current_time)
+ assert parsed.tzinfo is not None
+ # emitted "now" — within a generous window of the assertion time
+ delta = abs((datetime.now(timezone.utc) - parsed).total_seconds())
+ assert delta < 60
+
+
+class TestStatusNotificationHandler:
+ """Tests for the StatusNotification incoming handler."""
+
+ async def test_returns_empty_response(self, charge_point):
+ response = await charge_point.on_status_notification(
+ connector_id=TEST_CONNECTOR_ID,
+ error_code="NoError",
+ status="Available",
+ )
+ assert isinstance(response, ocpp.v16.call_result.StatusNotification)
+ assert response == ocpp.v16.call_result.StatusNotification()
+
+
+class TestAuthorizeHandler:
+ """Tests for the Authorize incoming handler."""
+
+ async def test_normal_mode_accepts(self, charge_point):
+ response = await charge_point.on_authorize(id_tag="any_token")
+ assert response.id_tag_info["status"] == AuthorizationStatus.accepted
+
+ async def test_whitelist_accepts_valid(self, whitelist_charge_point):
+ response = await whitelist_charge_point.on_authorize(id_tag=TEST_VALID_TOKEN)
+ assert response.id_tag_info["status"] == AuthorizationStatus.accepted
+
+ async def test_whitelist_blocks_unknown(self, whitelist_charge_point):
+ response = await whitelist_charge_point.on_authorize(id_tag="stranger")
+ assert response.id_tag_info["status"] == AuthorizationStatus.blocked
+
+ async def test_blacklist_blocks_blacklisted(self, blacklist_charge_point):
+ response = await blacklist_charge_point.on_authorize(id_tag=TEST_BLOCKED_TOKEN)
+ assert response.id_tag_info["status"] == AuthorizationStatus.blocked
+
+ async def test_blacklist_accepts_unlisted(self, blacklist_charge_point):
+ response = await blacklist_charge_point.on_authorize(id_tag="unlisted_token")
+ assert response.id_tag_info["status"] == AuthorizationStatus.accepted
+
+ async def test_offline_raises_internal_error(self, offline_charge_point):
+ from ocpp.exceptions import InternalError
+
+ with pytest.raises(InternalError):
+ await offline_charge_point.on_authorize(id_tag="any")
+
+ async def test_parent_id_tag_included(self, mock_connection):
+ cp = ChargePoint(
+ mock_connection,
+ auth_config=AuthConfig(
+ mode=AuthMode.normal,
+ whitelist=(),
+ blacklist=(),
+ offline=False,
+ default_status=AuthorizationStatus.accepted,
+ parent_id_tag="ParentTag",
+ ),
+ )
+ response = await cp.on_authorize(id_tag=TEST_TOKEN)
+ assert response.id_tag_info["parent_id_tag"] == "ParentTag"
+
+ async def test_parent_id_tag_absent_by_default(self, charge_point):
+ response = await charge_point.on_authorize(id_tag=TEST_TOKEN)
+ assert "parent_id_tag" not in response.id_tag_info
+
+
+class TestStartTransactionHandler:
+ """Tests for the StartTransaction incoming handler."""
+
+ async def test_returns_transaction_id_and_id_tag_info(self, charge_point):
+ response = await charge_point.on_start_transaction(
+ connector_id=TEST_CONNECTOR_ID,
+ id_tag=TEST_TOKEN,
+ meter_start=TEST_METER_START,
+ timestamp=TEST_TIMESTAMP,
+ )
+ assert isinstance(response, ocpp.v16.call_result.StartTransaction)
+ assert isinstance(response.transaction_id, int)
+ assert response.transaction_id > 0
+ assert response.id_tag_info["status"] == AuthorizationStatus.accepted
+
+ async def test_stores_active_transaction(self, charge_point):
+ response = await charge_point.on_start_transaction(
+ connector_id=TEST_CONNECTOR_ID,
+ id_tag=TEST_TOKEN,
+ meter_start=TEST_METER_START,
+ timestamp=TEST_TIMESTAMP,
+ )
+ assert response.transaction_id in charge_point._active_transactions
+ assert (
+ charge_point._active_transactions[response.transaction_id]
+ == TEST_CONNECTOR_ID
+ )
+
+
+class TestStopTransactionHandler:
+ """Tests for the StopTransaction incoming handler."""
+
+ async def test_returns_empty_when_no_id_tag(self, charge_point):
+ response = await charge_point.on_stop_transaction(
+ meter_stop=TEST_METER_STOP,
+ timestamp=TEST_TIMESTAMP,
+ transaction_id=TEST_TRANSACTION_ID,
+ )
+ assert isinstance(response, ocpp.v16.call_result.StopTransaction)
+ assert response.id_tag_info is None
+
+ async def test_returns_id_tag_info_when_id_tag_present(self, charge_point):
+ response = await charge_point.on_stop_transaction(
+ meter_stop=TEST_METER_STOP,
+ timestamp=TEST_TIMESTAMP,
+ transaction_id=TEST_TRANSACTION_ID,
+ id_tag=TEST_TOKEN,
+ )
+ assert response.id_tag_info["status"] == AuthorizationStatus.accepted
+
+ async def test_removes_active_transaction(self, charge_point):
+ charge_point._active_transactions[TEST_TRANSACTION_ID] = TEST_CONNECTOR_ID
+ await charge_point.on_stop_transaction(
+ meter_stop=TEST_METER_STOP,
+ timestamp=TEST_TIMESTAMP,
+ transaction_id=TEST_TRANSACTION_ID,
+ )
+ assert TEST_TRANSACTION_ID not in charge_point._active_transactions
+
+ async def test_detects_signed_transaction_data(self, charge_point, caplog):
+ caplog.set_level(logging.INFO)
+ transaction_data = [
+ {
+ "timestamp": TEST_TIMESTAMP,
+ "sampled_value": [
+ {
+ "value": "signed_blob",
+ "format": ValueFormat.signed_data.value,
+ "measurand": "Energy.Active.Import.Register",
+ "context": "Transaction.End",
+ }
+ ],
+ }
+ ]
+ await charge_point.on_stop_transaction(
+ meter_stop=TEST_METER_STOP,
+ timestamp=TEST_TIMESTAMP,
+ transaction_id=TEST_TRANSACTION_ID,
+ transaction_data=transaction_data,
+ )
+ assert any("signed meter value" in r.message.lower() for r in caplog.records)
+
+
+class TestNotificationHandlers:
+ """Tests for notification incoming handlers with empty responses."""
+
+ async def test_meter_values(self, charge_point):
+ response = await charge_point.on_meter_values(
+ connector_id=TEST_CONNECTOR_ID,
+ meter_value=[{"timestamp": TEST_TIMESTAMP}],
+ )
+ # Empty-response contract: exact type AND no extra payload fields (equals a
+ # fresh empty result — a regression adding stray fields would fail this).
+ assert isinstance(response, ocpp.v16.call_result.MeterValues)
+ assert response == ocpp.v16.call_result.MeterValues()
+
+ async def test_meter_values_with_signed_meter_value(self, charge_point, caplog):
+ caplog.set_level(logging.INFO)
+ meter_value = [
+ {
+ "timestamp": TEST_TIMESTAMP,
+ "sampled_value": [
+ {
+ "value": "signed_blob",
+ "format": ValueFormat.signed_data.value,
+ "measurand": "Energy.Active.Import.Register",
+ "context": "Sample.Periodic",
+ }
+ ],
+ }
+ ]
+ response = await charge_point.on_meter_values(
+ connector_id=TEST_CONNECTOR_ID, meter_value=meter_value
+ )
+ assert isinstance(response, ocpp.v16.call_result.MeterValues)
+ assert any("signed meter value" in r.message.lower() for r in caplog.records)
+
+ async def test_data_transfer(self, charge_point):
+ response = await charge_point.on_data_transfer(vendor_id=TEST_VENDOR_ID)
+ assert response.status == DataTransferStatus.accepted
+
+ async def test_firmware_status_notification(self, charge_point):
+ response = await charge_point.on_firmware_status_notification(
+ status="Installed"
+ )
+ assert isinstance(response, ocpp.v16.call_result.FirmwareStatusNotification)
+ assert response == ocpp.v16.call_result.FirmwareStatusNotification()
+
+ async def test_diagnostics_status_notification(self, charge_point):
+ response = await charge_point.on_diagnostics_status_notification(
+ status="Uploaded"
+ )
+ assert isinstance(response, ocpp.v16.call_result.DiagnosticsStatusNotification)
+ assert response == ocpp.v16.call_result.DiagnosticsStatusNotification()
+
+
+class TestLogSignedMeterValues:
+ """Tests for the _log_signed_meter_values helper."""
+
+ def test_logs_signed_data(self, caplog):
+ caplog.set_level(logging.INFO)
+ meter_value = [
+ {
+ "sampled_value": [
+ {
+ "value": "blob",
+ "format": ValueFormat.signed_data.value,
+ "measurand": "Energy.Active.Import.Register",
+ "context": "Sample.Periodic",
+ }
+ ],
+ }
+ ]
+ _log_signed_meter_values(meter_value)
+ assert any("signed meter value" in r.message.lower() for r in caplog.records)
+
+ def test_ignores_unsigned_data(self, caplog):
+ caplog.set_level(logging.INFO)
+ meter_value = [
+ {
+ "sampled_value": [
+ {
+ "value": "10.5",
+ "measurand": "Energy.Active.Import.Register",
+ }
+ ],
+ }
+ ]
+ _log_signed_meter_values(meter_value)
+ assert not any(
+ "signed meter value" in r.message.lower() for r in caplog.records
+ )
+
+ def test_empty_list_no_error(self, caplog):
+ caplog.set_level(logging.INFO)
+ _log_signed_meter_values([])
+ assert not any(
+ "signed meter value" in r.message.lower() for r in caplog.records
+ )
+
+
+class TestOutgoingCommands:
+ """Behavioral tests for the _send_* outgoing command methods."""
+
+ async def test_send_trigger_message(self, command_charge_point):
+ command_charge_point.call.return_value = ocpp.v16.call_result.TriggerMessage(
+ status=TriggerMessageStatus.accepted
+ )
+ await command_charge_point._send_trigger_message()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v16.call.TriggerMessage)
+ assert request.requested_message == MessageTrigger.status_notification
+ assert request.connector_id == command_charge_point._connector_id
+
+ async def test_send_trigger_message_custom_type(self, mock_connection):
+ cp = ChargePoint(
+ mock_connection,
+ trigger_message_type=MessageTrigger.boot_notification,
+ )
+ cp.call = AsyncMock(
+ return_value=ocpp.v16.call_result.TriggerMessage(
+ status=TriggerMessageStatus.accepted
+ )
+ )
+ await cp._send_trigger_message()
+ request = cp.call.call_args[0][0]
+ assert request.requested_message == MessageTrigger.boot_notification
+
+ async def test_send_remote_start_transaction(self, command_charge_point):
+ command_charge_point.call.return_value = (
+ ocpp.v16.call_result.RemoteStartTransaction(
+ status=RemoteStartStopStatus.accepted
+ )
+ )
+ await command_charge_point._send_remote_start_transaction()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v16.call.RemoteStartTransaction)
+ assert request.id_tag == DEFAULT_TEST_TOKEN
+ assert request.connector_id == command_charge_point._connector_id
+
+ async def test_send_remote_stop_transaction_fallback(self, command_charge_point):
+ command_charge_point.call.return_value = (
+ ocpp.v16.call_result.RemoteStopTransaction(
+ status=RemoteStartStopStatus.accepted
+ )
+ )
+ await command_charge_point._send_remote_stop_transaction()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v16.call.RemoteStopTransaction)
+ assert request.transaction_id == FALLBACK_TRANSACTION_ID
+
+ async def test_send_remote_stop_transaction_uses_active(self, command_charge_point):
+ command_charge_point._active_transactions[999] = TEST_CONNECTOR_ID
+ command_charge_point.call.return_value = (
+ ocpp.v16.call_result.RemoteStopTransaction(
+ status=RemoteStartStopStatus.accepted
+ )
+ )
+ await command_charge_point._send_remote_stop_transaction()
+ request = command_charge_point.call.call_args[0][0]
+ assert request.transaction_id == 999
+
+ async def test_send_reset(self, command_charge_point):
+ command_charge_point.call.return_value = ocpp.v16.call_result.Reset(
+ status=ResetStatus.accepted
+ )
+ await command_charge_point._send_reset()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v16.call.Reset)
+ assert request.type == ResetType.hard
+
+ async def test_send_reset_soft(self, mock_connection):
+ cp = ChargePoint(mock_connection, reset_type=ResetType.soft)
+ cp.call = AsyncMock(
+ return_value=ocpp.v16.call_result.Reset(status=ResetStatus.accepted)
+ )
+ await cp._send_reset()
+ request = cp.call.call_args[0][0]
+ assert request.type == ResetType.soft
+
+ async def test_send_change_availability(self, command_charge_point):
+ command_charge_point.call.return_value = (
+ ocpp.v16.call_result.ChangeAvailability(status=AvailabilityStatus.accepted)
+ )
+ await command_charge_point._send_change_availability()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v16.call.ChangeAvailability)
+ assert request.connector_id == command_charge_point._connector_id
+ assert request.type == AvailabilityType.operative
+
+ async def test_send_change_availability_inoperative(self, mock_connection):
+ cp = ChargePoint(
+ mock_connection,
+ availability_type=AvailabilityType.inoperative,
+ )
+ cp.call = AsyncMock(
+ return_value=ocpp.v16.call_result.ChangeAvailability(
+ status=AvailabilityStatus.accepted
+ )
+ )
+ await cp._send_change_availability()
+ request = cp.call.call_args[0][0]
+ assert request.type == AvailabilityType.inoperative
+
+ async def test_send_update_firmware(self, command_charge_point):
+ command_charge_point.call.return_value = ocpp.v16.call_result.UpdateFirmware()
+ await command_charge_point._send_update_firmware()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v16.call.UpdateFirmware)
+ assert request.location == DEFAULT_FIRMWARE_URL
+ assert isinstance(request.retrieve_date, str)
+
+ async def test_send_get_diagnostics(self, command_charge_point):
+ command_charge_point.call.return_value = ocpp.v16.call_result.GetDiagnostics(
+ file_name="diag.log"
+ )
+ await command_charge_point._send_get_diagnostics()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v16.call.GetDiagnostics)
+ assert request.location == DEFAULT_DIAGNOSTICS_URL
+
+ async def test_send_reserve_now(self, command_charge_point):
+ command_charge_point.call.return_value = ocpp.v16.call_result.ReserveNow(
+ status=ReservationStatus.accepted
+ )
+ await command_charge_point._send_reserve_now()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v16.call.ReserveNow)
+ assert request.connector_id == command_charge_point._reserve_connector_id
+ assert request.id_tag == DEFAULT_RESERVE_ID_TAG
+ assert request.reservation_id == command_charge_point._reservation_id
+ assert isinstance(request.expiry_date, str)
+
+ async def test_send_cancel_reservation(self, command_charge_point):
+ command_charge_point.call.return_value = ocpp.v16.call_result.CancelReservation(
+ status=CancelReservationStatus.accepted
+ )
+ await command_charge_point._send_cancel_reservation()
+ command_charge_point.call.assert_called_once()
+ request = command_charge_point.call.call_args[0][0]
+ assert isinstance(request, ocpp.v16.call.CancelReservation)
+ assert request.reservation_id == command_charge_point._reservation_id
+
+ FAILURE_PATH_CASES: ClassVar[list[tuple[str, type, object]]] = [
+ (
+ "_send_trigger_message",
+ ocpp.v16.call_result.TriggerMessage,
+ TriggerMessageStatus.rejected,
+ ),
+ (
+ "_send_remote_start_transaction",
+ ocpp.v16.call_result.RemoteStartTransaction,
+ RemoteStartStopStatus.rejected,
+ ),
+ (
+ "_send_remote_stop_transaction",
+ ocpp.v16.call_result.RemoteStopTransaction,
+ RemoteStartStopStatus.rejected,
+ ),
+ (
+ "_send_reset",
+ ocpp.v16.call_result.Reset,
+ ResetStatus.rejected,
+ ),
+ (
+ "_send_change_availability",
+ ocpp.v16.call_result.ChangeAvailability,
+ AvailabilityStatus.rejected,
+ ),
+ (
+ "_send_reserve_now",
+ ocpp.v16.call_result.ReserveNow,
+ ReservationStatus.rejected,
+ ),
+ (
+ "_send_cancel_reservation",
+ ocpp.v16.call_result.CancelReservation,
+ CancelReservationStatus.rejected,
+ ),
+ ]
+
+ @pytest.mark.parametrize(
+ ("method_name", "result_cls", "failed_status"),
+ FAILURE_PATH_CASES,
+ ids=[c[0] for c in FAILURE_PATH_CASES],
+ )
+ async def test_send_command_failure_logs(
+ self, command_charge_point, caplog, method_name, result_cls, failed_status
+ ):
+ caplog.set_level(logging.INFO)
+ command_charge_point.call.return_value = result_cls(status=failed_status)
+ await getattr(command_charge_point, method_name)()
+ assert any(
+ r.levelno == logging.INFO and "failed" in r.message.lower()
+ for r in caplog.records
+ )
+
+
+class TestSendCommandErrorHandling:
+ """Tests for error handling in the command dispatch layer."""
+
+ async def test_timeout_is_caught(self, charge_point, caplog):
+ caplog.set_level(logging.ERROR)
+ with patch.object(
+ charge_point, "_send_reset", side_effect=TimeoutError("timed out")
+ ):
+ await charge_point._send_command(command_name=Action.reset)
+ assert any(
+ r.levelno == logging.ERROR and "timeout waiting for" in r.message.lower()
+ for r in caplog.records
+ )
+
+ async def test_ocpp_error_is_caught(self, charge_point, caplog):
+ from ocpp.exceptions import InternalError as OCPPInternalError
+
+ caplog.set_level(logging.ERROR)
+ with patch.object(
+ charge_point,
+ "_send_reset",
+ side_effect=OCPPInternalError(description="test error"),
+ ):
+ await charge_point._send_command(command_name=Action.reset)
+ assert any(
+ r.levelno == logging.ERROR
+ and "ocpp error sending" in r.message.lower()
+ and "test error" in r.message
+ for r in caplog.records
+ )
+
+ async def test_connection_closed_is_caught(self, charge_point):
+ from websockets.exceptions import ConnectionClosedOK
+ from websockets.frames import Close
+
+ with (
+ patch.object(
+ charge_point,
+ "_send_reset",
+ side_effect=ConnectionClosedOK(
+ Close(1000, ""), Close(1000, ""), rcvd_then_sent=True
+ ),
+ ),
+ patch.object(charge_point, "handle_connection_closed"),
+ ):
+ await charge_point._send_command(command_name=Action.reset)
+ charge_point.handle_connection_closed.assert_called_once()
+
+ async def test_unexpected_error_is_caught(self, charge_point, caplog):
+ caplog.set_level(logging.ERROR)
+ with patch.object(
+ charge_point, "_send_reset", side_effect=RuntimeError("boom")
+ ):
+ await charge_point._send_command(command_name=Action.reset)
+ assert any(
+ r.levelno == logging.ERROR
+ and "unexpected error sending" in r.message.lower()
+ for r in caplog.records
+ )
+
+ async def test_unsupported_command_logs_warning(self, charge_point, caplog):
+ caplog.set_level(logging.WARNING)
+ unsupported = MagicMock(value="Unsupported")
+ await charge_point._send_command(command_name=unsupported)
+ assert any("not supported" in r.message.lower() for r in caplog.records)
+
+
+class TestSendCommand:
+ """Tests for Timer creation logic in send_command."""
+
+ async def test_delay_creates_one_shot_timer(self, charge_point):
+ with patch("server16.Timer") as MockTimer:
+ MockTimer.return_value = MagicMock()
+ await charge_point.send_command(Action.reset, delay=1.0, period=None)
+ MockTimer.assert_called_once()
+ args = MockTimer.call_args[0]
+ assert args[0] == 1.0
+ assert args[1] is False
+
+ async def test_period_creates_repeating_timer(self, charge_point):
+ with patch("server16.Timer") as MockTimer:
+ MockTimer.return_value = MagicMock()
+ await charge_point.send_command(Action.reset, delay=None, period=1.0)
+ MockTimer.assert_called_once()
+ args = MockTimer.call_args[0]
+ assert args[0] == 1.0
+ assert args[1] is True
+
+ async def test_no_timer_when_both_none(self, charge_point):
+ with patch("server16.Timer") as MockTimer:
+ await charge_point.send_command(Action.reset, delay=None, period=None)
+ MockTimer.assert_not_called()
+
+ async def test_second_call_no_op_when_timer_exists(self, charge_point):
+ charge_point._command_timer = MagicMock()
+ with patch("server16.Timer") as MockTimer:
+ await charge_point.send_command(Action.reset, delay=1.0, period=None)
+ MockTimer.assert_not_called()
+
+
+class TestHandleConnectionClosed:
+ """Tests for the handle_connection_closed cleanup method."""
+
+ def test_timer_cancelled_on_close(self, charge_point):
+ mock_timer = MagicMock()
+ charge_point._command_timer = mock_timer
+ charge_point.handle_connection_closed()
+ mock_timer.cancel.assert_called_once()
+
+ def test_commands_task_cancelled_on_close(self, charge_point):
+ mock_task = MagicMock()
+ charge_point._commands_task = mock_task
+ charge_point.handle_connection_closed()
+ mock_task.cancel.assert_called_once()
+
+ def test_timer_none_no_error(self, charge_point):
+ charge_point._command_timer = None
+ charge_point.handle_connection_closed()
+
+ def test_charge_point_removed_from_set(self, charge_point):
+ assert charge_point in charge_point._charge_points
+ charge_point.handle_connection_closed()
+ assert charge_point not in charge_point._charge_points
+
+ def test_charge_point_not_in_set_no_error(self, charge_point):
+ charge_point._charge_points = set()
+ charge_point.handle_connection_closed()
+
+
+class TestOnConnect:
+ """Tests for the on_connect WebSocket connection handler."""
+
+ @staticmethod
+ def _make_config(**overrides):
+ defaults: dict[str, Any] = {
+ "command_name": None,
+ "delay": None,
+ "period": None,
+ "auth_config": AuthConfig(
+ mode=AuthMode.normal,
+ whitelist=(),
+ blacklist=(),
+ offline=False,
+ default_status=AuthorizationStatus.accepted,
+ ),
+ "boot_sequence": (RegistrationStatus.accepted,),
+ "connector_id": TEST_CONNECTOR_ID,
+ "charge_points": set(),
+ }
+ defaults.update(overrides)
+ return ServerConfig(**defaults)
+
+ async def test_missing_subprotocol_header_closes_connection(self):
+ mock_ws = MagicMock()
+ mock_ws.request = MagicMock()
+ mock_ws.request.headers = {}
+ mock_ws.close = AsyncMock()
+ config = self._make_config()
+
+ await on_connect(mock_ws, config=config)
+ mock_ws.close.assert_called_once()
+ # Behavioural contract: a rejected connection must NOT register a charge point.
+ assert len(config.charge_points) == 0
+
+ async def test_protocol_mismatch_closes_connection(self):
+ mock_ws = MagicMock()
+ mock_ws.request = MagicMock()
+ mock_ws.request.headers = {"Sec-WebSocket-Protocol": "ocpp1.6"}
+ mock_ws.subprotocol = None
+ mock_ws.close = AsyncMock()
+ config = self._make_config()
+
+ await on_connect(mock_ws, config=config)
+ mock_ws.close.assert_called_once()
+ assert len(config.charge_points) == 0
+
+ async def test_successful_connection_creates_charge_point(self, mock_valid_ws):
+ config = self._make_config()
+
+ with patch("server16.ChargePoint") as MockCP:
+ mock_cp = AsyncMock()
+ MockCP.return_value = mock_cp
+ await on_connect(mock_valid_ws, config=config)
+ mock_cp.start.assert_called_once()
+
+ async def test_connection_closed_during_start_triggers_cleanup(self, mock_valid_ws):
+ from websockets.exceptions import ConnectionClosedOK
+ from websockets.frames import Close
+
+ config = self._make_config()
+ exc = ConnectionClosedOK(Close(1000, ""), Close(1000, ""), rcvd_then_sent=True)
+
+ with patch("server16.ChargePoint") as MockCP:
+ mock_cp = AsyncMock()
+ mock_cp.start = AsyncMock(side_effect=exc)
+ mock_cp.handle_connection_closed = MagicMock()
+ MockCP.return_value = mock_cp
+ await on_connect(mock_valid_ws, config=config)
+ mock_cp.handle_connection_closed.assert_called_once()
+
+ async def test_command_sent_on_connect_when_specified(self, mock_valid_ws):
+ config = self._make_config(command_name=Action.reset, delay=1.0, period=None)
+
+ with patch("server16.ChargePoint") as MockCP:
+ mock_cp = AsyncMock()
+ MockCP.return_value = mock_cp
+ await on_connect(mock_valid_ws, config=config)
+ mock_cp.send_command.assert_called_once_with(Action.reset, 1.0, None)
+
+ async def test_commands_sequence_scheduled_on_connect(self, mock_valid_ws):
+ config = self._make_config(commands=[(Action.reset, 1.0)])
+
+ with (
+ patch("server16.ChargePoint") as MockCP,
+ patch("server16.asyncio.create_task") as mock_create_task,
+ ):
+ mock_cp = AsyncMock()
+ mock_cp.send_commands = MagicMock()
+ MockCP.return_value = mock_cp
+ await on_connect(mock_valid_ws, config=config)
+ mock_create_task.assert_called_once()
+
+
+class TestMainGracefulShutdown:
+ """Tests for the main() graceful shutdown logic."""
+
+ @pytest.mark.parametrize("sig", [signal.SIGINT, signal.SIGTERM])
+ async def test_first_signal_closes_server_and_sets_event(self, main_mocks, sig):
+ mock_loop, mock_server, mock_event, signal_handlers = main_mocks
+
+ async def _fire_signal():
+ handler, args = signal_handlers[sig]
+ handler(*args)
+
+ mock_event.wait = AsyncMock(side_effect=_fire_signal)
+
+ with _patch_main(mock_loop, mock_server, mock_event):
+ await main()
+
+ mock_server.close.assert_called_once()
+ mock_event.set.assert_called_once()
+ mock_server.wait_closed.assert_called_once()
+
+ async def test_second_signal_forces_exit(self, main_mocks):
+ mock_loop, mock_server, mock_event, signal_handlers = main_mocks
+
+ async def _fire_twice():
+ handler, args = signal_handlers[signal.SIGINT]
+ handler(*args)
+ handler(*args)
+
+ mock_event.wait = AsyncMock(side_effect=_fire_twice)
+
+ with _patch_main(mock_loop, mock_server, mock_event):
+ with pytest.raises(SystemExit) as exc_info:
+ await main()
+ assert exc_info.value.code == 128 + signal.SIGINT.value
+
+ async def test_shutdown_timeout_logs_warning(self, main_mocks, caplog):
+ mock_loop, mock_server, mock_event, signal_handlers = main_mocks
+ mock_server.wait_closed = AsyncMock(side_effect=TimeoutError)
+
+ async def _fire_sigint():
+ handler, args = signal_handlers[signal.SIGINT]
+ handler(*args)
+
+ mock_event.wait = AsyncMock(side_effect=_fire_sigint)
+ caplog.set_level(logging.WARNING)
+
+ with _patch_main(mock_loop, mock_server, mock_event):
+ await main()
+
+ assert any(
+ r.levelno == logging.WARNING and "timed out" in r.message.lower()
+ for r in caplog.records
+ )
+
+ async def test_windows_handler_schedules_via_call_soon_threadsafe(self, main_mocks):
+ mock_loop, mock_server, mock_event, _ = main_mocks
+ mock_loop.add_signal_handler = MagicMock(side_effect=NotImplementedError)
+ mock_loop.call_soon_threadsafe = MagicMock()
+ mock_event.wait = AsyncMock()
+
+ captured_handlers: dict[int, Any] = {}
+
+ def _capture_signal(sig: int, handler: Any) -> None:
+ captured_handlers[sig] = handler
+
+ with _patch_main(
+ mock_loop,
+ mock_server,
+ mock_event,
+ extra_patches=[patch("_common.signal.signal", side_effect=_capture_signal)],
+ ):
+ await main()
+
+ sigint_handler = captured_handlers[signal.SIGINT]
+ assert callable(sigint_handler)
+ sigint_handler(signal.SIGINT.value, None)
+ mock_loop.call_soon_threadsafe.assert_called_once()
+
+ async def test_windows_fallback_registers_signal_handlers(self, main_mocks):
+ mock_loop, mock_server, mock_event, _ = main_mocks
+ mock_loop.add_signal_handler = MagicMock(side_effect=NotImplementedError)
+ mock_event.wait = AsyncMock()
+
+ mock_signal_fn = MagicMock()
+
+ with _patch_main(
+ mock_loop,
+ mock_server,
+ mock_event,
+ extra_patches=[patch("_common.signal.signal", mock_signal_fn)],
+ ):
+ await main()
+
+ assert mock_signal_fn.call_count == 2
+ registered_signals = {call.args[0] for call in mock_signal_fn.call_args_list}
+ assert registered_signals == {signal.SIGINT, signal.SIGTERM}
+
+ async def test_boot_status_sequence_parsed(self, main_mocks):
+ mock_loop, mock_server, mock_event, signal_handlers = main_mocks
+
+ async def _fire_sigint():
+ handler, args = signal_handlers[signal.SIGINT]
+ handler(*args)
+
+ mock_event.wait = AsyncMock(side_effect=_fire_sigint)
+
+ captured: dict[str, ServerConfig] = {}
+ real_server_config = ServerConfig
+
+ def _capture_server_config(*args, **kwargs):
+ config = real_server_config(*args, **kwargs)
+ captured["config"] = config
+ return config
+
+ with _patch_main(
+ mock_loop,
+ mock_server,
+ mock_event,
+ extra_patches=[
+ patch("server16.ServerConfig", side_effect=_capture_server_config)
+ ],
+ boot_status_sequence="Pending,Accepted",
+ ):
+ await main()
+
+ mock_server.close.assert_called_once()
+ assert captured["config"].boot_sequence == (
+ RegistrationStatus.pending,
+ RegistrationStatus.accepted,
+ )
+
+ async def test_boot_status_single_value(self, main_mocks):
+ mock_loop, mock_server, mock_event, signal_handlers = main_mocks
+
+ async def _fire_sigint():
+ handler, args = signal_handlers[signal.SIGINT]
+ handler(*args)
+
+ mock_event.wait = AsyncMock(side_effect=_fire_sigint)
+
+ captured: dict[str, ServerConfig] = {}
+ real_server_config = ServerConfig
+
+ def _capture_server_config(*args, **kwargs):
+ config = real_server_config(*args, **kwargs)
+ captured["config"] = config
+ return config
+
+ with _patch_main(
+ mock_loop,
+ mock_server,
+ mock_event,
+ extra_patches=[
+ patch("server16.ServerConfig", side_effect=_capture_server_config)
+ ],
+ boot_status=RegistrationStatus.rejected,
+ ):
+ await main()
+
+ mock_server.close.assert_called_once()
+ assert captured["config"].boot_sequence == (RegistrationStatus.rejected,)
repositories / e-mobility-charging-stations-simulator.git / commitdiff
