diff --git a/.github/ISSUE_TEMPLATE/plc-test-result.yml b/.github/ISSUE_TEMPLATE/plc-test-result.yml new file mode 100644 index 00000000..2ac31ef6 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/plc-test-result.yml @@ -0,0 +1,78 @@ +name: Real PLC test result +description: Record one reproducible python-snap7 acceptance run against a real PLC +title: "[hardware test]: " +labels: ["hardware-test"] +body: + - type: markdown + attributes: + value: | + Attach the generated JSON and JUnit XML. Review all attachments before uploading. Never include PLC addresses, credentials, private keys, plant names, or unreviewed packet captures. + - type: input + id: source + attributes: + label: Source revision + description: Immutable 40-character commit SHA and, if applicable, release version. + placeholder: abcdef0123456789abcdef0123456789abcdef01 + validations: + required: true + - type: dropdown + id: result + attributes: + label: Overall result + options: [pass, fail, partial] + validations: + required: true + - type: dropdown + id: protocol + attributes: + label: Protocol path + options: [legacy-s7, s7commplus] + validations: + required: true + - type: input + id: plc + attributes: + label: PLC family, model, and order code + placeholder: S7-1500, CPU 1511-1 PN, 6ES7 ... + validations: + required: true + - type: input + id: firmware + attributes: + label: Firmware and security mode + placeholder: V2.9, PUT/GET enabled, non-optimized fixture DBs + validations: + required: true + - type: input + id: host + attributes: + label: Host environment + placeholder: Python 3.13, Linux x86_64 + validations: + required: true + - type: input + id: started + attributes: + label: Run time (UTC) + placeholder: 2026-09-11T12:34:56Z + validations: + required: true + - type: textarea + id: artifacts + attributes: + label: Sanitized report artifacts + description: Drag the generated JSON and JUnit XML here. State any manually reviewed capture separately. + validations: + required: true + - type: textarea + id: skips + attributes: + label: Skips, failures, and notes + description: Include machine-readable skip reasons and whether this result supersedes an earlier issue. + - type: checkboxes + id: safety + attributes: + label: Safety confirmation + options: + - label: I reviewed the attachments and they contain no PLC address, credentials, private keys, plant/site identifiers, or proprietary PLC program. + required: true diff --git a/CHANGES.md b/CHANGES.md index f664baeb..215efee1 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -6,6 +6,9 @@ CHANGES Major release: new `s7commplus` package with S7CommPlus protocol support. +* Add a safe, Gherkin-based real-PLC acceptance program with restored scratch + writes and sanitized JSON/JUnit reports (#840). + * Decode corroborating CPU execution attributes so S7CommPlus `get_cpu_state()` distinguishes RUN from STOP on S7-1500 and returns UNKNOWN for absent or inconsistent state attributes, including S7-1200 responses that omit them. diff --git a/README.rst b/README.rst index 233ceefe..00accb23 100644 --- a/README.rst +++ b/README.rst @@ -27,6 +27,8 @@ As of version 3.0, the C library is no longer used, but the name is kept for backwards compatibility. The full documentation is available on `Read The Docs `_. +The safe volunteer hardware-test procedure is in +`Real-PLC acceptance testing `_. Quick Start diff --git a/doc/index.rst b/doc/index.rst index 0c9b1f7a..4737c956 100644 --- a/doc/index.rst +++ b/doc/index.rst @@ -34,6 +34,7 @@ Welcome to python-snap7's documentation! :caption: Development development + real-plc-testing .. toctree:: :maxdepth: 2 diff --git a/doc/real-plc-testing.rst b/doc/real-plc-testing.rst new file mode 100644 index 00000000..b7fcacbf --- /dev/null +++ b/doc/real-plc-testing.rst @@ -0,0 +1,90 @@ +Real-PLC acceptance testing +=========================== + +The versioned Gherkin specifications in ``tests/features/real_plc`` exercise the +same client APIs and canonical 37-byte fixture used by the diagnostic end-to-end +tests. ``pytest-bdd`` was selected because it keeps pytest as the runner, reuses +fixtures, maps tags to markers, and supports the existing JUnit workflow. + +Safety model +------------ + +Use only a dedicated, non-production test PLC. The default runner selects +``@smoke`` and is read-only. It never enables writes or administrative actions. +Scratch writes require ``--allow-write``, a separate DB (DB2 by default), and are +wrapped in a guard that saves, restores, and reads back the original bytes from +fixture teardown even when a step fails or pytest handles an interruption. +Hard process termination or power loss cannot run cleanup, so inspect the scratch +DB before reuse after either event. + +Administrative actions (CPU start/stop, clock changes, program transfers, or +protection changes) require ``--allow-admin`` and must never target an in-service +PLC. No administrative scenarios are included in the initial suite. + +PLC preparation +--------------- + +Import ``tests/plc_setup/e2e_test_dbs.scl`` into TIA Portal, or recreate its exact +layout. It defines read-only DB1 and scratch DB2. On S7-1200/1500, disable +``Optimized block access`` for both fixture DBs. Enable PUT/GET only when the +chosen legacy protocol path and the PLC security policy require it. Defaults are +rack 0, slot 1, and TCP port 102. Apply the least privilege that permits the +selected scenarios; do not weaken a PLC that is in service. + +Running and reporting +--------------------- + +Install the test extras, then run the safe profile. Values supplied to reportable +metadata must not contain addresses, credentials, certificate paths, plant names, +or other site identifiers:: + + uv sync --extra test --extra s7commplus + uv run python tools/run_real_plc_acceptance.py \ + --plc-ip YOUR_PRIVATE_ADDRESS \ + --protocol legacy_s7 \ + --tester @YOUR_HANDLE \ + --plc-family S7-1500 \ + --plc-model "CPU 1511-1 PN" \ + --plc-firmware V2.9 \ + --plc-security-mode "PUT/GET, non-optimized fixture DBs" + +The runner prints a Gherkin-aware terminal summary and writes JUnit XML plus a +schema-versioned, sanitized JSON report beneath ``real-plc-results/``. The JSON +records exact source state, host environment, PLC metadata, scenario tags, +pass/fail/skip results, and bounded diagnostics. It deliberately never reads the +PLC address into the report. Review both artifacts before publishing them. + +Add ``--allow-write`` only after confirming DB2 is disposable scratch space. +``--allow-admin`` remains separate so write permission never implies permission +for operational actions. + +Result policy +------------- + +File one ``Real PLC test result`` issue per tester, PLC configuration, source +revision, and run. Attach both generated artifacts and apply exactly one result +label and one protocol label: + +* ``hardware-test`` +* ``test-result: pass``, ``test-result: fail``, or ``test-result: partial`` +* ``protocol: legacy-s7`` or ``protocol: s7commplus`` + +``pass`` means every selected applicable scenario passed. ``fail`` means at least +one selected scenario failed. ``partial`` means the run completed with capability +skips or an incomplete selection; unsupported capabilities are not failures. +Reruns get a new issue and link the earlier result. Close the earlier issue as +superseded only after the replacement artifacts exist. A result becomes stale +when the tested code, feature schema, relevant protocol implementation, PLC +firmware, or PLC configuration changes—not merely with age. + +Useful searches: + +* `Open hardware failures `_ +* `S7CommPlus hardware runs `_ +* Search a release or commit by adding its version or SHA to ``is:issue label:hardware-test``. + +For a release candidate, prioritize one supported legacy S7 PLC and one +S7CommPlus-capable S7-1200/1500 before expanding the host matrix; hosted CI already +covers supported Python versions and operating systems without hardware. Test +evidence stays in issues. Volunteer coordination may use Discussions, but is not +a substitute for an attached structured result. diff --git a/pyproject.toml b/pyproject.toml index 0e048e21..2f7b5b74 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ Homepage = "https://github.com/gijzelaerr/python-snap7" Documentation = "https://python-snap7.readthedocs.io/en/latest/" [project.optional-dependencies] -test = ["pytest", "pytest-asyncio", "pytest-cov", "pytest-html", "hypothesis", "mypy", "types-setuptools", "ruff", "tox", "tox-uv", "types-click", "uv"] +test = ["pytest", "pytest-asyncio", "pytest-bdd>=8.1,<9", "pytest-cov", "pytest-html", "hypothesis", "mypy", "types-setuptools", "ruff", "tox", "tox-uv", "types-click", "uv"] s7commplus = ["cryptography"] cli = ["rich", "click" ] demo = ["psutil", "rich", "click"] @@ -56,6 +56,13 @@ markers =[ "client", "common", "e2e: end-to-end tests requiring a real PLC connection", + "real_plc: versioned real-PLC acceptance scenario", + "smoke: read-only real-PLC acceptance scenario", + "write: real-PLC scenario that changes and restores scratch data", + "plc_write: legacy alias for real-PLC tests that modify scratch data", + "administrative: disruptive real-PLC scenario requiring separate opt-in", + "legacy_s7: scenario using the legacy S7 protocol path", + "s7commplus: scenario using the S7CommPlus protocol path", "hypothesis: property-based tests using Hypothesis", "logo", "mainloop", diff --git a/tests/conftest.py b/tests/conftest.py index 28490345..3e0e677b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,9 +2,15 @@ import socket import sys +from pathlib import Path +from typing import Any import pytest +from tests.real_plc.reporting import RealPLCReport, report_metadata + +_REAL_PLC_REPORT = RealPLCReport() + def get_free_tcp_port() -> int: """Return a TCP port that is free *right now* on 127.0.0.1. @@ -76,6 +82,27 @@ def pytest_addoption(parser: pytest.Parser) -> None: default=2, help="Read-write DB number for e2e tests (default: 2)", ) + parser.addoption( + "--allow-plc-write", + action="store_true", + default=False, + help="Allow tests that modify and restore the dedicated scratch DB", + ) + parser.addoption( + "--allow-plc-admin", + action="store_true", + default=False, + help="Allow disruptive administrative tests on a dedicated non-production PLC", + ) + parser.addoption("--plc-protocol", choices=("legacy_s7", "s7commplus"), default="legacy_s7") + parser.addoption("--plc-report-json", default="", help="Write a sanitized real-PLC JSON report") + parser.addoption("--tester", default="", help="GitHub handle of the volunteer running the test") + parser.addoption("--plc-family", default="", help="Reportable PLC family, such as S7-1500") + parser.addoption("--plc-model", default="", help="Reportable PLC model") + parser.addoption("--plc-order-code", default="", help="Reportable PLC order code") + parser.addoption("--plc-firmware", default="", help="Reportable PLC firmware version") + parser.addoption("--plc-security-mode", default="", help="Reportable access/TLS mode; never enter credentials") + parser.addoption("--plc-tia-configuration", default="", help="Reportable TIA settings, without site details") def pytest_configure(config: pytest.Config) -> None: @@ -84,6 +111,9 @@ def pytest_configure(config: pytest.Config) -> None: "markers", "e2e: mark test as end-to-end test requiring real PLC connection", ) + global _REAL_PLC_REPORT + _REAL_PLC_REPORT = RealPLCReport() + config._real_plc_report = _REAL_PLC_REPORT def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None: @@ -104,14 +134,26 @@ def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item e2e.DB_READ_ONLY = int(config.getoption("--plc-db-read")) e2e.DB_READ_WRITE = int(config.getoption("--plc-db-write")) - # Skip e2e tests if flag not provided - if config.getoption("--e2e"): - return - - skip_e2e = pytest.mark.skip(reason="Need --e2e option to run end-to-end tests") for item in items: - if "e2e" in item.keywords: - item.add_marker(skip_e2e) + if "e2e" in item.keywords and not config.getoption("--e2e"): + item.add_marker(pytest.mark.skip(reason="Need --e2e option to run end-to-end tests")) + if ("write" in item.keywords or "plc_write" in item.keywords) and not config.getoption("--allow-plc-write"): + item.add_marker(pytest.mark.skip(reason="WRITE_OPT_IN_REQUIRED: pass --allow-plc-write")) + if "administrative" in item.keywords and not config.getoption("--allow-plc-admin"): + item.add_marker(pytest.mark.skip(reason="ADMIN_OPT_IN_REQUIRED: pass --allow-plc-admin")) + + +def pytest_runtest_logreport(report: Any) -> None: + """Capture final BDD outcomes without recording connection secrets.""" + _REAL_PLC_REPORT.record(report) + + +def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: + """Write the optional structured report after all scenario outcomes are known.""" + del exitstatus + report_path = session.config.getoption("--plc-report-json") + if report_path: + session.config._real_plc_report.write(Path(report_path), report_metadata(session.config)) @pytest.fixture(scope="session") diff --git a/tests/features/real_plc/connection.feature b/tests/features/real_plc/connection.feature new file mode 100644 index 00000000..d189a1e9 --- /dev/null +++ b/tests/features/real_plc/connection.feature @@ -0,0 +1,26 @@ +@real_plc +Feature: Connect to and identify a real PLC + + Background: + Given I am using a dedicated test PLC + And the test configuration contains no secrets in reportable fields + + @smoke + Scenario Outline: Establish and close a session + Given the client uses the "" protocol path + When I connect to the configured PLC + Then the client reports that it is connected + And the negotiated protocol and security mode are recorded + And the PLC identity and CPU state are recorded when available + When I disconnect + Then the client reports that it is disconnected + + @legacy_s7 + Examples: legacy S7 + | protocol | + | legacy_s7 | + + @s7commplus + Examples: S7CommPlus + | protocol | + | s7commplus | diff --git a/tests/features/real_plc/lifecycle.feature b/tests/features/real_plc/lifecycle.feature new file mode 100644 index 00000000..b9149d7b --- /dev/null +++ b/tests/features/real_plc/lifecycle.feature @@ -0,0 +1,37 @@ +@real_plc +Feature: Recover from normal connection lifecycle events + + @smoke + Scenario Outline: Reconnect after a clean disconnect + Given the client uses the "" protocol path + And I connected to and disconnected from the PLC + When I reconnect with the same configuration + Then a known read succeeds + + @legacy_s7 + Examples: legacy S7 + | protocol | + | legacy_s7 | + + @s7commplus + Examples: S7CommPlus + | protocol | + | s7commplus | + + @smoke + Scenario Outline: Repeated operations do not corrupt the session + Given the client uses the "" protocol path + And I am connected to the PLC + When I read the canonical fixture repeatedly + Then every read succeeds with the expected value + And disconnect completes cleanly + + @legacy_s7 + Examples: legacy S7 + | protocol | + | legacy_s7 | + + @s7commplus + Examples: S7CommPlus + | protocol | + | s7commplus | diff --git a/tests/features/real_plc/read.feature b/tests/features/real_plc/read.feature new file mode 100644 index 00000000..d937033b --- /dev/null +++ b/tests/features/real_plc/read.feature @@ -0,0 +1,37 @@ +@real_plc +Feature: Read a known test data block + + @smoke + Scenario Outline: Read the canonical fixture + Given the client uses the "" protocol path + And the read-only test DB has the documented canonical layout + When I read the complete fixture DB + Then INT, REAL, BYTE, WORD, DWORD, DINT, CHAR and BOOL values match the fixture + And individual reads return the same values as the complete-block read + + @legacy_s7 + Examples: legacy S7 + | protocol | + | legacy_s7 | + + @s7commplus + Examples: S7CommPlus + | protocol | + | s7commplus | + + @smoke + Scenario Outline: Read multiple values in one request + Given the client uses the "" protocol path + And the read-only test DB has the documented canonical layout + When I read values of different sizes in one multi-variable request + Then every value and result code matches the fixture + + @legacy_s7 + Examples: legacy S7 + | protocol | + | legacy_s7 | + + @s7commplus + Examples: S7CommPlus + | protocol | + | s7commplus | diff --git a/tests/features/real_plc/report.schema.json b/tests/features/real_plc/report.schema.json new file mode 100644 index 00000000..14613373 --- /dev/null +++ b/tests/features/real_plc/report.schema.json @@ -0,0 +1,32 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/gijzelaerr/python-snap7/blob/master/tests/features/real_plc/report.schema.json", + "title": "python-snap7 real-PLC acceptance result", + "type": "object", + "additionalProperties": false, + "required": ["schema_version", "started_at", "ended_at", "overall_result", "source", "environment", "tester", "plc", "scenarios"], + "properties": { + "schema_version": {"const": "1.0"}, + "started_at": {"type": "string", "format": "date-time"}, + "ended_at": {"type": "string", "format": "date-time"}, + "overall_result": {"enum": ["pass", "fail", "partial"]}, + "source": {"type": "object"}, + "environment": {"type": "object"}, + "tester": {"type": "string"}, + "plc": {"type": "object"}, + "scenarios": { + "type": "array", + "items": { + "type": "object", + "required": ["scenario_id", "status", "tags", "duration_seconds", "diagnostic"], + "properties": { + "scenario_id": {"type": "string"}, + "status": {"enum": ["passed", "failed", "skipped"]}, + "tags": {"type": "array", "items": {"type": "string"}}, + "duration_seconds": {"type": "number", "minimum": 0}, + "diagnostic": {"type": ["string", "null"]} + } + } + } + } +} diff --git a/tests/features/real_plc/write.feature b/tests/features/real_plc/write.feature new file mode 100644 index 00000000..c3325ca8 --- /dev/null +++ b/tests/features/real_plc/write.feature @@ -0,0 +1,35 @@ +@real_plc @write +Feature: Write safely to a scratch data block + + Scenario Outline: Round-trip a value and restore the PLC + Given the client uses the "" protocol path + And writing has been explicitly enabled + And I have a dedicated scratch DB + And the original bytes for have been saved + When I write a valid value + Then reading the address returns the written value + And the original bytes are restored and verified + + @legacy_s7 + Examples: legacy S7 + | protocol | type | + | legacy_s7 | INT | + | legacy_s7 | REAL | + | legacy_s7 | BYTE | + | legacy_s7 | WORD | + | legacy_s7 | DWORD | + | legacy_s7 | DINT | + | legacy_s7 | CHAR | + | legacy_s7 | BOOL | + + @s7commplus + Examples: S7CommPlus + | protocol | type | + | s7commplus | INT | + | s7commplus | REAL | + | s7commplus | BYTE | + | s7commplus | WORD | + | s7commplus | DWORD | + | s7commplus | DINT | + | s7commplus | CHAR | + | s7commplus | BOOL | diff --git a/tests/real_plc/__init__.py b/tests/real_plc/__init__.py new file mode 100644 index 00000000..b60fd2d5 --- /dev/null +++ b/tests/real_plc/__init__.py @@ -0,0 +1 @@ +"""Shared real-PLC acceptance-test support.""" diff --git a/tests/real_plc/reporting.py b/tests/real_plc/reporting.py new file mode 100644 index 00000000..8263a663 --- /dev/null +++ b/tests/real_plc/reporting.py @@ -0,0 +1,115 @@ +"""Sanitized, schema-versioned report generation for real-PLC runs.""" + +from __future__ import annotations + +import importlib.metadata +import json +import platform +import re +import subprocess +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +SCHEMA_VERSION = "1.0" +_IPV4 = re.compile(r"(? str: + """Remove common connection/credential material and cap report size.""" + text = _IPV4.sub("", str(value)) + text = _SENSITIVE_ASSIGNMENT.sub(lambda match: f"{match.group(1)}=", text) + return text[:limit] + + +def _git_source() -> dict[str, object]: + try: + commit = subprocess.run(["git", "rev-parse", "HEAD"], check=True, capture_output=True, text=True).stdout.strip() + dirty = bool(subprocess.run(["git", "status", "--porcelain"], check=True, capture_output=True, text=True).stdout) + return {"kind": "git", "commit": commit, "dirty": dirty} + except (OSError, subprocess.CalledProcessError): + return {"kind": "installed", "version": importlib.metadata.version("python-snap7")} + + +@dataclass +class ScenarioResult: + scenario_id: str + status: str + tags: list[str] + duration_seconds: float + diagnostic: str | None = None + + +@dataclass +class RealPLCReport: + """Accumulate final pytest outcomes and emit the public-safe JSON format.""" + + started_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + scenarios: dict[str, ScenarioResult] = field(default_factory=dict) + runtime_metadata: dict[str, str | int] = field(default_factory=dict) + + def record(self, report: Any) -> None: + nodeid = str(report.nodeid).replace("\\", "/") + if "real_plc/test_acceptance.py::" not in nodeid: + return + if report.when == "call": + status = "passed" if report.passed else "skipped" if report.skipped else "failed" + elif report.when == "setup" and report.skipped: + status = "skipped" + else: + return + tags = sorted( + tag for tag in ("smoke", "write", "administrative", "legacy_s7", "s7commplus", "real_plc") if tag in report.keywords + ) + diagnostic = None if report.passed else sanitize_diagnostic(report.longrepr) + self.scenarios[nodeid] = ScenarioResult(nodeid, status, tags, report.duration, diagnostic) + + def payload(self, supplied_metadata: dict[str, str | int]) -> dict[str, object]: + scenarios = [vars(result) for result in self.scenarios.values()] + statuses = {result["status"] for result in scenarios} + overall = "fail" if "failed" in statuses else "partial" if not scenarios or "skipped" in statuses else "pass" + try: + snap7_version = importlib.metadata.version("python-snap7") + except importlib.metadata.PackageNotFoundError: + snap7_version = "source-tree" + return { + "schema_version": SCHEMA_VERSION, + "started_at": self.started_at.isoformat(), + "ended_at": datetime.now(timezone.utc).isoformat(), + "overall_result": overall, + "source": _git_source(), + "environment": { + "python": platform.python_version(), + "implementation": platform.python_implementation(), + "os": platform.system(), + "os_release": platform.release(), + "architecture": platform.machine(), + "python_snap7": snap7_version, + "pytest_bdd": importlib.metadata.version("pytest-bdd"), + }, + "tester": supplied_metadata.pop("tester", ""), + "plc": {**supplied_metadata, **self.runtime_metadata}, + "scenarios": scenarios, + } + + def write(self, path: Path, supplied_metadata: dict[str, str | int]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(self.payload(dict(supplied_metadata)), indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +def report_metadata(config: Any) -> dict[str, str | int]: + """Return only explicitly public-safe CLI metadata (never network details).""" + return { + "tester": config.getoption("--tester"), + "family": config.getoption("--plc-family"), + "model": config.getoption("--plc-model"), + "order_code": config.getoption("--plc-order-code"), + "firmware": config.getoption("--plc-firmware"), + "rack": config.getoption("--plc-rack"), + "slot": config.getoption("--plc-slot"), + "protocol_path": config.getoption("--plc-protocol"), + "security_mode": config.getoption("--plc-security-mode"), + "tia_configuration": config.getoption("--plc-tia-configuration"), + } diff --git a/tests/real_plc/support.py b/tests/real_plc/support.py new file mode 100644 index 00000000..78834a07 --- /dev/null +++ b/tests/real_plc/support.py @@ -0,0 +1,266 @@ +"""Protocol-neutral helpers for safe real-PLC acceptance scenarios.""" + +from __future__ import annotations + +import math +import struct +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from typing import Protocol + +from s7commplus.client import S7CommPlusClient +from snap7.client import Client +from snap7.type import Area, S7DataItem, WordLen +from snap7.util import get_bool, get_byte, get_char, get_dint, get_dword, get_int, get_real, get_word + +DB_SIZE = 37 +OFFSET_INT1 = 0 +OFFSET_INT2 = 2 +OFFSET_FLOAT1 = 4 +OFFSET_FLOAT2 = 8 +OFFSET_BYTE1 = 12 +OFFSET_BYTE2 = 13 +OFFSET_WORD1 = 14 +OFFSET_WORD2 = 16 +OFFSET_DWORD1 = 18 +OFFSET_DWORD2 = 22 +OFFSET_DINT1 = 26 +OFFSET_DINT2 = 30 +OFFSET_CHAR1 = 34 +OFFSET_CHAR2 = 35 +OFFSET_BOOLS = 36 + +EXPECTED_INT1 = 10 +EXPECTED_INT2 = 255 +EXPECTED_FLOAT1 = 123.45 +EXPECTED_FLOAT2 = 543.21 +EXPECTED_BYTE1 = 0x0F +EXPECTED_BYTE2 = 0xF0 +EXPECTED_WORD1 = 0xABCD +EXPECTED_WORD2 = 0x1234 +EXPECTED_DWORD1 = 0x12345678 +EXPECTED_DWORD2 = 0x89ABCDEF +EXPECTED_DINT1 = 2147483647 +EXPECTED_DINT2 = 42 +EXPECTED_CHAR1 = "F" +EXPECTED_CHAR2 = "-" +EXPECTED_BOOLS = (True, False, False, False, False, False, False, False) +( + EXPECTED_BOOL0, + EXPECTED_BOOL1, + EXPECTED_BOOL2, + EXPECTED_BOOL3, + EXPECTED_BOOL4, + EXPECTED_BOOL5, + EXPECTED_BOOL6, + EXPECTED_BOOL7, +) = EXPECTED_BOOLS + + +@dataclass(frozen=True) +class PLCConfig: + """Connection settings; ``host`` is deliberately never reportable.""" + + host: str + port: int + rack: int + slot: int + read_db: int + write_db: int + + +class PLCAdapter(Protocol): + """Small common surface used by both Gherkin and diagnostic pytest tests.""" + + protocol_name: str + + def connect(self) -> None: ... + + def disconnect(self) -> None: ... + + def is_connected(self) -> bool: ... + + def read(self, db_number: int, offset: int, size: int) -> bytes: ... + + def write(self, db_number: int, offset: int, data: bytes) -> None: ... + + def read_multi(self, db_number: int, regions: Sequence[tuple[int, int]]) -> list[bytes]: ... + + def runtime_metadata(self) -> dict[str, str | int]: ... + + +class LegacyS7Adapter: + protocol_name = "legacy_s7" + + def __init__(self, config: PLCConfig) -> None: + self.config = config + self.client = Client() + + def connect(self) -> None: + self.client.connect(self.config.host, self.config.rack, self.config.slot, self.config.port) + + def disconnect(self) -> None: + self.client.disconnect() + + def is_connected(self) -> bool: + return self.client.get_connected() + + def read(self, db_number: int, offset: int, size: int) -> bytes: + return bytes(self.client.db_read(db_number, offset, size)) + + def write(self, db_number: int, offset: int, data: bytes) -> None: + self.client.db_write(db_number, offset, bytearray(data)) + + def read_multi(self, db_number: int, regions: Sequence[tuple[int, int]]) -> list[bytes]: + from ctypes import POINTER, c_int32, c_uint8, cast, create_string_buffer, pointer + + items = (S7DataItem * len(regions))() + buffers = [] + for item, (offset, size) in zip(items, regions): + item.Area = c_int32(Area.DB.value) + item.WordLen = c_int32(WordLen.Byte.value) + item.Result = c_int32(0) + item.DBNumber = c_int32(db_number) + item.Start = c_int32(offset) + item.Amount = c_int32(size) + buffer = create_string_buffer(size) + buffers.append(buffer) + item.pData = cast(pointer(buffer), POINTER(c_uint8)) + result, returned = self.client.read_multi_vars(items) + if result != 0: + raise RuntimeError(f"legacy multi-read failed with result {result}") + return [bytes(item.pData[:size]) for item, (_, size) in zip(returned, regions)] + + def runtime_metadata(self) -> dict[str, str | int]: + metadata: dict[str, str | int] = {"protocol_path": self.protocol_name} + try: + order_code = self.client.get_order_code() + metadata["order_code"] = order_code.Code.rstrip(b"\x00").decode("ascii", errors="replace") + metadata["firmware"] = f"{order_code.V1}.{order_code.V2}.{order_code.V3}" + except Exception as error: # noqa: BLE001 - optional metadata differs by PLC family + metadata["order_code_status"] = f"unavailable:{type(error).__name__}" + try: + metadata["cpu_state"] = self.client.get_cpu_state() + except Exception as error: # noqa: BLE001 - optional metadata differs by PLC family + metadata["cpu_state_status"] = f"unavailable:{type(error).__name__}" + return metadata + + +class S7CommPlusAdapter: + protocol_name = "s7commplus" + + def __init__(self, config: PLCConfig) -> None: + self.config = config + self.client = S7CommPlusClient() + + def connect(self) -> None: + self.client.connect(self.config.host, self.config.port, self.config.rack, self.config.slot) + + def disconnect(self) -> None: + self.client.disconnect() + + def is_connected(self) -> bool: + return self.client.connected + + def read(self, db_number: int, offset: int, size: int) -> bytes: + return self.client.db_read(db_number, offset, size) + + def write(self, db_number: int, offset: int, data: bytes) -> None: + self.client.db_write(db_number, offset, data) + + def read_multi(self, db_number: int, regions: Sequence[tuple[int, int]]) -> list[bytes]: + return self.client.db_read_multi([(db_number, offset, size) for offset, size in regions]) + + def runtime_metadata(self) -> dict[str, str | int]: + return { + "protocol_path": self.protocol_name, + "protocol_version": self.client.protocol_version, + "security_mode": "tls" if getattr(self.client._connection, "_tls_active", False) else "plain", + } + + +def make_adapter(protocol_name: str, config: PLCConfig) -> PLCAdapter: + """Create the selected client behind the shared acceptance-test surface.""" + if protocol_name == "legacy_s7": + return LegacyS7Adapter(config) + if protocol_name == "s7commplus": + return S7CommPlusAdapter(config) + raise ValueError(f"Unsupported PLC protocol path: {protocol_name}") + + +def canonical_fixture_bytes() -> bytes: + """Return the canonical 37-byte DB image documented in ``plc_setup``.""" + return b"".join( + ( + struct.pack( + ">hhffBBHHIIii", + EXPECTED_INT1, + EXPECTED_INT2, + EXPECTED_FLOAT1, + EXPECTED_FLOAT2, + EXPECTED_BYTE1, + EXPECTED_BYTE2, + EXPECTED_WORD1, + EXPECTED_WORD2, + EXPECTED_DWORD1, + EXPECTED_DWORD2, + EXPECTED_DINT1, + EXPECTED_DINT2, + ), + EXPECTED_CHAR1.encode() + EXPECTED_CHAR2.encode() + bytes([1]), + ) + ) + + +def assert_canonical_fixture(data: bytes) -> None: + """Validate every documented scalar while keeping useful assertion names.""" + data = bytes(data) + mutable = bytearray(data) + assert len(data) == DB_SIZE + assert get_int(mutable, OFFSET_INT1) == EXPECTED_INT1 + assert get_int(mutable, OFFSET_INT2) == EXPECTED_INT2 + assert math.isclose(get_real(mutable, OFFSET_FLOAT1), EXPECTED_FLOAT1, abs_tol=0.001) + assert math.isclose(get_real(mutable, OFFSET_FLOAT2), EXPECTED_FLOAT2, abs_tol=0.001) + assert get_byte(mutable, OFFSET_BYTE1) == EXPECTED_BYTE1 + assert get_byte(mutable, OFFSET_BYTE2) == EXPECTED_BYTE2 + assert get_word(mutable, OFFSET_WORD1) == EXPECTED_WORD1 + assert get_word(mutable, OFFSET_WORD2) == EXPECTED_WORD2 + assert get_dword(mutable, OFFSET_DWORD1) == EXPECTED_DWORD1 + assert get_dword(mutable, OFFSET_DWORD2) == EXPECTED_DWORD2 + assert get_dint(mutable, OFFSET_DINT1) == EXPECTED_DINT1 + assert get_dint(mutable, OFFSET_DINT2) == EXPECTED_DINT2 + assert get_char(mutable, OFFSET_CHAR1) == EXPECTED_CHAR1 + assert get_char(mutable, OFFSET_CHAR2) == EXPECTED_CHAR2 + assert tuple(get_bool(mutable, OFFSET_BOOLS, bit) for bit in range(8)) == EXPECTED_BOOLS + + +WRITE_VALUES: dict[str, tuple[int, bytes, Callable[[bytes], object], object]] = { + "INT": (0, struct.pack(">h", -1234), lambda data: get_int(bytearray(data), 0), -1234), + "REAL": (4, struct.pack(">f", 456.75), lambda data: get_real(bytearray(data), 0), 456.75), + "BYTE": (12, b"\xa5", lambda data: get_byte(bytearray(data), 0), 0xA5), + "WORD": (14, struct.pack(">H", 0x5AA5), lambda data: get_word(bytearray(data), 0), 0x5AA5), + "DWORD": (18, struct.pack(">I", 0xDEADBEEF), lambda data: get_dword(bytearray(data), 0), 0xDEADBEEF), + "DINT": (26, struct.pack(">i", -123456789), lambda data: get_dint(bytearray(data), 0), -123456789), + "CHAR": (34, b"X", lambda data: get_char(bytearray(data), 0), "X"), + "BOOL": (36, b"\x81", lambda data: get_bool(bytearray(data), 0, 7), True), +} + + +class ScratchRestoreGuard: + """Save, restore, and verify one scratch region; restoration is idempotent.""" + + def __init__(self, adapter: PLCAdapter, db_number: int, offset: int, size: int) -> None: + self.adapter = adapter + self.db_number = db_number + self.offset = offset + self.original = adapter.read(db_number, offset, size) + self.restored = False + + def restore(self) -> None: + if self.restored: + return + self.adapter.write(self.db_number, self.offset, self.original) + restored = self.adapter.read(self.db_number, self.offset, len(self.original)) + if restored != self.original: + raise AssertionError("scratch DB restoration verification failed") + self.restored = True diff --git a/tests/real_plc/test_acceptance.py b/tests/real_plc/test_acceptance.py new file mode 100644 index 00000000..39f128a4 --- /dev/null +++ b/tests/real_plc/test_acceptance.py @@ -0,0 +1,221 @@ +"""Executable pytest-bdd bindings for the versioned real-PLC specification.""" + +from __future__ import annotations + +from typing import Any + +import pytest +from pytest_bdd import given, parsers, scenarios, then, when + +from tests.real_plc.support import ( + DB_SIZE, + WRITE_VALUES, + PLCAdapter, + PLCConfig, + ScratchRestoreGuard, + assert_canonical_fixture, + canonical_fixture_bytes, + make_adapter, +) + +pytestmark = [pytest.mark.e2e, pytest.mark.real_plc] +scenarios("../features/real_plc") + + +@pytest.fixture +def plc_config(request: pytest.FixtureRequest) -> PLCConfig: + return PLCConfig( + host=request.config.getoption("--plc-ip"), + port=request.config.getoption("--plc-port"), + rack=request.config.getoption("--plc-rack"), + slot=request.config.getoption("--plc-slot"), + read_db=request.config.getoption("--plc-db-read"), + write_db=request.config.getoption("--plc-db-write"), + ) + + +@pytest.fixture +def scenario_state() -> dict[str, Any]: + return {} + + +@given("I am using a dedicated test PLC") +def dedicated_plc() -> None: + """The runbook makes this an explicit operator precondition.""" + + +@given("the test configuration contains no secrets in reportable fields") +def reportable_fields_are_safe(request: pytest.FixtureRequest) -> None: + sensitive_fragments = ("password", "secret", "private key", "plc ip", "hostname") + values = ( + request.config.getoption("--plc-family"), + request.config.getoption("--plc-model"), + request.config.getoption("--plc-order-code"), + request.config.getoption("--plc-firmware"), + request.config.getoption("--plc-security-mode"), + request.config.getoption("--plc-tia-configuration"), + ) + assert not any(fragment in value.lower() for value in values for fragment in sensitive_fragments) + + +@given(parsers.parse('the client uses the "{protocol}" protocol path'), target_fixture="plc_adapter") +def selected_adapter(protocol: str, plc_config: PLCConfig, request: pytest.FixtureRequest) -> PLCAdapter: + selected = request.config.getoption("--plc-protocol") + if protocol != selected: + pytest.skip(f"CAPABILITY_PROTOCOL_NOT_SELECTED: selected {selected}") + adapter = make_adapter(protocol, plc_config) + + def disconnect() -> None: + if adapter.is_connected(): + adapter.disconnect() + + request.addfinalizer(disconnect) + return adapter + + +def _ensure_connected(adapter: PLCAdapter) -> None: + if not adapter.is_connected(): + adapter.connect() + + +@when("I connect to the configured PLC") +@given("I am connected to the PLC") +def connect(plc_adapter: PLCAdapter) -> None: + _ensure_connected(plc_adapter) + + +@then("the client reports that it is connected") +def reports_connected(plc_adapter: PLCAdapter) -> None: + assert plc_adapter.is_connected() + + +@then("the negotiated protocol and security mode are recorded") +@then("the PLC identity and CPU state are recorded when available") +def record_runtime_metadata(plc_adapter: PLCAdapter, request: pytest.FixtureRequest) -> None: + request.config._real_plc_report.runtime_metadata.update(plc_adapter.runtime_metadata()) + + +@when("I disconnect") +def disconnect(plc_adapter: PLCAdapter) -> None: + plc_adapter.disconnect() + + +@then("the client reports that it is disconnected") +def reports_disconnected(plc_adapter: PLCAdapter) -> None: + assert not plc_adapter.is_connected() + + +@given("the read-only test DB has the documented canonical layout") +def canonical_layout(plc_adapter: PLCAdapter, plc_config: PLCConfig) -> None: + _ensure_connected(plc_adapter) + assert_canonical_fixture(plc_adapter.read(plc_config.read_db, 0, DB_SIZE)) + + +@when("I read the complete fixture DB") +def read_complete_fixture(plc_adapter: PLCAdapter, plc_config: PLCConfig, scenario_state: dict[str, Any]) -> None: + scenario_state["complete"] = plc_adapter.read(plc_config.read_db, 0, DB_SIZE) + + +@then("INT, REAL, BYTE, WORD, DWORD, DINT, CHAR and BOOL values match the fixture") +def complete_values_match(scenario_state: dict[str, Any]) -> None: + assert_canonical_fixture(scenario_state["complete"]) + + +@then("individual reads return the same values as the complete-block read") +def individual_values_match(plc_adapter: PLCAdapter, plc_config: PLCConfig, scenario_state: dict[str, Any]) -> None: + complete = scenario_state["complete"] + for offset, size in ((0, 2), (4, 4), (12, 1), (14, 2), (18, 4), (26, 4), (34, 1), (36, 1)): + assert plc_adapter.read(plc_config.read_db, offset, size) == complete[offset : offset + size] + + +@when("I read values of different sizes in one multi-variable request") +def read_multiple(plc_adapter: PLCAdapter, plc_config: PLCConfig, scenario_state: dict[str, Any]) -> None: + regions = ((0, 2), (4, 4), (12, 1), (18, 4), (34, 1), (36, 1)) + scenario_state["multi_regions"] = regions + scenario_state["multi_values"] = plc_adapter.read_multi(plc_config.read_db, regions) + + +@then("every value and result code matches the fixture") +def multiple_values_match(scenario_state: dict[str, Any]) -> None: + expected = canonical_fixture_bytes() + assert scenario_state["multi_values"] == [ + expected[offset : offset + size] for offset, size in scenario_state["multi_regions"] + ] + + +@given("writing has been explicitly enabled") +def write_is_enabled(request: pytest.FixtureRequest) -> None: + assert request.config.getoption("--allow-plc-write") + + +@given("I have a dedicated scratch DB") +def scratch_db(plc_adapter: PLCAdapter, plc_config: PLCConfig) -> None: + _ensure_connected(plc_adapter) + if plc_config.read_db == plc_config.write_db: + pytest.fail("scratch DB must differ from the read-only fixture DB") + + +@given(parsers.parse("the original bytes for {value_type} have been saved")) +def save_original( + value_type: str, + plc_adapter: PLCAdapter, + plc_config: PLCConfig, + scenario_state: dict[str, Any], + request: pytest.FixtureRequest, +) -> None: + offset, encoded, _, _ = WRITE_VALUES[value_type] + guard = ScratchRestoreGuard(plc_adapter, plc_config.write_db, offset, len(encoded)) + scenario_state["guard"] = guard + scenario_state["value_type"] = value_type + request.addfinalizer(guard.restore) + + +@when(parsers.parse("I write a valid {value_type} value")) +def write_value(value_type: str, plc_adapter: PLCAdapter, plc_config: PLCConfig) -> None: + offset, encoded, _, _ = WRITE_VALUES[value_type] + plc_adapter.write(plc_config.write_db, offset, encoded) + + +@then("reading the address returns the written value") +def value_round_trips(plc_adapter: PLCAdapter, plc_config: PLCConfig, scenario_state: dict[str, Any]) -> None: + offset, encoded, decoder, expected = WRITE_VALUES[scenario_state["value_type"]] + actual = plc_adapter.read(plc_config.write_db, offset, len(encoded)) + assert decoder(actual) == expected + + +@then("the original bytes are restored and verified") +def original_is_restored(scenario_state: dict[str, Any]) -> None: + scenario_state["guard"].restore() + + +@given("I connected to and disconnected from the PLC") +def connected_then_disconnected(plc_adapter: PLCAdapter) -> None: + plc_adapter.connect() + plc_adapter.disconnect() + + +@when("I reconnect with the same configuration") +def reconnect(plc_adapter: PLCAdapter) -> None: + plc_adapter.connect() + + +@then("a known read succeeds") +def known_read_succeeds(plc_adapter: PLCAdapter, plc_config: PLCConfig) -> None: + assert plc_adapter.read(plc_config.read_db, 0, 2) == canonical_fixture_bytes()[:2] + + +@when("I read the canonical fixture repeatedly") +def read_repeatedly(plc_adapter: PLCAdapter, plc_config: PLCConfig, scenario_state: dict[str, Any]) -> None: + scenario_state["repeated"] = [plc_adapter.read(plc_config.read_db, 0, DB_SIZE) for _ in range(10)] + + +@then("every read succeeds with the expected value") +def repeated_reads_match(scenario_state: dict[str, Any]) -> None: + for data in scenario_state["repeated"]: + assert_canonical_fixture(data) + + +@then("disconnect completes cleanly") +def final_disconnect(plc_adapter: PLCAdapter) -> None: + plc_adapter.disconnect() + assert not plc_adapter.is_connected() diff --git a/tests/test_client_e2e.py b/tests/test_client_e2e.py index 947fdc95..4159cc25 100644 --- a/tests/test_client_e2e.py +++ b/tests/test_client_e2e.py @@ -36,30 +36,71 @@ """ import os -import pytest import unittest -from ctypes import c_int32, POINTER, pointer, create_string_buffer, cast, c_uint8 +from ctypes import POINTER, c_int32, c_uint8, cast, create_string_buffer, pointer from datetime import datetime +import pytest + from snap7.client import Client -from snap7.type import Area, Block, S7DataItem, WordLen, Parameter +from snap7.type import Area, Block, Parameter, S7DataItem, WordLen from snap7.util import ( + get_bool, + get_byte, + get_char, + get_dint, + get_dword, get_int, get_real, - get_byte, get_word, - get_dword, - get_dint, - get_char, - get_bool, + set_bool, + set_byte, + set_char, + set_dint, + set_dword, set_int, set_real, - set_byte, set_word, - set_dword, - set_dint, - set_char, - set_bool, +) +from tests.real_plc.support import ( + DB_SIZE, + EXPECTED_BOOL0, + EXPECTED_BOOL1, + EXPECTED_BOOL2, + EXPECTED_BOOL3, + EXPECTED_BOOL4, + EXPECTED_BOOL5, + EXPECTED_BOOL6, + EXPECTED_BOOL7, + EXPECTED_BYTE1, + EXPECTED_BYTE2, + EXPECTED_CHAR1, + EXPECTED_CHAR2, + EXPECTED_DINT1, + EXPECTED_DINT2, + EXPECTED_DWORD1, + EXPECTED_DWORD2, + EXPECTED_FLOAT1, + EXPECTED_FLOAT2, + EXPECTED_INT1, + EXPECTED_INT2, + EXPECTED_WORD1, + EXPECTED_WORD2, + OFFSET_BOOLS, + OFFSET_BYTE1, + OFFSET_BYTE2, + OFFSET_CHAR1, + OFFSET_CHAR2, + OFFSET_DINT1, + OFFSET_DINT2, + OFFSET_DWORD1, + OFFSET_DWORD2, + OFFSET_FLOAT1, + OFFSET_FLOAT2, + OFFSET_INT1, + OFFSET_INT2, + OFFSET_WORD1, + OFFSET_WORD2, ) # ============================================================================= @@ -76,55 +117,6 @@ DB_READ_WRITE = int(os.environ.get("PLC_DB_WRITE", "2")) -# ============================================================================= -# DB Structure - Byte offsets for each variable -# ============================================================================= -OFFSET_INT1 = 0 # Int (2 bytes) -OFFSET_INT2 = 2 # Int (2 bytes) -OFFSET_FLOAT1 = 4 # Real (4 bytes) -OFFSET_FLOAT2 = 8 # Real (4 bytes) -OFFSET_BYTE1 = 12 # Byte (1 byte) -OFFSET_BYTE2 = 13 # Byte (1 byte) -OFFSET_WORD1 = 14 # Word (2 bytes) -OFFSET_WORD2 = 16 # Word (2 bytes) -OFFSET_DWORD1 = 18 # DWord (4 bytes) -OFFSET_DWORD2 = 22 # DWord (4 bytes) -OFFSET_DINT1 = 26 # DInt (4 bytes) -OFFSET_DINT2 = 30 # DInt (4 bytes) -OFFSET_CHAR1 = 34 # Char (1 byte) -OFFSET_CHAR2 = 35 # Char (1 byte) -OFFSET_BOOLS = 36 # 8 Bools packed in 1 byte - -# Total size of DB -DB_SIZE = 37 - -# ============================================================================= -# Expected values from DB1 "Read_only" -# ============================================================================= -EXPECTED_INT1 = 10 -EXPECTED_INT2 = 255 -EXPECTED_FLOAT1 = 123.45 -EXPECTED_FLOAT2 = 543.21 -EXPECTED_BYTE1 = 0x0F -EXPECTED_BYTE2 = 0xF0 -EXPECTED_WORD1 = 0xABCD -EXPECTED_WORD2 = 0x1234 -EXPECTED_DWORD1 = 0x12345678 -EXPECTED_DWORD2 = 0x89ABCDEF -EXPECTED_DINT1 = 2147483647 -EXPECTED_DINT2 = 42 -EXPECTED_CHAR1 = "F" -EXPECTED_CHAR2 = "-" -EXPECTED_BOOL0 = True -EXPECTED_BOOL1 = False -EXPECTED_BOOL2 = False -EXPECTED_BOOL3 = False -EXPECTED_BOOL4 = False -EXPECTED_BOOL5 = False -EXPECTED_BOOL6 = False -EXPECTED_BOOL7 = False - - # ============================================================================= # Test Classes # ============================================================================= @@ -262,6 +254,7 @@ def test_db_read_entire_block(self) -> None: @pytest.mark.e2e +@pytest.mark.plc_write class TestClientDBWrite(unittest.TestCase): """Tests for db_write() method - writing to DB2 (read/write).""" @@ -391,6 +384,7 @@ def test_read_area_db(self) -> None: @pytest.mark.e2e +@pytest.mark.plc_write class TestClientWriteArea(unittest.TestCase): """Tests for write_area() method.""" @@ -715,11 +709,13 @@ def test_set_connection_type(self) -> None: self.client.set_connection_type(2) # OP self.client.set_connection_type(3) # S7Basic + @pytest.mark.administrative def test_set_session_password(self) -> None: """Test set_session_password() method.""" result = self.client.set_session_password("testpass") self.assertEqual(0, result) + @pytest.mark.administrative def test_clear_session_password(self) -> None: """Test clear_session_password() method.""" result = self.client.clear_session_password() @@ -750,6 +746,7 @@ def test_error_text(self) -> None: text = self.client.error_text(0x01E00000) self.assertEqual("CPU : Invalid password", text) + @pytest.mark.plc_write def test_iso_exchange_buffer(self) -> None: """Test iso_exchange_buffer() method.""" # Write a value first diff --git a/tests/test_real_plc_support.py b/tests/test_real_plc_support.py new file mode 100644 index 00000000..7cd97b7d --- /dev/null +++ b/tests/test_real_plc_support.py @@ -0,0 +1,77 @@ +"""Unit tests for real-PLC safety and reporting helpers (no hardware needed).""" + +from __future__ import annotations + +import json +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from tests.real_plc.reporting import RealPLCReport, sanitize_diagnostic +from tests.real_plc.support import ScratchRestoreGuard, assert_canonical_fixture, canonical_fixture_bytes + + +def test_canonical_fixture_validates_all_documented_types() -> None: + data = canonical_fixture_bytes() + assert len(data) == 37 + assert_canonical_fixture(data) + + +@pytest.mark.parametrize("failure", [None, AssertionError("scenario failed"), KeyboardInterrupt()]) +def test_scratch_guard_restores_after_success_failure_or_interrupt(failure: BaseException | None) -> None: + original = b"\x12\x34" + adapter = MagicMock() + adapter.read.side_effect = [original, original] + guard = ScratchRestoreGuard(adapter, 2, 0, 2) + + if failure is None: + guard.restore() + else: + with pytest.raises(type(failure)): + try: + raise failure + finally: + guard.restore() + + adapter.write.assert_called_once_with(2, 0, original) + + +def test_scratch_guard_detects_failed_restoration() -> None: + adapter = MagicMock() + adapter.read.side_effect = [b"old", b"bad"] + guard = ScratchRestoreGuard(adapter, 2, 0, 3) + with pytest.raises(AssertionError, match="restoration verification failed"): + guard.restore() + + +def test_report_sanitizes_network_and_credentials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("tests.real_plc.reporting._git_source", lambda: {"kind": "git", "commit": "a" * 40, "dirty": False}) + report = RealPLCReport() + fake = SimpleNamespace( + location=("tests/real_plc/test_acceptance.py", 1, "test"), + when="call", + passed=False, + skipped=False, + keywords={"smoke": 1, "real_plc": 1}, + longrepr="connect 192.168.10.2 password=hunter2 private_key=/tmp/key", + nodeid="tests/real_plc/test_acceptance.py::test_failure", + duration=0.1, + ) + report.record(fake) + target = tmp_path / "report.json" + report.write(target, {"tester": "@tester", "model": "CPU 1511"}) + + serialized = target.read_text() + payload = json.loads(serialized) + assert payload["schema_version"] == "1.0" + assert payload["overall_result"] == "fail" + assert "192.168.10.2" not in serialized + assert "hunter2" not in serialized + assert "/tmp/key" not in serialized + assert serialized.count("= 3 + + +def test_sanitize_diagnostic_caps_size() -> None: + assert len(sanitize_diagnostic("x" * 20, limit=5)) == 5 diff --git a/tests/test_s7_e2e.py b/tests/test_s7_e2e.py index be6de1c9..9a8f8bc9 100644 --- a/tests/test_s7_e2e.py +++ b/tests/test_s7_e2e.py @@ -48,6 +48,40 @@ import pytest from s7commplus.client import S7CommPlusClient +from tests.real_plc.support import ( + DB_SIZE, + EXPECTED_BOOL0, + EXPECTED_BOOL1, + EXPECTED_BYTE1, + EXPECTED_BYTE2, + EXPECTED_CHAR1, + EXPECTED_CHAR2, + EXPECTED_DINT1, + EXPECTED_DINT2, + EXPECTED_DWORD1, + EXPECTED_DWORD2, + EXPECTED_FLOAT1, + EXPECTED_FLOAT2, + EXPECTED_INT1, + EXPECTED_INT2, + EXPECTED_WORD1, + EXPECTED_WORD2, + OFFSET_BOOLS, + OFFSET_BYTE1, + OFFSET_BYTE2, + OFFSET_CHAR1, + OFFSET_CHAR2, + OFFSET_DINT1, + OFFSET_DINT2, + OFFSET_DWORD1, + OFFSET_DWORD2, + OFFSET_FLOAT1, + OFFSET_FLOAT2, + OFFSET_INT1, + OFFSET_INT2, + OFFSET_WORD1, + OFFSET_WORD2, +) # Enable DEBUG logging for all s7 modules so we get full hex dumps logging.basicConfig( @@ -71,49 +105,6 @@ DB_READ_WRITE = int(os.environ.get("PLC_DB_WRITE", "2")) -# ============================================================================= -# DB Structure - Byte offsets for each variable (same as regular S7 e2e tests) -# ============================================================================= -OFFSET_INT1 = 0 # Int (2 bytes) -OFFSET_INT2 = 2 # Int (2 bytes) -OFFSET_FLOAT1 = 4 # Real (4 bytes) -OFFSET_FLOAT2 = 8 # Real (4 bytes) -OFFSET_BYTE1 = 12 # Byte (1 byte) -OFFSET_BYTE2 = 13 # Byte (1 byte) -OFFSET_WORD1 = 14 # Word (2 bytes) -OFFSET_WORD2 = 16 # Word (2 bytes) -OFFSET_DWORD1 = 18 # DWord (4 bytes) -OFFSET_DWORD2 = 22 # DWord (4 bytes) -OFFSET_DINT1 = 26 # DInt (4 bytes) -OFFSET_DINT2 = 30 # DInt (4 bytes) -OFFSET_CHAR1 = 34 # Char (1 byte) -OFFSET_CHAR2 = 35 # Char (1 byte) -OFFSET_BOOLS = 36 # 8 Bools packed in 1 byte - -# Total size of DB -DB_SIZE = 37 - -# ============================================================================= -# Expected values from DB1 "Read_only" -# ============================================================================= -EXPECTED_INT1 = 10 -EXPECTED_INT2 = 255 -EXPECTED_FLOAT1 = 123.45 -EXPECTED_FLOAT2 = 543.21 -EXPECTED_BYTE1 = 0x0F -EXPECTED_BYTE2 = 0xF0 -EXPECTED_WORD1 = 0xABCD -EXPECTED_WORD2 = 0x1234 -EXPECTED_DWORD1 = 0x12345678 -EXPECTED_DWORD2 = 0x89ABCDEF -EXPECTED_DINT1 = 2147483647 -EXPECTED_DINT2 = 42 -EXPECTED_CHAR1 = "F" -EXPECTED_CHAR2 = "-" -EXPECTED_BOOL0 = True -EXPECTED_BOOL1 = False - - # ============================================================================= # Test Classes # ============================================================================= @@ -253,6 +244,7 @@ def test_db_read_entire_block(self) -> None: @pytest.mark.e2e +@pytest.mark.plc_write class TestS7CommPlusDBWrite(unittest.TestCase): """Tests for db_write() - writing to DB2 (read/write).""" @@ -535,6 +527,7 @@ def test_diag_raw_get_multi_variables(self) -> None: print(f"\n{'=' * 60}") + @pytest.mark.administrative def test_diag_raw_set_variable(self) -> None: """Try SetVariable (0x04F2) instead of SetMultiVariables to see if PLC responds differently.""" from s7commplus.protocol import FunctionCode @@ -563,8 +556,8 @@ def test_diag_raw_set_variable(self) -> None: def test_diag_explore_then_read(self) -> None: """Explore first to discover object IDs, then try reading using those IDs.""" - from s7commplus.protocol import FunctionCode, ElementID - from s7commplus.vlq import encode_uint32_vlq, decode_uint32_vlq + from s7commplus.protocol import ElementID, FunctionCode + from s7commplus.vlq import decode_uint32_vlq, encode_uint32_vlq print(f"\n{'=' * 60}") print("DIAGNOSTIC: Explore -> extract object IDs -> try reading") diff --git a/tools/run_real_plc_acceptance.py b/tools/run_real_plc_acceptance.py new file mode 100755 index 00000000..184efd97 --- /dev/null +++ b/tools/run_real_plc_acceptance.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""Run the safe real-PLC acceptance profile and produce shareable artifacts.""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +from pathlib import Path + +import pytest + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--plc-ip", required=True, help="PLC address (used for connection only; never written to reports)") + parser.add_argument("--protocol", required=True, choices=("legacy_s7", "s7commplus")) + parser.add_argument("--tester", required=True, help="GitHub handle") + parser.add_argument("--plc-family", required=True) + parser.add_argument("--plc-model", required=True) + parser.add_argument("--plc-firmware", required=True) + parser.add_argument("--plc-order-code", default="") + parser.add_argument("--plc-security-mode", required=True) + parser.add_argument("--plc-tia-configuration", default="") + parser.add_argument("--plc-port", type=int, default=102) + parser.add_argument("--plc-rack", type=int, default=0) + parser.add_argument("--plc-slot", type=int, default=1) + parser.add_argument("--plc-db-read", type=int, default=1) + parser.add_argument("--plc-db-write", type=int, default=2) + parser.add_argument("--allow-write", action="store_true", help="Also run scratch writes with verified restoration") + parser.add_argument("--allow-admin", action="store_true", help="Also run disruptive administrative scenarios") + parser.add_argument("--output-dir", type=Path, default=Path("real-plc-results")) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + args.output_dir.mkdir(parents=True, exist_ok=True) + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + stem = f"real-plc-{args.protocol}-{stamp}" + junit = args.output_dir / f"{stem}.junit.xml" + report = args.output_dir / f"{stem}.json" + marker = "smoke" + if args.allow_write: + marker += " or write" + if args.allow_admin: + marker += " or administrative" + pytest_args = [ + "tests/real_plc/test_acceptance.py", + "--e2e", + "-v", + "--gherkin-terminal-reporter", + "-m", + marker, + f"--junitxml={junit}", + f"--plc-report-json={report}", + f"--plc-ip={args.plc_ip}", + f"--plc-protocol={args.protocol}", + f"--tester={args.tester}", + f"--plc-family={args.plc_family}", + f"--plc-model={args.plc_model}", + f"--plc-firmware={args.plc_firmware}", + f"--plc-order-code={args.plc_order_code}", + f"--plc-security-mode={args.plc_security_mode}", + f"--plc-tia-configuration={args.plc_tia_configuration}", + f"--plc-port={args.plc_port}", + f"--plc-rack={args.plc_rack}", + f"--plc-slot={args.plc_slot}", + f"--plc-db-read={args.plc_db_read}", + f"--plc-db-write={args.plc_db_write}", + ] + if args.allow_write: + pytest_args.append("--allow-plc-write") + if args.allow_admin: + pytest_args.append("--allow-plc-admin") + return pytest.main(pytest_args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/uv.lock b/uv.lock index 8fe3e833..7aff6ab4 100644 --- a/uv.lock +++ b/uv.lock @@ -525,6 +525,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/52/af/9b01bcf5c91e81899bb890b87bd9077732a9b3365c098e67fe77958c39ed/filelock-3.30.0-py3-none-any.whl", hash = "sha256:40632998f0772e64183bb819f086a1b9def6be1090cf1dcb9d45f46806ef279b", size = 93131, upload-time = "2026-07-16T03:53:56.727Z" }, ] +[[package]] +name = "gherkin-official" +version = "29.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f3/d8/7a28537efd7638448f7512a0cce011d4e3bf1c7f4794ad4e9c87b3f1e98e/gherkin_official-29.0.0.tar.gz", hash = "sha256:dbea32561158f02280d7579d179b019160d072ce083197625e2f80a6776bb9eb", size = 32303, upload-time = "2024-08-12T09:41:09.595Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f8/fc/b86c22ad3b18d8324a9d6fe5a3b55403291d2bf7572ba6a16efa5aa88059/gherkin_official-29.0.0-py3-none-any.whl", hash = "sha256:26967b0d537a302119066742669e0e8b663e632769330be675457ae993e1d1bc", size = 37085, upload-time = "2024-08-12T09:41:07.954Z" }, +] + [[package]] name = "hypothesis" version = "6.167.1" @@ -755,6 +764,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/5f/5d/3dcec2884ba1b0806d1408612555c38dd5d68e90156b59f75f6e36435c3a/librt-0.13.0-cp314-cp314t-win_arm64.whl", hash = "sha256:2f281549a4c52ac7bb97997f14353f8bd0e53a34ca0dad1c905cfd0b4a58ae99", size = 110771, upload-time = "2026-07-08T12:26:12.303Z" }, ] +[[package]] +name = "mako" +version = "1.4.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markupsafe" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/2a/12/b5fa2353e2754cd67fb9f83793fa48ff42c213a5da7e719869d2301f6ab8/mako-1.4.1.tar.gz", hash = "sha256:d7904710b662996425a21627710c4777c45053146942cf8a7aebf757c92b8c27", size = 410165, upload-time = "2026-08-05T06:10:56.611Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a5/54/12ed58d458474aaab5c3d180173e745a4fe131bb330370596876d19ff60f/mako-1.4.1-py3-none-any.whl", hash = "sha256:a359d9a94a541213958742b2698d0a7757bb83551767bc468a74b9905aba9617", size = 80010, upload-time = "2026-08-05T06:10:58.248Z" }, +] + [[package]] name = "markdown-it-py" version = "4.2.0" @@ -946,6 +967,28 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" }, ] +[[package]] +name = "parse" +version = "1.22.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a4/f2/0b504486c2a5564798607d3860e48ed19c6443d5e9cc3ec61cc6b8b4ef58/parse-1.22.1.tar.gz", hash = "sha256:d3a4740ec3da338e2b258b2d69741b731eadfddca59e24a14bc4ee5fce38c911", size = 36970, upload-time = "2026-05-26T03:44:52.624Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/6f/c5/7c16e99869e1f422629092cfd23e3b58e461988c3f9c36fd3624bb4142e6/parse-1.22.1-py2.py3-none-any.whl", hash = "sha256:20f0925a46f06602485ac90d751764d0697fd8455aaa97489ba8953a4b66de32", size = 20925, upload-time = "2026-05-26T03:44:51.156Z" }, +] + +[[package]] +name = "parse-type" +version = "0.6.6" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "parse" }, + { name = "six" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/19/ea/42ba6ce0abba04ab6e0b997dcb9b528a4661b62af1fe1b0d498120d5ea78/parse_type-0.6.6.tar.gz", hash = "sha256:513a3784104839770d690e04339a8b4d33439fcd5dd99f2e4580f9fc1097bfb2", size = 98012, upload-time = "2025-08-11T22:53:48.066Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/85/8d/eef3d8cdccc32abdd91b1286884c99b8c3a6d3b135affcc2a7a0f383bb32/parse_type-0.6.6-py2.py3-none-any.whl", hash = "sha256:3ca79bbe71e170dfccc8ec6c341edfd1c2a0fc1e5cfd18330f93af938de2348c", size = 27085, upload-time = "2025-08-11T22:53:46.396Z" }, +] + [[package]] name = "pathspec" version = "1.1.1" @@ -1077,6 +1120,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930, upload-time = "2026-05-26T09:56:02.576Z" }, ] +[[package]] +name = "pytest-bdd" +version = "8.1.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "gherkin-official" }, + { name = "mako" }, + { name = "packaging" }, + { name = "parse" }, + { name = "parse-type" }, + { name = "pytest" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/2d/2f/14c2e55372a5718a93b56aea48cd6ccc15d2d245364e516cd7b19bbd07ad/pytest_bdd-8.1.0.tar.gz", hash = "sha256:ef0896c5cd58816dc49810e8ff1d632f4a12019fb3e49959b2d349ffc1c9bfb5", size = 56147, upload-time = "2024-12-05T21:45:58.83Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9f/7d/1461076b0cc9a9e6fa8b51b9dea2677182ba8bc248d99d95ca321f2c666f/pytest_bdd-8.1.0-py3-none-any.whl", hash = "sha256:2124051e71a05ad7db15296e39013593f72ebf96796e1b023a40e5453c47e5fb", size = 49149, upload-time = "2024-12-05T21:45:56.184Z" }, +] + [[package]] name = "pytest-cov" version = "7.1.0" @@ -1162,6 +1223,7 @@ test = [ { name = "mypy" }, { name = "pytest" }, { name = "pytest-asyncio" }, + { name = "pytest-bdd" }, { name = "pytest-cov" }, { name = "pytest-html" }, { name = "ruff" }, @@ -1183,6 +1245,7 @@ requires-dist = [ { name = "psutil", marker = "extra == 'demo'" }, { name = "pytest", marker = "extra == 'test'" }, { name = "pytest-asyncio", marker = "extra == 'test'" }, + { name = "pytest-bdd", marker = "extra == 'test'", specifier = ">=8.1,<9" }, { name = "pytest-cov", marker = "extra == 'test'" }, { name = "pytest-html", marker = "extra == 'test'" }, { name = "rich", marker = "extra == 'cli'" }, @@ -1285,6 +1348,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/46/b4/21180f6fbbf7ed2043bc353608c7b9cc6f7b3d342ad8c287a097e5ca8477/setuptools_scm-10.2.0-py3-none-any.whl", hash = "sha256:6cc5ac7da8e54d74e43503be2fa023eb7a1ca820d5c318e6ae20efc48110d6c7", size = 27741, upload-time = "2026-06-25T05:27:09.037Z" }, ] +[[package]] +name = "six" +version = "1.17.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031, upload-time = "2024-12-04T17:35:28.174Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" }, +] + [[package]] name = "snowballstemmer" version = "3.1.1"