From: Jérôme Benoit Date: Mon, 27 Jul 2026 12:53:40 +0000 (+0200) Subject: test(ocpp-server): add OCPP 1.6 mock server and extend reservation coverage (#2047) X-Git-Tag: cli@v4.11.0~3 X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=82ae6d1a78470403022fbb31dac35dfd46e91c85;p=e-mobility-charging-stations-simulator.git test(ocpp-server): add OCPP 1.6 mock server and extend reservation coverage (#2047) * test(ocpp-server): add OCPP 1.6 mock server and extend reservation coverage - Add server16.py: standalone OCPP 1.6 mock server mirroring server.py architecture (AuthConfig, ServerConfig, ChargePoint, on_connect, main) - Add test_server16.py: 59 tests covering handlers, outgoing commands, signed meter values, timer, on_connect and main (~96% branch coverage) - Add _send_reserve_now/_send_cancel_reservation to server.py (OCPP 2.0.1) with matching ServerConfig fields and DEFAULT_RESERVATION_* constants - Extend test_server.py: tests for ReserveNow/CancelReservation commands, entries in EXPECTED_OUTGOING_COMMANDS and FAILURE_PATH_CASES, strengthen timestamp assertions (fromisoformat + tzinfo check) - Update pyproject.toml: server16 run task, typecheck and coverage source - Add tools/http_broadcast_timing.py and tools/reused_ws_request_id.py: ad-hoc diagnostic scripts for UI-server fixes #2037 and #2033 * chore(serena): update project knowledge base * revert: remove tools/ from tracked files (kept locally) * fix(ocpp-server): address review findings M1-M3 and m1-m8 M1 - test_boot_status_sequence_parsed: reuse _patch_main (add **args_overrides), capture ServerConfig via side_effect, assert boot_sequence == (pending, accepted) M2 - DRY: extract check_positive_number to _common.py; update typecheck task, coverage source, and mypy override to include it M3 - README: rename to 'OCPP Mock Servers', add Running the Servers section, add full OCPP 1.6 server documentation (20 CLI flags, supported commands, transaction tracking) m1 - remove dead DEFAULT_VENDOR_ID constant (server16.py) m2 - align --reservation-id flag (server.py: --reserve-id -> --reservation-id) and fix coupled Namespace in test_server.py m3 - _call_and_log: log response.status on failure (server.py) m4 - pyproject.toml: add test_server16 to mypy disallow_untyped_defs override m5 - add type hints to on_start_transaction, on_stop_transaction, on_status_notification handlers (server16.py) m6 - add test_windows_handler_schedules_via_call_soon_threadsafe (test_server16.py) m7 - document intentional 1.6 command scope in _COMMAND_HANDLERS m8 - remove dead AuthMode.offline member (server.py) INFO - boot_index: clarify shared-across-all-stations behavior INFO - magic number 128: add Unix convention comment INFO - argparse: 'OCPP2 Server' -> 'OCPP 2.0.1 Server' (server.py) INFO - _parse_commands: raw_entry/entry variable naming (server.py) Coverage tests: test_custom_auth_config, token vide x2, test_meter_values_with_signed_meter_value (test_server16.py) Quality gates: 335 passed, format/lint/typecheck clean, 94.78% * fix(ocpp-server): address post-review findings N1-N6 N1/M2 - server16.py: import check_positive_number from _common, remove local duplicate (567-576); _common.py now has a single consumer N1/m1 - server16.py: remove dead DEFAULT_VENDOR_ID constant N1/m5 - server16.py: add type hints to on_start_transaction (connector_id, meter_start: int; id_tag: str), on_stop_transaction (meter_stop, transaction_id: int), on_status_notification (connector_id: int; error_code, status: str) N1/m7 - server16.py: document intentional subset scope above _COMMAND_HANDLERS with runtime behavior note N1/INFO - server16.py: add Unix 128+signal comment before sys.exit; expand boot_index one-liner to full NOTE (shared across ALL stations, not per-station) N2 - test_server.py: extend _patch_main with **args_overrides (same pattern as test_server16.py); add test_boot_status_sequence_parsed in TestMainGracefulShutdown — covers server.py:1134-1149 N3 - test_server16.py: add test_boot_status_single_value — covers server16.py:786-787 (elif --boot-status single value branch) N4 - pyproject.toml: update description to cover both 1.6 and 2.0.1 N5 - README.md: remove duplicated sentence at line 38 N6 - test_server16.py: fix RuntimeWarning (coroutine never awaited) in test_commands_sequence_scheduled_on_connect — mock_cp.send_commands was AsyncMock, its coroutine was passed to mocked create_task unrun; replace with MagicMock() to prevent the orphan coroutine Quality gates: 337 passed (0 warnings), format/lint/typecheck clean, coverage 95.95% (+1.17pp) * test(charging-station): relax template file count assertion Replace strictEqual(15) with ok(count >= 15) so adding new station templates does not require updating the hardcoded count. * fix(ocpp-server): address round-3 review findings MAJOR-1 - README.md 2.0.1: add ReserveNow and CancelReservation to Outgoing Commands list; add --reservation-id, --reserve-id-token, --reserve-evse-id flags to Command-specific options; add example command MAJOR-2a - test_server.py: add caplog assertions to 4 signed MeterValue tests (started/updated/ended/meter_values) — verify _log_signed_meter_values actually logs, not just that the handler returns correct type MAJOR-2b - test_server.py: rewrite test_unsupported_command_logs_warning with caplog + assert 'not supported' in log output MAJOR-2c - test_server.py: align 7 empty-response handlers on equality pattern (== EmptyResult()) on top of isinstance — matches test_server16 MINOR-1 - _common.py: add generic parse_commands(str, type[ActionT]) factored from the two identical _parse_commands; both servers now use a one-line wrapper; remove import math from server16/server (now in _common only); coverage 96.76% (+0.81pp) MINOR-2 - server.py: add intentional-subset scope comment above _COMMAND_HANDLERS (mirrors server16.py:451-453) MINOR-3 - server.py:342: (kwargs.get('evse') or {}).get('id', 0) prevents AttributeError when payload carries evse=null explicitly; add regression test test_transaction_event_started_null_evse_defaults_to_zero MINOR-4 - test_server.py: port 3 missing tests from test_server16: - test_unexpected_error_is_caught (covers server.py:782-783) - test_commands_sequence_scheduled_on_connect (covers server.py:866) - test_boot_status_single_value via main() (covers server.py:1151) MINOR-5 - README.md 1.6: enumerate valid --trigger-message types (StatusNotification, BootNotification, Heartbeat, MeterValues, FirmwareStatusNotification, DiagnosticsStatusNotification, LogStatusNotification, SignChargePointCertificate) Quality gates: 341 passed (0 warnings), format/lint/typecheck clean, coverage 96.76% (+0.81pp from 95.95%) * fix(ocpp-server): address round-4 review findings MAJOR-A - README.md 2.0.1: replace invalid --trigger-message value SignCertificate with the 11 exact MessageTriggerEnumType members (SignChargingStationCertificate/SignV2GCertificate/SignCombinedCertificate + TransactionEvent + PublishFirmwareStatusNotification); the 1.6 list was fixed in round-3, this is its 2.0.1 counterpart MAJOR-B - test_server.py: add TestLogSignedMeterValues (3 tests) adapted to 2.0.1 semantics (signed_meter_value field, not format=SignedData); covers the previously-uncovered unsigned branch server.py:97->95 MINOR-C - test_server.py + test_server16.py: 6 error-handling tests (timeout/ocpp_error/unexpected) upgraded from swallowing-only to caplog assertions on level ERROR + message substring MINOR-D - DRY: extract negotiate_subprotocol + install_signal_handlers_and_wait to _common.py (version-agnostic lifecycle, logger passed explicitly to preserve observability); both servers use them; remove import signal from both servers; retarget 4 signal.signal patches to _common. Instance methods (_call_and_log/_send_command/...) and DEFAULT_RESERVATION_* constants left in place (OCPP coupling / locality cost > DRY gain for a test mock) MINOR-E - _common.py to 100%: add non-numeric-delay and empty-entry tests for _parse_commands in both suites INFO MORT-1 - remove unreachable dead code (if not boot_sequence) in both servers' main() INFO-D - server.py: handle_connection_closed(self) -> None annotation Quality gates: 348 passed (0 warnings), format/lint/typecheck clean, coverage 97.51% (+0.75pp); _common.py 100% * test(ocpp-server): close round-5 parity + config findings MINOR parity - test_server.py: assert len(config.charge_points) == 0 on both on_connect rejection tests (missing subprotocol / protocol mismatch), mirroring test_server16.py; guards against a regression registering a ChargePoint on a rejected connection in 2.0.1 MINOR parity - test_server.py: strengthen Heartbeat test_returns_current_time with a freshness window (delta < 60s) + import timezone, mirroring the 1.6 suite (was isinstance + tzinfo only) INFO I4 - pyproject.toml: mypy python_version 3.14 -> 3.12 to match the requires-python floor (>=3.12); mypy stays clean (verified live) INFO I5 - README.md: add the --boot-status shorthand sentence to the 1.6 section for parity with 2.0.1 (identical behavior) Deliberately NOT changed (accepted debt, evidence-based): - DRY mixin for send_command/send_commands/handle_connection_closed: moving them to _common would emit lifecycle logs under _common instead of server/server16 (silent observability regression) for ~24 lines in a mock - mypy test override kept: real guard (test files carry untyped defs) - docstrings on _common helpers: 0/4 documented, keeping style consistent Quality gates: 348 passed, format/lint/typecheck clean (python 3.12), coverage 97.51% * test(ocpp-server): close round-6 parity + hygiene findings MINOR-1 - test_server16.py: add test_parent_id_tag_absent_by_default, mirroring 2.0.1 test_authorize_no_enrichment_by_default; asserts parent_id_tag is absent from id_tag_info under the default AuthConfig (closes the last inter-suite rigor asymmetry) MINOR-2 - server.py: harmonize _call_and_log docstring with server16.py (both now '...based on its status') MINOR-3 - test_server.py: add test_parse_set_variable_specs_skips_empty_entries and test_parse_get_variable_specs_invalid_no_dot, covering the two previously-uncovered branches of _parse_variable_specs (empty-entry skip, require_value=False missing-dot); server.py 875/885 now covered I1 - server.py + server16.py: derive --auth-mode choices/default from the local AuthMode StrEnum ([mode.value for mode in AuthMode] / AuthMode.normal.value) instead of string literals (AGENTS.md: avoid string literals when an enumeration exists); order/content identical (verified live), no behavior change Quality gates: 351 passed, format/lint/typecheck clean (python 3.12), coverage 97.88% (server.py 98%) * test(ocpp-server): close round-7 style + DRY findings - test_server.py: fix stale docstring (20 -> 24 _send_* methods) - server.py/server16.py: --auth-mode type=str -> type=AuthMode to match the sibling enum-typed CLI args (--boot-status, --reset-type, ...); keep choices for unchanged help/UX - server.py/server16.py: extract DEFAULT_WHITELIST/DEFAULT_BLACKLIST module constants (single source of truth; removes __init__/argparse duplication) - test_server.py/test_server16.py: TestHandlerCoverage now asserts @on registration (_on_action) so an undecorated handler is caught * test(ocpp-server): close round-8 findings (harmonize auth-mode, DRY, handler routing) - _common.py: host DEFAULT_WHITELIST/DEFAULT_BLACKLIST as the single source of truth; server.py and server16.py now import them (removes cross-file literal duplication) - server.py/server16.py: drop now-dead choices=list(AuthMode) from --auth-mode and move valid values into the help text, matching the sibling enum-typed args (--boot-status, --reset-type, ...); uniform invalid-value error message - server.py/server16.py: mode=args.auth_mode (args value is already an AuthMode via type=AuthMode; drop redundant re-wrap) - test_server.py/test_server16.py: TestHandlerCoverage now asserts each handler is registered for the CORRECT OCPP action (_on_action == expected), not merely decorated with some @on * test(ocpp-server): address bot review comments on server16 - test_server16.py: use supported CSMS->CS commands (TriggerMessage/Reset) instead of ClearCache/Heartbeat in the send_commands sequencing and _parse_commands tests, so the fixtures reflect commands actually in _COMMAND_HANDLERS rather than unsupported outgoing actions - server.py/server16.py: replace the empty `except KeyboardInterrupt: pass` with contextlib.suppress(KeyboardInterrupt) (clears the empty-except finding without a comment; both servers for parity) - server.py/server16.py: _resolve_auth_status returns default_status via an unconditional trailing return instead of a wildcard `case _`, making the control flow unambiguous to static analysis (both servers for parity) --- diff --git a/.serena/project.yml b/.serena/project.yml index f80a9ca1..92bec931 100644 --- a/.serena/project.yml +++ b/.serena/project.yml @@ -7,6 +7,13 @@ ignore_all_files_in_gitignore: true # 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: [] @@ -56,41 +63,6 @@ included_optional_tools: [] # 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. @@ -167,3 +139,38 @@ activation_command: # 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 diff --git a/tests/charging-station/TemplateValidation.test.ts b/tests/charging-station/TemplateValidation.test.ts index 1c5747df..24febf91 100644 --- a/tests/charging-station/TemplateValidation.test.ts +++ b/tests/charging-station/TemplateValidation.test.ts @@ -279,13 +279,16 @@ await describe('TemplateValidation', async () => { }) 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') diff --git a/tests/ocpp-server/README.md b/tests/ocpp-server/README.md index 5dd3c2aa..a6ea44c0 100644 --- a/tests/ocpp-server/README.md +++ b/tests/ocpp-server/README.md @@ -1,6 +1,9 @@ -# 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 @@ -18,13 +21,21 @@ Then install dependencies: 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 @@ -103,7 +114,7 @@ poetry run python server.py --commands "RequestStartTransaction:5,RequestStopTra These flags customize the payload of specific commands: - `--trigger-message `: 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 `: Reset type (default: `Immediate`) - `Immediate` — Reset now - `OnIdle` — Reset when no transaction is active @@ -113,6 +124,9 @@ These flags customize the payload of specific commands: - `--set-variables `: SetVariables data as `Component.Variable=Value,...` (values must not contain commas) - `--get-variables `: GetVariables data as `Component.Variable,...` - `--local-list-tokens TOKEN ...`: Tokens to include in SendLocalList (default: test token) +- `--reservation-id `: ReserveNow/CancelReservation reservation id (default: `1`) +- `--reserve-id-token `: ReserveNow id_token value the reservation is bound to (default: `reserved_token`) +- `--reserve-evse-id `: ReserveNow target EVSE id (default: `1`) ```shell poetry run python server.py --command TriggerMessage --trigger-message BootNotification --delay 5 @@ -122,12 +136,14 @@ poetry run python server.py --command SetVariables --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 @@ -143,6 +159,7 @@ poetry run python server.py --command SendLocalList --delay 5 --local-list-token - `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 @@ -173,6 +190,128 @@ poetry run python server.py --command SendLocalList --delay 5 --local-list-token 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 `: Bind address (default: `127.0.0.1`) +- `--port `: Listening port (default: `9000`) +- `--connector-id `: Connector id used in CSMS→CS commands (default: `1`) + +### Boot Behavior + +- `--boot-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 `: 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 `: 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 `: 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 `: OCPP command to send (see supported commands below) +- `--delay `: One-shot delay before sending (mutually exclusive with `--period`) +- `--period `: 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 `: 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 `: TriggerMessage requested message type (default: `StatusNotification`) + - `StatusNotification`, `BootNotification`, `Heartbeat`, `MeterValues`, `FirmwareStatusNotification`, `DiagnosticsStatusNotification`, `LogStatusNotification`, `SignChargePointCertificate` +- `--reset-type `: Reset type (default: `Hard`) + - `Hard` — Reset now + - `Soft` — Graceful reset +- `--availability-type `: ChangeAvailability type (default: `Operative`) + - `Operative` — Connector available + - `Inoperative` — Connector unavailable +- `--reserve-connector-id `: ReserveNow connector id (default: `1`) +- `--reserve-id-tag `: ReserveNow id tag reserving the connector (default: `reserved_tag`) +- `--reservation-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 diff --git a/tests/ocpp-server/_common.py b/tests/ocpp-server/_common.py new file mode 100644 index 00000000..f7676ff7 --- /dev/null +++ b/tests/ocpp-server/_common.py @@ -0,0 +1,131 @@ +"""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, + ) diff --git a/tests/ocpp-server/pyproject.toml b/tests/ocpp-server/pyproject.toml index 5b73d1eb..89fb0920 100644 --- a/tests/ocpp-server/pyproject.toml +++ b/tests/ocpp-server/pyproject.toml @@ -1,7 +1,7 @@ [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" @@ -20,9 +20,10 @@ pytest-cov = "^7.0.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 ." @@ -36,7 +37,7 @@ select = ["E", "W", "F", "ASYNC", "S", "B", "A", "Q", "RUF", "I"] 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 @@ -49,12 +50,12 @@ module = ["ocpp.*"] 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 diff --git a/tests/ocpp-server/server.py b/tests/ocpp-server/server.py index e19f271e..98a7e5db 100644 --- a/tests/ocpp-server/server.py +++ b/tests/ocpp-server/server.py @@ -2,9 +2,8 @@ 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 @@ -20,6 +19,7 @@ from ocpp.routing import on from ocpp.v201.enums import ( Action, AuthorizationStatusEnumType, + CancelReservationStatusEnumType, CertificateSignedStatusEnumType, CertificateSigningUseEnumType, ChangeAvailabilityStatusEnumType, @@ -41,6 +41,7 @@ from ocpp.v201.enums import ( OperationalStatusEnumType, RegistrationStatusEnumType, ReportBaseEnumType, + ReserveNowStatusEnumType, ResetEnumType, ResetStatusEnumType, SendLocalListStatusEnumType, @@ -53,6 +54,14 @@ from ocpp.v201.enums import ( ) 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__) @@ -75,6 +84,9 @@ DEFAULT_FIRMWARE_URL = "https://example.com/firmware/v2.0.bin" 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 @@ -109,7 +121,6 @@ class AuthMode(StrEnum): whitelist = "whitelist" blacklist = "blacklist" rate_limit = "rate_limit" - offline = "offline" @dataclass(frozen=True) @@ -137,7 +148,10 @@ class ServerConfig: 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 = ( @@ -148,6 +162,9 @@ class ServerConfig: 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): @@ -166,6 +183,9 @@ 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, @@ -187,6 +207,9 @@ class ChargePoint(ocpp.v201.ChargePoint): 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] @@ -209,13 +232,16 @@ class ChargePoint(ocpp.v201.ChargePoint): 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, ) @@ -239,8 +265,7 @@ class ChargePoint(ocpp.v201.ChargePoint): ) 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.""" @@ -320,7 +345,7 @@ class ChargePoint(ocpp.v201.ChargePoint): 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: @@ -427,12 +452,12 @@ class ChargePoint(ocpp.v201.ChargePoint): # --- 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() @@ -688,8 +713,33 @@ class ChargePoint(ocpp.v201.ChargePoint): 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", @@ -713,6 +763,8 @@ class ChargePoint(ocpp.v201.ChargePoint): 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): @@ -764,7 +816,7 @@ class ChargePoint(ocpp.v201.ChargePoint): 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() @@ -779,22 +831,8 @@ async def on_connect( 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( @@ -810,6 +848,9 @@ async def on_connect( 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) @@ -824,47 +865,8 @@ async def on_connect( 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]: @@ -898,7 +900,7 @@ def _parse_variable_specs(specs_str: str, require_value: bool = False) -> list[d 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( @@ -983,27 +985,53 @@ async def main(): 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( @@ -1086,15 +1114,13 @@ async def main(): ) 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, @@ -1119,6 +1145,9 @@ async def main(): 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( @@ -1128,7 +1157,6 @@ async def main(): ) loop = asyncio.get_running_loop() - shutdown_count = 0 shutdown_event = asyncio.Event() async with websockets.serve( @@ -1141,53 +1169,15 @@ async def main(): 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) diff --git a/tests/ocpp-server/server16.py b/tests/ocpp-server/server16.py new file mode 100644 index 00000000..f0f87214 --- /dev/null +++ b/tests/ocpp-server/server16.py @@ -0,0 +1,805 @@ +"""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/) + 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) diff --git a/tests/ocpp-server/test_server.py b/tests/ocpp-server/test_server.py index 18a0d19a..ed96f830 100644 --- a/tests/ocpp-server/test_server.py +++ b/tests/ocpp-server/test_server.py @@ -4,6 +4,7 @@ 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 @@ -13,6 +14,7 @@ import pytest from ocpp.v201.enums import ( Action, AuthorizationStatusEnumType, + CancelReservationStatusEnumType, CertificateSignedStatusEnumType, ChangeAvailabilityStatusEnumType, ClearCacheStatusEnumType, @@ -33,6 +35,7 @@ from ocpp.v201.enums import ( RegistrationStatusEnumType, ReportBaseEnumType, RequestStartStopStatusEnumType, + ReserveNowStatusEnumType, ResetEnumType, ResetStatusEnumType, SendLocalListStatusEnumType, @@ -45,8 +48,11 @@ from ocpp.v201.enums import ( ) 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, @@ -54,6 +60,7 @@ from server import ( AuthMode, ChargePoint, ServerConfig, + _log_signed_meter_values, _parse_commands, _parse_variable_specs, _random_request_id, @@ -194,7 +201,9 @@ def main_mocks(): @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, @@ -217,7 +226,12 @@ def _patch_main(mock_loop, mock_server, mock_event, extra_patches=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) @@ -369,6 +383,8 @@ class TestHandlerCoverage: "_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) @@ -377,6 +393,14 @@ class TestHandlerCoverage: 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): @@ -446,7 +470,7 @@ class TestBootNotificationHandler: 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( @@ -558,7 +582,13 @@ class TestHeartbeatHandler: 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: @@ -572,6 +602,7 @@ class TestStatusNotificationHandler: connector_status="Available", ) assert isinstance(response, ocpp.v201.call_result.StatusNotification) + assert response == ocpp.v201.call_result.StatusNotification() class TestAuthorizeHandler: @@ -663,7 +694,8 @@ class TestRicherAuthorizeResponse: ) 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( @@ -730,7 +762,8 @@ class TestTransactionEventHandler: 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", @@ -758,8 +791,10 @@ class TestTransactionEventHandler: ) 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", @@ -786,8 +821,10 @@ class TestTransactionEventHandler: ) 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", @@ -825,6 +862,7 @@ class TestTransactionEventHandler: 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: @@ -909,6 +947,20 @@ class TestTransactionTracking: ) 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.""" @@ -948,8 +1000,10 @@ class TestNotificationHandlers: 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", @@ -970,6 +1024,7 @@ class TestNotificationHandlers: 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( @@ -978,24 +1033,28 @@ class TestNotificationHandlers: 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( @@ -1005,26 +1064,91 @@ class TestNotificationHandlers: 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 @@ -1043,13 +1167,27 @@ class TestSendCommandErrorHandling: 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 --- @@ -1324,7 +1462,30 @@ class TestOutgoingCommands: 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]]] = [ ( @@ -1407,6 +1568,16 @@ class TestOutgoingCommands: 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( @@ -1458,6 +1629,7 @@ class TestOnConnect: 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() @@ -1469,6 +1641,7 @@ class TestOnConnect: 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() @@ -1506,6 +1679,19 @@ class TestOnConnect: 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.""" @@ -1636,7 +1822,7 @@ class TestMainGracefulShutdown: 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() @@ -1659,7 +1845,7 @@ class TestMainGracefulShutdown: 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() @@ -1668,6 +1854,73 @@ class TestMainGracefulShutdown: 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.""" @@ -1797,6 +2050,14 @@ class TestCommandSequencing: 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.""" @@ -1832,6 +2093,22 @@ class TestMultiVariableCommands: ): _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 = [ { diff --git a/tests/ocpp-server/test_server16.py b/tests/ocpp-server/test_server16.py new file mode 100644 index 00000000..22a5785f --- /dev/null +++ b/tests/ocpp-server/test_server16.py @@ -0,0 +1,1357 @@ +"""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,)