From 17b3f4a5c8ad4025da63adc9f9f9f8628ce2d600 Mon Sep 17 00:00:00 2001 From: Kaiser <104786001+KaiserIIII@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:24:07 +0800 Subject: [PATCH 1/2] fix: harden acceptance lifecycle, protocols and report lineage AI-assisted reliability review with 113 added offline regressions. Preserve original task evidence, fail closed on persistent storage failure, validate protocol boundaries and append CSV retest parent identity. --- CHANGELOG.md | 10 + config/CONFIG_GUIDE.md | 2 + core/config_manager.py | 27 ++- core/protocol_engine.py | 30 ++- docs/API.md | 6 + docs/VALIDATION.md | 15 ++ inspection/http.py | 57 ++++- inspection/reports.py | 3 +- inspection/service.py | 52 ++-- inspection/store.py | 4 +- tests/fixtures/csv-lineage-legacy29.csv | 3 + tests/fixtures/csv-lineage-retest.json | 130 ++++++++++ tests/test_csv_lineage.py | 106 ++++++++ tests/test_hierarchy_boundary.py | 108 +++++++++ tests/test_protocol_boundaries.py | 101 ++++++++ tests/test_rejected_post_delivery.py | 106 ++++++++ tests/test_reliability_http.py | 134 +++++++++++ tests/test_reliability_lifecycle.py | 219 +++++++++++++++++ tests/test_reliability_storage.py | 307 ++++++++++++++++++++++++ tests/test_request_admission.py | 43 ++++ 20 files changed, 1417 insertions(+), 46 deletions(-) create mode 100644 tests/fixtures/csv-lineage-legacy29.csv create mode 100644 tests/fixtures/csv-lineage-retest.json create mode 100644 tests/test_csv_lineage.py create mode 100644 tests/test_hierarchy_boundary.py create mode 100644 tests/test_protocol_boundaries.py create mode 100644 tests/test_rejected_post_delivery.py create mode 100644 tests/test_reliability_http.py create mode 100644 tests/test_reliability_lifecycle.py create mode 100644 tests/test_reliability_storage.py create mode 100644 tests/test_request_admission.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 84c0c8e..075c3e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## Unreleased — 2026-10-15 reliability review + +- Reject untrusted POST headers before reading request bodies; validate duplicate/ambiguous headers and preserve bounded error-response delivery. +- Protect finished task history from stale/conflicting callbacks, reject work during shutdown and fail closed after an unrecoverable worker persistence error. +- Persist retest parent identity in the initial insert; clean up failed SQLite bootstrap and worker startup. +- Apply shared response constraints to `one_of` candidates, enforce read-function-specific Modbus quantities/address spans and reject non-finite or non-integral settings. +- Reject unsupported device descendants beyond one child level instead of silently reducing the expected-device count. +- Append `parent_job_id` to CSV exports as column 30. Existing 29 columns keep their order and values; consumers requiring exactly 29 columns need updating. +- Add 113 automated regressions above the 100-test public baseline. The isolated Windows candidate passes 213 tests, including combined worker-start/storage failure; remote matrix outcomes are recorded separately in PR checks. These changes and tests were developed with AI assistance. + ## 3.1 — 2026-10-08 - Add English/Chinese interfaces for the acceptance workbench and compatibility scan page, with browser-language defaults and a remembered preference. diff --git a/config/CONFIG_GUIDE.md b/config/CONFIG_GUIDE.md index 785a072..c75c2ad 100644 --- a/config/CONFIG_GUIDE.md +++ b/config/CONFIG_GUIDE.md @@ -290,6 +290,8 @@ 中间模块 `is_intermediate_module` 设置为 `true`，子设备写在 `child_devices` 里。子设备默认继承父设备接口和通信方式；父设备通信失败时，子设备会标记为无法测试。 +当前执行与设备选择仅支持父设备及其直接子设备。子设备不能再包含非空的 `child_devices`；更深层级会在配置校验时拒绝，避免遗漏设备后仍生成“完整配置通过”的报告。省略或清空叶子设备的 `child_devices` 不受影响。 + ```json { "device_id": "SER_IO_001", diff --git a/core/config_manager.py b/core/config_manager.py index 05484f9..2d02710 100644 --- a/core/config_manager.py +++ b/core/config_manager.py @@ -1,10 +1,11 @@ import copy import json +import math import os import re from typing import Dict, Iterable, List, Optional -from .protocol_engine import ProtocolConfigError, decode_hex +from .protocol_engine import ProtocolConfigError, decode_hex, validate_modbus_read DEFAULT_CONFIG = os.path.join( @@ -300,6 +301,9 @@ def _validate_device( seen_ids: Optional[set] = None, ): device_id = device.get("device_id") or label + if parent is not None and device.get("child_devices"): + errors.append(f"{device_id} child_devices仅支持一级子设备，不支持继续嵌套") + return for key in ("device_id", "device_name", "device_type"): if not device.get(key): errors.append(f"{device_id} 缺少 {key}") @@ -353,7 +357,7 @@ def _validate_serial_config(self, config: Dict[str, object], errors: List[str], errors.append(f"{device_id} parity必须是N、E、O、M或S") if stopbits not in (1, 1.5, 2): errors.append(f"{device_id} stopbits必须是1、1.5或2") - if timeout <= 0 or timeout > 60: + if not math.isfinite(timeout) or timeout <= 0 or timeout > 60: errors.append(f"{device_id} timeout必须在0到60秒之间") def _validate_network_config(self, config: Dict[str, object], errors: List[str], device_id: str): @@ -387,16 +391,19 @@ def _validate_protocol(self, protocol: Dict[str, object], errors: List[str], dev continue if protocol_type in ("modbus_rtu", "modbus_tcp"): try: - function_code = int(probe.get("function_code", protocol.get("function_code", 3)) or 3) - address = int(probe.get("address", 0) or 0) - count = int(probe.get("count", 1) or 1) - except (TypeError, ValueError): + values = (probe.get("function_code", protocol.get("function_code", 3)), + probe.get("address", 0), probe.get("count", 1)) + if any(isinstance(value, bool) or isinstance(value, float) and + (not math.isfinite(value) or not value.is_integer()) for value in values): + raise ValueError("Non-integral protocol parameter") + function_code, address, count = (int(value) for value in values) + except (TypeError, ValueError, OverflowError): errors.append(f"{device_id} 第{index}个Modbus规则参数必须是整数") continue - if function_code not in (1, 2, 3, 4): - errors.append(f"{device_id} 自动检测只允许Modbus只读功能码1、2、3、4") - if not 0 <= address <= 65535 or not 1 <= count <= 2000: - errors.append(f"{device_id} 第{index}个Modbus规则address/count超出范围") + try: + validate_modbus_read(function_code, address, count) + except ProtocolConfigError as exc: + errors.append(f"{device_id} 第{index}个Modbus规则: {exc}") operation = str(probe.get("operation") or protocol.get("operation") or "read").lower() if operation == "write" and not self.allow_write_tests(): errors.append(f"{device_id} 配置了写测试，但settings.allow_write_tests不是true") diff --git a/core/protocol_engine.py b/core/protocol_engine.py index 9b1d227..d33db6f 100644 --- a/core/protocol_engine.py +++ b/core/protocol_engine.py @@ -140,13 +140,6 @@ def match_response(data: bytes, rule: Optional[Dict[str, object]] = None) -> Tup if not data: return False, "未收到响应" - if "one_of" in rule: - matches = [match_response(data, item or {}) for item in rule.get("one_of", []) or []] - for ok, note in matches: - if ok: - return True, note - return False, "响应不符合任一候选规则" - length = len(data) if "min_length" in rule and length < int(rule["min_length"]): return False, f"响应长度{length}小于要求{rule['min_length']}" @@ -181,12 +174,28 @@ def match_response(data: bytes, rule: Optional[Dict[str, object]] = None) -> Tup return False, "响应十六进制内容不符合正则表达式" if rule.get("checksum") not in (None, "") and not _check_crc(data, str(rule["checksum"])): return False, "响应校验和不正确" + if "one_of" in rule: + for item in rule.get("one_of", []) or []: + ok, note = match_response(data, item or {}) + if ok: + return True, note + return False, "响应不符合任一候选规则" return True, f"响应符合配置规则（{length}字节）" -def build_modbus_rtu_read(slave_id: int, function_code: int, address: int, count: int) -> bytes: +def validate_modbus_read(function_code: int, address: int, count: int) -> None: + """Validate read PDU limits before bytes are packed or hardware is opened.""" + if any(type(value) is not int for value in (function_code, address, count)): + raise ProtocolConfigError("Modbus功能码、地址和数量必须是整数") if function_code not in READ_ONLY_MODBUS_FUNCTIONS: - raise ProtocolConfigError(f"Modbus RTU自动检测只允许只读功能码1、2、3、4，当前为{function_code}") + raise ProtocolConfigError(f"Modbus自动检测只允许只读功能码1、2、3、4，当前为{function_code}") + maximum = 125 if function_code in (3, 4) else 2000 + if not 0 <= address <= 65535 or not 1 <= count <= maximum or address + count > 65536: + raise ProtocolConfigError(f"Modbus读取地址/数量超出范围：功能码{function_code}数量须为1到{maximum}且不能跨越65535地址") + + +def build_modbus_rtu_read(slave_id: int, function_code: int, address: int, count: int) -> bytes: + validate_modbus_read(function_code, address, count) payload = bytes([ slave_id & 0xFF, function_code & 0xFF, @@ -237,8 +246,7 @@ def find_modbus_rtu_response(data: bytes, slave_id: int, function_code: int, def build_modbus_tcp_read(transaction_id: int, unit_id: int, function_code: int, address: int, count: int) -> bytes: - if function_code not in READ_ONLY_MODBUS_FUNCTIONS: - raise ProtocolConfigError(f"Modbus TCP自动检测只允许只读功能码1、2、3、4，当前为{function_code}") + validate_modbus_read(function_code, address, count) pdu = bytes([ function_code & 0xFF, (address >> 8) & 0xFF, diff --git a/docs/API.md b/docs/API.md index a4f6b9f..3ac9914 100644 --- a/docs/API.md +++ b/docs/API.md @@ -6,6 +6,8 @@ Base URL defaults to `http://127.0.0.1:8080`. JSON is UTF-8. When `QLDC_ACCESS_T Exports accept `lang=zh` (default) or `lang=en`. HTML headings, labels and explanations use that language; metadata, configuration and raw evidence stay original. JSON/CSV schemas and content remain unchanged by language. Unsupported language values return400. Example: `/api/jobs/{id}/export?format=html&lang=en`. +CSV appends `parent_job_id` as its final,30th column. An ordinary task (including an older record without that field) exports an empty cell; a retest exports its immediate source task ID. The value comes from the task record, not a per-device field. The previous29 columns keep their positions and values, including quoting of commas, newlines and Unicode evidence. CSV readers should use the header; consumers requiring exactly29 columns need updating for this additive field. JSON and HTML keep their existing parent-link representation. + | Method | Route | Behavior | | --- | --- | --- | | GET | `/api/health` | Service mode, version, active job and persistence status | @@ -34,6 +36,8 @@ Example task payload: Omit `device_ids` to include the complete catalog; provide an array of known unique IDs to select a subset. Metadata strings have 128-character limits, except notes at 2,000. `scenario` is `healthy`, `faults`, `timeout`, `crc` or `missing` in demo; live accepts only `healthy` (the scenario label does not predetermine actual device results). Metadata may alternatively be nested under `metadata`. +The supported catalog hierarchy is a parent with direct children. A child cannot contain further nonempty `child_devices`; configuration validation rejects deeper trees before task persistence or adapter invocation. Empty/omitted child lists remain valid. This prevents an unsupported descendant from silently disappearing from the expected-device count. + Retest accepts `scenario`, `operator`, `notes`, and optional non-passing `device_ids`; it cannot change unit/batch, widen scope or switch mode. An empty eligible set is rejected. The returned task has `parent_job_id`. ## Record semantics @@ -44,6 +48,8 @@ Task IDs are 32 lowercase hex characters. Times are ISO 8601 UTC. Status: `queue Errors use `{ "ok": false, "error": "..." }`: 400 invalid fields/configuration, 401 token required, 403 origin/host or mode restriction, 404 missing resource, 409 station busy/service conflict, 413 oversized input, 500 unexpected failure. Errors omit exception traces and private paths. Do not automatically replay a hardware POST after an ambiguous connection failure; first inspect current task/history. +If a worker cannot persist either its outcome or a failed terminal state, `/api/health` reports `persistence_ready: false` and `persistence_error: {"code":"PERSISTENCE_UNAVAILABLE","job_id":"...","restart_required":true}`. New tasks, retests and legacy operations are denied until restart, even if storage becomes writable again. The last durable task can remain `running`; after storage is repaired, restart marks unfinished records `interrupted` without replaying hardware. Back up the database before recovery. HTTP 200 from health only means the endpoint responded; inspect the persistence fields before starting work. + ## Legacy interfaces `/api/scan`, `/api/scan-interface`, `/api/run`, `/api/test-device`, `/api/standards` retain the original live interface and return `legacy: true`. They are disabled in demo. `/legacy` is the compatibility UI; `/report/` serves existing report files. Legacy operations share the acceptance execution lock but do not create new SQLite acceptance records. diff --git a/docs/VALIDATION.md b/docs/VALIDATION.md index 14a4230..6a47643 100644 --- a/docs/VALIDATION.md +++ b/docs/VALIDATION.md @@ -48,6 +48,21 @@ See [DEMO.md](DEMO.md) for the optional browser workflow. CI runs the Python off - Browser workflow: the new bilingual workflow passed on Chromium with English and Chinese switching, task creation, evidence display, English HTML download, failed-only retest, history filters, cross-page comparison selection retention, reload preference, blocked `localStorage` fallback, the compatibility scan page and 390 px layout. The existing Chinese workflow also passed with zero page errors. - API boundary: omitted `lang` keeps the Chinese default; `lang=zh` and `lang=en` return the requested report language; unsupported or explicitly blank language values return HTTP 400. No live device write or unknown hardware operation was used by these checks. +## Recorded results — 2026-10-15 reliability review + +The AI-assisted review used public baseline `c7e3ddcf6e6594ea9874cd63b870183bdd6850ea`, isolated source copies and an independent clone for the merge candidate. No real devices were accessed and no new dependency was installed during the local verification. Existing CPython 3.13.0, pytest 8.4.2 and coverage 7.10.6 were used on Windows. + +- Full isolated-clone regression: **212 passed, zero failed/skipped**, in 38.01 seconds. The public baseline had 100 tests; the added regressions cover lifecycle, real temporary SQLite lock/commit/bootstrap/recovery faults, worker startup failure, stale/conflicting callbacks, repeated cancellation, HTTP admission/framing/teardown, protocol boundaries, nested-device rejection and CSV retest provenance. +- Original offline self-test: **23 checks passed**, including actual TCP/UDP exchange with loopback fixtures. Configuration validation passed with 5 root devices, 1 child and writes disabled. Python compilation passed. +- Broad runtime coverage, using `--branch --source=. --omit=tests/*,tools/*,self_test.py,validate_config.py`: **1,816/3,627 executable lines (50.07%)** and **558/1,414 branches (39.46%)**. This is a broader denominator than earlier targeted figures; it is not an 80% whole-project coverage claim. +- Real bounded HTTP tests deliver 401/403 JSON while a rejected body is already in flight and verify teardown byte/time limits. They do not establish resilience against arbitrary internet load. +- CSV tests exercise an actual three-task Demo retest chain and the loopback export endpoint. All original 29 columns/cells are preserved; column 30 contains the immediate `parent_job_id` (empty for roots). English/Chinese exports share the same CSV schema. Exact-29-column external consumers require adaptation. +- Storage failure cannot manufacture a durable terminal result when writes are unavailable. The tested response is explicit degradation, refusal of further operations, then interrupted-state recovery after repair/restart; there is no automatic hardware retry. + +Linux and Python 3.10 were not executed locally during this review. Use the final PR's existing Windows/Linux × Python 3.10/3.13 CI checks for remote results. Browser checks are separate: the earlier 203-test candidate was exercised on desktop/mobile with synthetic faults, retest, reload, keyboard controls and exports; the added CSV lineage column is covered by Python/API tests, not a new browser acceptance claim. No new release bundle or physical-device certification is implied. + +Pre-merge review found and reproduced one additional compound failure: a worker could fail to start while its fallback terminal write also failed, leaving a queued record without the persistence diagnostic. The guarded fallback now sets the same restart-required state and preserves the original startup exception. A real temporary SQLite-trigger regression checks blocked creation/retest/legacy work, continued refusal after storage repair, and restart recovery without replay. The final staged Git tree, including normalized CSV/JSON fixtures, passed **213 tests in 34.70 seconds**, plus configuration validation, all 23 self-checks and compilation. Final broad runtime coverage is **1,819/3,630 lines (50.11%)** and **558/1,414 branches (39.46%)**, using the same command above. This final run supersedes the earlier 212-test candidate; no physical hardware or sustained-load conclusion is inferred. + ## Scope and limits Coverage is collected for `inspection/`, `core.protocol_engine` and `core.transport`, not the entire legacy detector or repository. The original offline self-test adds loopback checks but does not establish field hardware compatibility. Browser tests are separate from the Python CI job and use an existing local browser runtime. diff --git a/inspection/http.py b/inspection/http.py index c9078b8..aea3dd6 100644 --- a/inspection/http.py +++ b/inspection/http.py @@ -2,6 +2,8 @@ import ipaddress import json import re +import socket +import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, unquote, urlsplit @@ -47,6 +49,32 @@ def setup(self): super().setup() self.connection.settimeout(5) + def finish(self): + try: + super().finish() + finally: + if getattr(self, '_rejected_post', False): + self._drain_rejected_input() + + def _drain_rejected_input(self): + # Flush/send the rejection first, then half-close and discard a bounded + # tail. Immediate close with unread input can reset the error response. + try: + self.connection.shutdown(socket.SHUT_WR) + deadline = time.monotonic() + .1 + remaining_bytes = MAX_BODY + 1 + while remaining_bytes: + remaining_time = deadline - time.monotonic() + if remaining_time <= 0: + break + self.connection.settimeout(min(.1, remaining_time)) + data = self.connection.recv(min(16384, remaining_bytes)) + if not data: + break + remaining_bytes -= len(data) + except OSError: + pass + def do_GET(self): self._dispatch('GET') @@ -55,11 +83,11 @@ def do_POST(self): def _dispatch(self, method): try: - if method == 'POST': - self._body() parsed = urlsplit(self.path) path = unquote(parsed.path) self._access(path) + if method == 'POST': + self._body() query = parse_qs(parsed.query) if method == 'GET' and path in ('/', '/index.html', '/legacy', '/assets/app.js', '/assets/i18n.js', '/assets/style.css'): relative = {'/': 'web/index.html', '/index.html': 'web/index.html', '/legacy': 'web/legacy.html', '/assets/app.js': 'web/assets/app.js', '/assets/i18n.js': 'web/assets/i18n.js', '/assets/style.css': 'web/assets/style.css'}[path] @@ -72,7 +100,9 @@ def _dispatch(self, method): return self._send(b'', 'image/x-icon', 204) svc = self.server.service if method == 'GET' and path == '/api/health': - return self._json({'ok': True, 'version': '3.1', 'mode': svc.mode, 'active_job_id': svc.active_id, 'persistence_ready': not svc.closed}) + return self._json({'ok': True, 'version': '3.1', 'mode': svc.mode, 'active_job_id': svc.active_id, + 'persistence_ready': not svc.closed and not svc.persistence_error, + 'persistence_error': svc.persistence_error}) if method == 'GET' and path == '/api/catalog': return self._json(dict(ok=True, **svc.catalog())) if method == 'POST' and path == '/api/config/validate': @@ -138,6 +168,9 @@ def _dispatch(self, method): return self._send(target.read_bytes(), mime + '; charset=utf-8') raise HttpError(404, 'Route not found') except HttpError as exc: + if method == 'POST' and not hasattr(self, '_parsed_body'): + self._rejected_post = True + self.close_connection = True self._json({'ok': False, 'error': str(exc)}, exc.status) except KeyError: self._json({'ok': False, 'error': 'Job not found'}, 404) @@ -149,12 +182,16 @@ def _dispatch(self, method): self._json({'ok': False, 'error': 'Operation failed'}, 500) def _access(self, path): + if len(self.headers.get_all('Host', [])) != 1: + raise HttpError(403, 'Invalid host') host = self.headers.get('Host', '') try: parsed = urlsplit('http://' + host) hostname, port = parsed.hostname or '', parsed.port or 80 except ValueError: raise HttpError(403, 'Invalid host') from None + if parsed.netloc != host or parsed.username is not None or parsed.path or parsed.query or parsed.fragment: + raise HttpError(403, 'Invalid host') if port != self.server.server_port or (loopback(self.server.bind_host) and not loopback(hostname)): raise HttpError(403, 'Untrusted host') origin = self.headers.get('Origin') @@ -164,7 +201,7 @@ def _access(self, path): raise HttpError(403, 'Cross-site requests are not allowed') token = self.server.access_token if token and (path.startswith('/api/') or path.startswith('/report/')): - if not hmac.compare_digest(self.headers.get('Authorization', ''), 'Bearer ' + token): + if len(self.headers.get_all('Authorization', [])) != 1 or not hmac.compare_digest(self.headers.get('Authorization', '').encode('utf-8'), ('Bearer ' + token).encode('utf-8')): raise HttpError(401, 'Access token required') def _body(self): @@ -172,8 +209,11 @@ def _body(self): return self._parsed_body if self.headers.get('Transfer-Encoding'): raise HttpError(400, 'Transfer encoding is not supported') + lengths = self.headers.get_all('Content-Length', ['0']) + if len(lengths) != 1 or not re.fullmatch('[0-9]+', lengths[0].strip()): + raise HttpError(400, 'Invalid content length') try: - length = int(self.headers.get('Content-Length', '0')) + length = int(lengths[0]) except ValueError: raise HttpError(400, 'Invalid content length') from None if length < 0: @@ -187,7 +227,10 @@ def _body(self): except OSError: pass raise HttpError(413, 'Request exceeds 64 KiB') - raw = self.rfile.read(length) + try: + raw = self.rfile.read(length) + except OSError: + raise HttpError(400, 'Incomplete request body') from None if len(raw) != length: raise HttpError(400, 'Incomplete request body') try: @@ -209,6 +252,8 @@ def _send(self, content, content_type, status=200, headers=None): self.send_header('Cache-Control', 'no-store') self.send_header('X-Content-Type-Options', 'nosniff') self.send_header('Referrer-Policy', 'no-referrer') + if getattr(self, '_rejected_post', False): + self.send_header('Connection', 'close') for key, value in (headers or {}).items(): self.send_header(key, value) self.end_headers() diff --git a/inspection/reports.py b/inspection/reports.py index 704d99f..30e1ab4 100644 --- a/inspection/reports.py +++ b/inspection/reports.py @@ -85,13 +85,14 @@ def export_job(job, format, lang='zh'): if format == 'csv': stream = io.StringIO(newline='') writer = csv.writer(stream) - columns = ['report_marker', 'job_id', 'status', 'mode', 'scenario', 'station_id', 'batch', 'operator', 'notes', 'config_hash', 'config_snapshot', 'created_at', 'finished_at', 'device_id', 'name', 'connection_type', 'interface', 'protocol', 'scope', 'verdict', 'fault_code', 'summary', 'suggestion', 'request', 'response', 'duration_ms', 'simulated', 'attempts', 'supporting_checks'] + columns = ['report_marker', 'job_id', 'status', 'mode', 'scenario', 'station_id', 'batch', 'operator', 'notes', 'config_hash', 'config_snapshot', 'created_at', 'finished_at', 'device_id', 'name', 'connection_type', 'interface', 'protocol', 'scope', 'verdict', 'fault_code', 'summary', 'suggestion', 'request', 'response', 'duration_ms', 'simulated', 'attempts', 'supporting_checks', 'parent_job_id'] writer.writerow(columns) def safe(value): text = str(value if value is not None else '') return "'" + text if text.lstrip().startswith(('=', '+', '-', '@')) or text.startswith(('\t', '\r', '\n')) else text for result in job['results']: row = dict(job, **job['metadata'], **result, report_marker=marker, job_id=job['id'], config_snapshot=json.dumps(job['config_snapshot'], ensure_ascii=False)) + row['parent_job_id'] = job.get('parent_job_id') for key in ('attempts', 'supporting_checks'): row[key] = json.dumps(result.get(key, []), ensure_ascii=False) writer.writerow([safe(row.get(key, '')) for key in columns]) diff --git a/inspection/service.py b/inspection/service.py index 46bc75d..f3db9f7 100644 --- a/inspection/service.py +++ b/inspection/service.py @@ -26,6 +26,8 @@ def __init__(self, db_path, demo=False, config_path=None, adapter=None): raise self.legacy_state = {'devices': {}, 'updated_at': '', 'config_hash': None} self.closed = False + self.closing = False + self.persistence_error = None self.worker = None self.active_id = None self.active_started = None @@ -54,7 +56,7 @@ def _snapshot(self): @contextmanager def legacy_operation(self): with self.lock: - if self.closed or not self.hardware_lock.acquire(blocking=False): + if self.closed or self.closing or self.persistence_error or not self.hardware_lock.acquire(blocking=False): raise RuntimeError('A station operation is already running or service is closed') try: yield @@ -69,27 +71,38 @@ def catalog(self): return {'mode': self.mode, 'config_hash': config_hash(snapshot), 'config_snapshot': snapshot, 'devices': catalog_devices(snapshot), 'scenarios': SCENARIOS if self.demo else ['healthy']} - def create(self, payload, snapshot=None): + def create(self, payload, snapshot=None, parent_job_id=None): snapshot = copy.deepcopy(snapshot if snapshot is not None else self._snapshot()) devices = catalog_devices(snapshot) ensure_read_only(snapshot) scenario, metadata, selected = validate_payload(payload, devices, self.mode) with self.lock: - if self.closed: - raise RuntimeError('Service is closed') + if self.closed or self.closing: + raise RuntimeError('Service is closed or closing') + if self.persistence_error: + raise RuntimeError('Persistence is unavailable; restart required') if not self.hardware_lock.acquire(blocking=False): raise RuntimeError('A station operation is already running') try: job = refresh({'id': uuid.uuid4().hex, 'status': 'queued', 'mode': self.mode, 'scenario': scenario, 'metadata': metadata, 'config_hash': config_hash(snapshot), 'config_snapshot': snapshot, - 'device_ids': selected, 'parent_job_id': None, 'created_at': utc_now(), 'started_at': None, + 'device_ids': selected, 'parent_job_id': parent_job_id, 'created_at': utc_now(), 'started_at': None, 'finished_at': None, 'duration_ms': 0, 'results': [initial_result(d, self.demo) for d in devices if d['device_id'] in selected], 'events': [{'at': utc_now(), 'type': 'queued', 'message': 'Acceptance queued'}]}) self.store.insert(job) self.active_id = job['id'] self.cancel_event = threading.Event() - self.worker = threading.Thread(target=self._run, args=(job['id'],), daemon=True, name='acceptance-worker') - self.worker.start() + try: + self.worker = threading.Thread(target=self._run, args=(job['id'],), daemon=True, name='acceptance-worker') + self.worker.start() + except Exception: + self.active_id = None + self.worker = None + try: + self._finish(job, 'failed', time.monotonic()) + except Exception: + self.persistence_error = {'code': 'PERSISTENCE_UNAVAILABLE', 'job_id': job['id'], 'restart_required': True} + raise return copy.deepcopy(job) except Exception: self.hardware_lock.release() @@ -105,11 +118,14 @@ def _run(self, job_id): job['started_at'] = utc_now() job['events'].append({'at': utc_now(), 'type': 'running', 'message': 'Acceptance started'}) self.store.update(job) + reported = set() def on_result(result): with self.lock: - if self.closed: + if self.closed or self.active_id != job_id: return job = self.store.get(job_id) + if job['status'] in TERMINAL: + return matches = [i for i, r in enumerate(job['results']) if r['device_id'] == result.get('device_id')] if not matches or result.get('verdict') not in ('PASS', 'FAIL', 'REVIEW', 'NOT_RUN'): raise ValueError('Invalid adapter result') @@ -117,9 +133,14 @@ def on_result(result): value = dict(expected, **copy.deepcopy(result)) value['scope'] = expected['scope'] value['simulated'] = self.demo + if value['device_id'] in reported: + if value == expected: + return + raise ValueError('Conflicting adapter result') job['results'][matches[0]] = value job['events'].append({'at': utc_now(), 'type': 'result', 'device_id': value['device_id'], 'message': value['verdict']}) self.store.update(refresh(job)) + reported.add(value['device_id']) if not self.cancel_event.is_set(): self.adapter.run(copy.deepcopy(job['config_snapshot']), list(job['device_ids']), job['scenario'], self.cancel_event.is_set, on_result) with self.lock: @@ -134,7 +155,10 @@ def on_result(result): except Exception: with self.lock: if not self.closed: - self._finish(self.store.get(job_id), 'failed', started) + try: + self._finish(self.store.get(job_id), 'failed', started) + except Exception: + self.persistence_error = {'code': 'PERSISTENCE_UNAVAILABLE', 'job_id': job_id, 'restart_required': True} finally: with self.lock: self.active_id = None @@ -157,7 +181,7 @@ def list_jobs(self, limit=20, offset=0, status='', query='', sort='newest'): def cancel(self, job_id): with self.lock: job = self.get(job_id) - if job['status'] in TERMINAL: + if job['status'] in TERMINAL or job['status'] == 'cancelling': return job if self.active_id != job_id: raise RuntimeError('Job is not active') @@ -186,12 +210,7 @@ def retest(self, job_id, payload): raise ValueError('No failed or pending devices to retest') data = dict(original['metadata'], scenario=payload.get('scenario', original['scenario']), device_ids=selected) data.update({k: v for k, v in payload.items() if k in ('operator', 'notes')}) - job = self.create(data, snapshot=original['config_snapshot']) - saved = self.store.get(job['id']) - saved['parent_job_id'] = original['id'] - self.store.update(saved) - job['parent_job_id'] = original['id'] - return job + return self.create(data, snapshot=original['config_snapshot'], parent_job_id=original['id']) def insights(self): return self.store.insights() @@ -200,6 +219,7 @@ def close(self): with self.lock: if self.closed: return + self.closing = True self.cancel_event.set() worker = self.worker if worker: diff --git a/inspection/store.py b/inspection/store.py index 3bc964f..1b94438 100644 --- a/inspection/store.py +++ b/inspection/store.py @@ -26,9 +26,9 @@ def __init__(self, path): Path(path).parent.mkdir(parents=True, exist_ok=True) self.lock = threading.RLock() self.connection = sqlite3.connect(str(path), check_same_thread=False, timeout=5) - self.connection.execute('PRAGMA journal_mode=WAL') - self.connection.execute('CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, created_at TEXT NOT NULL, status TEXT NOT NULL, record TEXT NOT NULL)') try: + self.connection.execute('PRAGMA journal_mode=WAL') + self.connection.execute('CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, created_at TEXT NOT NULL, status TEXT NOT NULL, record TEXT NOT NULL)') with self.connection: # sqlite3 does not start a transaction for DDL automatically. self.connection.execute('BEGIN') diff --git a/tests/fixtures/csv-lineage-legacy29.csv b/tests/fixtures/csv-lineage-legacy29.csv new file mode 100644 index 0000000..e4e26ed --- /dev/null +++ b/tests/fixtures/csv-lineage-legacy29.csv @@ -0,0 +1,3 @@ +report_marker,job_id,status,mode,scenario,station_id,batch,operator,notes,config_hash,config_snapshot,created_at,finished_at,device_id,name,connection_type,interface,protocol,scope,verdict,fault_code,summary,suggestion,request,response,duration_ms,simulated,attempts,supporting_checks +SIMULATED,ea2350ff1f8646388db86344c3058dd7,completed,demo,healthy,R5-MOBILE-RELOAD,SIMULATED-MOBILE,Browser QA,"Synthetic only: 一,二 +""quoted"" Unicode evidence",a4d0da170ec831f15649204fabd846b592b5ad155f13967d60039c9705dbf10a,"{""schema_version"": 2, ""workstation_id"": ""DEMO-01"", ""description"": ""SIMULATED acceptance catalog"", ""settings"": {""allow_write_tests"": false}, ""devices"": [{""device_id"": ""DEMO_RELAY"", ""device_name"": ""透明 IO 继电器通信路径"", ""device_type"": ""relay"", ""connection_type"": ""serial"", ""interface"": ""SIM-COM1"", ""serial_config"": {""baudrate"": 9600}, ""protocol"": {""type"": ""modbus_rtu"", ""function_code"": 3, ""address"": 0, ""count"": 2}, ""test_scope"": ""communication_path""}, {""device_id"": ""DEMO_PLC"", ""device_name"": ""PLC 通信寄存器"", ""device_type"": ""plc"", ""connection_type"": ""network"", ""interface"": ""SIM-ETH0"", ""network_config"": {""ip"": ""127.0.0.1"", ""port"": 502}, ""protocol"": {""type"": ""modbus_tcp"", ""function_code"": 3, ""address"": 0, ""count"": 2}}, {""device_id"": ""DEMO_PCI"", ""device_name"": ""PCI 通信卡被动状态"", ""device_type"": ""pci"", ""connection_type"": ""pci"", ""pci_slot"": ""SIM-PCI0"", ""pci_match"": {""vendor"": ""1234""}}]}",2026-10-15T11:45:48.666Z,2026-10-15T11:45:48.701Z,DEMO_RELAY,透明 IO 继电器通信路径,serial,SIM-COM1,modbus_rtu,communication_path,PASS,,SIMULATED Communication path responded,Business and electrical function require separate verification,SIMULATED READ 01 03 00 00 00 02,SIMULATED 01 03 04 00 00 00 00,30,True,[],[] diff --git a/tests/fixtures/csv-lineage-retest.json b/tests/fixtures/csv-lineage-retest.json new file mode 100644 index 0000000..29f6402 --- /dev/null +++ b/tests/fixtures/csv-lineage-retest.json @@ -0,0 +1,130 @@ +{ + "id": "ea2350ff1f8646388db86344c3058dd7", + "status": "completed", + "mode": "demo", + "scenario": "healthy", + "metadata": { + "station_id": "R5-MOBILE-RELOAD", + "batch": "SIMULATED-MOBILE", + "operator": "Browser QA", + "notes": "Synthetic only: 一,二\n\"quoted\" Unicode evidence" + }, + "config_hash": "a4d0da170ec831f15649204fabd846b592b5ad155f13967d60039c9705dbf10a", + "config_snapshot": { + "schema_version": 2, + "workstation_id": "DEMO-01", + "description": "SIMULATED acceptance catalog", + "settings": { + "allow_write_tests": false + }, + "devices": [ + { + "device_id": "DEMO_RELAY", + "device_name": "透明 IO 继电器通信路径", + "device_type": "relay", + "connection_type": "serial", + "interface": "SIM-COM1", + "serial_config": { + "baudrate": 9600 + }, + "protocol": { + "type": "modbus_rtu", + "function_code": 3, + "address": 0, + "count": 2 + }, + "test_scope": "communication_path" + }, + { + "device_id": "DEMO_PLC", + "device_name": "PLC 通信寄存器", + "device_type": "plc", + "connection_type": "network", + "interface": "SIM-ETH0", + "network_config": { + "ip": "127.0.0.1", + "port": 502 + }, + "protocol": { + "type": "modbus_tcp", + "function_code": 3, + "address": 0, + "count": 2 + } + }, + { + "device_id": "DEMO_PCI", + "device_name": "PCI 通信卡被动状态", + "device_type": "pci", + "connection_type": "pci", + "pci_slot": "SIM-PCI0", + "pci_match": { + "vendor": "1234" + } + } + ] + }, + "device_ids": [ + "DEMO_RELAY" + ], + "parent_job_id": "28f70e72a3164ba996e7508c345ff267", + "created_at": "2026-10-15T11:45:48.666Z", + "started_at": "2026-10-15T11:45:48.669Z", + "finished_at": "2026-10-15T11:45:48.701Z", + "duration_ms": 32.41, + "results": [ + { + "device_id": "DEMO_RELAY", + "name": "透明 IO 继电器通信路径", + "connection_type": "serial", + "interface": "SIM-COM1", + "protocol": "modbus_rtu", + "scope": "communication_path", + "verdict": "PASS", + "fault_code": "", + "summary": "SIMULATED Communication path responded", + "suggestion": "Business and electrical function require separate verification", + "request": "SIMULATED READ 01 03 00 00 00 02", + "response": "SIMULATED 01 03 04 00 00 00 00", + "duration_ms": 30, + "simulated": true, + "attempts": [], + "supporting_checks": [] + } + ], + "events": [ + { + "at": "2026-10-15T11:45:48.666Z", + "type": "queued", + "message": "Acceptance queued" + }, + { + "at": "2026-10-15T11:45:48.669Z", + "type": "running", + "message": "Acceptance started" + }, + { + "at": "2026-10-15T11:45:48.701Z", + "type": "result", + "device_id": "DEMO_RELAY", + "message": "PASS" + }, + { + "at": "2026-10-15T11:45:48.701Z", + "type": "completed", + "message": "Acceptance completed" + } + ], + "summary": { + "expected": 1, + "passed": 1, + "failed": 0, + "review": 0, + "not_run": 0, + "verdict": "PASS" + }, + "progress": { + "completed": 1, + "total": 1 + } +} diff --git a/tests/test_csv_lineage.py b/tests/test_csv_lineage.py new file mode 100644 index 0000000..3aa039c --- /dev/null +++ b/tests/test_csv_lineage.py @@ -0,0 +1,106 @@ +"""Append-only CSV provenance for ordinary tasks and retest chains.""" +import copy +import csv +import io +import json +from pathlib import Path + +import pytest + +from inspection.reports import export_job +from inspection.service import InspectionService +from test_http import api # Existing temporary Demo/loopback fixture. +from test_workbench import wait_finished + +FIXTURES = Path(__file__).parent / 'fixtures' + + +@pytest.fixture +def retest_record(): + return json.loads((FIXTURES / 'csv-lineage-retest.json').read_text(encoding='utf-8')) + + +def read_csv(data): + reader = csv.DictReader(io.StringIO(data.decode('utf-8-sig'),newline='')) + return reader.fieldnames,list(reader) + + +def test_csv_appends_parent_column_after_every_existing_column(retest_record): + legacy_columns,_ = read_csv((FIXTURES / 'csv-lineage-legacy29.csv').read_bytes()) + columns,_ = read_csv(export_job(retest_record,'csv')[0]) + assert len(legacy_columns) == 29 + assert columns == legacy_columns + ['parent_job_id'] + + +def test_all_legacy_cells_and_input_evidence_remain_unchanged(retest_record): + original = copy.deepcopy(retest_record) + old_rows = list(csv.reader(io.StringIO((FIXTURES / 'csv-lineage-legacy29.csv').read_text(encoding='utf-8-sig'),newline=''))) + new_rows = list(csv.reader(io.StringIO(export_job(retest_record,'csv')[0].decode('utf-8-sig'),newline=''))) + assert [row[:-1] for row in new_rows] == old_rows + assert retest_record == original + + +def test_parent_identity_is_consistent_across_csv_json_html(retest_record): + _,rows = read_csv(export_job(retest_record,'csv')[0]) + raw = json.loads(export_job(retest_record,'json')[0]) + html = export_job(retest_record,'html',lang='en')[0].decode('utf-8') + assert rows[0].get('parent_job_id') == raw['parent_job_id'] == retest_record['parent_job_id'] + assert raw['parent_job_id'] in html + assert rows[0]['job_id'] == raw['id'] + + +@pytest.mark.parametrize('form',['null','absent']) +def test_ordinary_or_legacy_task_exports_empty_parent(retest_record,form): + if form == 'null': + retest_record['parent_job_id'] = None + else: + del retest_record['parent_job_id'] + columns,rows = read_csv(export_job(retest_record,'csv')[0]) + assert columns[-1] == 'parent_job_id' + assert all(row['parent_job_id'] == '' for row in rows) + + +def test_device_result_cannot_replace_task_level_parent(retest_record): + retest_record['results'][0]['parent_job_id'] = 'SYNTHETIC-DEVICE-RELATION' + _,rows = read_csv(export_job(retest_record,'csv')[0]) + assert rows[0].get('parent_job_id') == retest_record['parent_job_id'] + + +def test_immediate_retest_chain_is_joinable_using_only_csv(tmp_path): + service = InspectionService(tmp_path / 'jobs.db',demo=True) + try: + first = wait_finished(service,service.create({'scenario':'faults','station_id':'SYNTHETIC-CSV-CHAIN'})['id']) + second = wait_finished(service,service.retest(first['id'],{'scenario':'faults'})['id']) + third = wait_finished(service,service.retest(second['id'],{'scenario':'healthy'})['id']) + edges = {} + for task in [first,second,third]: + _,rows = read_csv(export_job(task,'csv')[0]) + assert all('parent_job_id' in row for row in rows) + assert len({row['parent_job_id'] for row in rows}) == 1 + edges[rows[0]['job_id']] = rows[0]['parent_job_id'] + assert edges == {first['id']:'',second['id']:first['id'],third['id']:second['id']} + assert service.get(first['id']) == first and service.get(second['id']) == second + finally: + service.close() + + +def test_csv_provenance_is_independent_of_ui_language(retest_record): + chinese = export_job(retest_record,'csv',lang='zh')[0] + english = export_job(retest_record,'csv',lang='en')[0] + assert chinese == english + assert read_csv(chinese)[1][0].get('parent_job_id') == retest_record['parent_job_id'] + + +def test_loopback_export_route_preserves_actual_persisted_parent(api): + service,server,request = api + status,body = request('POST','/api/jobs',{'scenario':'faults','station_id':'SYNTHETIC-CSV-HTTP'}) + assert status == 202 + first = wait_finished(service,body['job']['id']) + status,body = request('POST','/api/jobs/' + first['id'] + '/retest',{'scenario':'healthy'}) + assert status == 202 + second = wait_finished(service,body['job']['id']) + status,content = request('GET','/api/jobs/' + second['id'] + '/export?format=csv&lang=en') + assert status == 200 + _,rows = read_csv(content) + assert {row.get('parent_job_id') for row in rows} == {first['id']} + assert service.get(first['id']) == first diff --git a/tests/test_hierarchy_boundary.py b/tests/test_hierarchy_boundary.py new file mode 100644 index 0000000..f66599a --- /dev/null +++ b/tests/test_hierarchy_boundary.py @@ -0,0 +1,108 @@ +"""Unsupported catalog nodes must not disappear from acceptance denominators.""" +import copy +import threading + +import pytest + +from core.config_manager import StandardDeviceConfig +from inspection.adapters import demo_config +from inspection.domain import catalog_devices, initial_result +from inspection.http import make_server +from inspection.service import InspectionService +from test_http import requester + + +def device_tree(depth=2, empty_children=None): + snapshot = demo_config() + parent = copy.deepcopy(snapshot['devices'][1]) + parent.update(device_id='SYNTH_PARENT', device_name='Synthetic parent', interface='') + node = parent + for index in range(1, depth + 1): + child = copy.deepcopy(snapshot['devices'][1]) + child.update(device_id='SYNTH_CHILD_' + str(index), device_name='Synthetic child ' + str(index), interface='') + node['child_devices'] = [child] + node = child + if empty_children is not None: + node['child_devices'] = empty_children + snapshot['devices'] = [parent] + return snapshot + + +class AlwaysPassFake: + def __init__(self): + self.calls = 0 + + def run(self, config, device_ids, scenario, cancelled, on_result): + self.calls += 1 + for device in catalog_devices(config): + if device['device_id'] in device_ids: + result = initial_result(device, True) + result.update(verdict='PASS', fault_code='', summary='Synthetic Fake response') + on_result(result) + + +@pytest.mark.parametrize('depth', [2, 3]) +def test_configuration_rejects_children_beyond_the_supported_execution_depth(depth): + with pytest.raises(ValueError): + StandardDeviceConfig(data=device_tree(depth)) + + +@pytest.mark.parametrize('demo', [False, True]) +def test_unsupported_tree_never_creates_record_or_calls_adapter(tmp_path, demo): + adapter = AlwaysPassFake() + service = InspectionService(tmp_path / 'jobs.db', demo=demo, adapter=adapter) + try: + with pytest.raises(ValueError): + service.create({'station_id':'SYNTHETIC-ONLY'}, snapshot=device_tree()) + assert service.list_jobs()['total'] == 0 + assert adapter.calls == 0 and service.active_id is None + assert not service.hardware_lock.locked() + finally: + service.close() + + +@pytest.fixture +def hierarchy_api(tmp_path, monkeypatch): + adapter = AlwaysPassFake() + service = InspectionService(tmp_path / 'jobs.db', demo=True, adapter=adapter) + monkeypatch.setattr(service, '_snapshot', device_tree) + server = make_server('127.0.0.1', 0, service) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield service, adapter, requester(server) + finally: + server.shutdown() + server.server_close() + thread.join(2) + service.close() + + +@pytest.mark.parametrize('endpoint,payload', [ + ('/api/config/validate', {'config':device_tree()}), + ('/api/jobs', {'station_id':'SYNTHETIC-ONLY'}), +]) +def test_http_rejects_unsupported_tree_before_reporting_valid_or_starting_job(hierarchy_api, endpoint, payload): + service, adapter, request = hierarchy_api + status, body = request('POST', endpoint, payload) + assert status == 400 and body['ok'] is False + assert service.list_jobs()['total'] == 0 and adapter.calls == 0 + + +@pytest.mark.parametrize('empty_children', [None, [], ()]) +def test_supported_parent_child_still_runs_with_complete_results(tmp_path, empty_children): + # None means omitted; [] is the documented empty form. Existing tuples in + # trusted in-process configurations are retained, not a new JSON feature. + snapshot = device_tree(1, empty_children) + adapter = AlwaysPassFake() + service = InspectionService(tmp_path / 'jobs.db', demo=True, adapter=adapter) + try: + job = service.create({}, snapshot=snapshot) + service.worker.join(2) + finished = service.get(job['id']) + assert finished['status'] == 'completed' + assert finished['summary']['expected'] == finished['summary']['passed'] == 2 + assert {result['device_id'] for result in finished['results']} == {'SYNTH_PARENT','SYNTH_CHILD_1'} + assert adapter.calls == 1 + finally: + service.close() diff --git a/tests/test_protocol_boundaries.py b/tests/test_protocol_boundaries.py new file mode 100644 index 0000000..c67ab93 --- /dev/null +++ b/tests/test_protocol_boundaries.py @@ -0,0 +1,101 @@ +"""Pure byte/config regressions; never create transports or contact devices.""" +import copy + +import pytest + +from core.config_manager import StandardDeviceConfig +from core.protocol_engine import ( + ProtocolConfigError, append_checksum, build_modbus_rtu_read, + build_modbus_tcp_read, match_response, response_rule, +) +from inspection.adapters import demo_config + + +@pytest.mark.parametrize('guard', [ + {'min_length': 4}, {'max_length': 1}, {'length': 4}, + {'exact_hex': '00 00'}, {'prefix_hex': '00'}, {'suffix_hex': '00'}, + {'contains_hex': '00'}, {'exact_text': 'NO'}, {'contains_text': 'NO'}, + {'regex_text': '^NO$'}, {'regex_hex': '^00'}, {'checksum': 'sum8'}, +]) +def test_one_of_cannot_bypass_other_response_conditions(guard): + rule=dict(guard, one_of=[{'exact_text':'OK'}, {'exact_text':'READY'}]) + assert not match_response(b'OK', rule)[0], guard + + +def test_parent_checksum_still_applies_after_probe_rule_merge(): + protocol={'response_match':{'checksum':'crc16_modbus'}} + probe={'response_match':{'one_of':[{'prefix_hex':'01 03'}, {'prefix_hex':'01 04'}]}} + rule=response_rule(protocol,probe) + valid=append_checksum(bytes.fromhex('01 03 02 00 2A'),'crc16_modbus') + assert match_response(valid,rule)[0] + assert not match_response(valid[:-1]+bytes([valid[-1]^1]),rule)[0] + + +@pytest.mark.parametrize('function,limit', [(1,2000),(2,2000),(3,125),(4,125)]) +@pytest.mark.parametrize('transport', ['rtu','tcp']) +def test_modbus_request_quantity_boundary(function,limit,transport): + build=(lambda count:build_modbus_rtu_read(1,function,0,count)) if transport=='rtu' else (lambda count:build_modbus_tcp_read(1,1,function,0,count)) + assert build(limit) + for count in [0,-1,limit+1,65536]: + with pytest.raises(ProtocolConfigError): + build(count) + + +@pytest.mark.parametrize('transport', ['rtu','tcp']) +@pytest.mark.parametrize('address,count', [(-1,1),(65536,1),(65535,2),(65534,3)]) +def test_request_never_wraps_an_address_or_requested_span(transport,address,count): + with pytest.raises(ProtocolConfigError): + if transport=='rtu':build_modbus_rtu_read(1,3,address,count) + else:build_modbus_tcp_read(1,1,3,address,count) + + +@pytest.mark.parametrize('function,limit', [(1,2000),(2,2000),(3,125),(4,125)]) +@pytest.mark.parametrize('kind', ['modbus_rtu','modbus_tcp']) +def test_config_rejects_function_specific_quantity_and_address_overrun(function,limit,kind): + for address,count in [(0,limit+1),(0,0),(65535,2)]: + config=demo_config() + index=0 if kind=='modbus_rtu' else 1 + config['devices'][index]['protocol']={'type':kind,'function_code':function,'address':address,'count':count} + with pytest.raises(ValueError):StandardDeviceConfig(data=config) + + +@pytest.mark.parametrize('function,count,address',[(1,2000,63536),(2,2000,63536),(3,125,65411),(4,125,65411),(3,1,65535)]) +def test_last_valid_address_span_stays_allowed(function,count,address): + for kind in ['modbus_rtu','modbus_tcp']: + config=demo_config() + index=0 if kind=='modbus_rtu' else 1 + config['devices'][index]['protocol']={'type':kind,'function_code':function,'address':address,'count':count} + assert StandardDeviceConfig(data=config) + + +def test_response_conditions_are_pure_and_conjunctive_on_valid_data(): + packet=append_checksum(b'OK','sum8') + rule={'one_of':[{'prefix_hex':'4F'},{'exact_text':'NO'}], 'prefix_hex':'4F 4B', 'min_length':3,'max_length':3,'checksum':'sum8'} + before=copy.deepcopy(rule) + assert match_response(packet,rule)[0] + assert rule==before + assert not match_response(packet,dict(rule,one_of=[{'exact_text':'NO'}]))[0] + + +@pytest.mark.parametrize('field,value', [ + ('function_code',True),('function_code',3.9),('address',True), + ('address',1.9),('count',True),('count',1.9),('count',float('inf')), +]) +def test_protocol_integer_config_never_truncates_or_accepts_bool(field,value): + config=demo_config() + config['devices'][0]['protocol'][field]=value + with pytest.raises(ValueError):StandardDeviceConfig(data=config) + + +@pytest.mark.parametrize('timeout', [float('nan'),'nan']) +def test_serial_timeout_must_be_finite(timeout): + config=demo_config() + config['devices'][0]['serial_config']['timeout']=timeout + with pytest.raises(ValueError):StandardDeviceConfig(data=config) + + +def test_numeric_string_and_exact_integral_float_compatibility(): + for values in [{'function_code':'3','address':'0','count':'2'}, {'function_code':3.0,'address':0.0,'count':2.0}]: + config=demo_config() + config['devices'][0]['protocol'].update(values) + assert StandardDeviceConfig(data=config) diff --git a/tests/test_rejected_post_delivery.py b/tests/test_rejected_post_delivery.py new file mode 100644 index 0000000..acc70d4 --- /dev/null +++ b/tests/test_rejected_post_delivery.py @@ -0,0 +1,106 @@ +"""Event-controlled real loopback teardown, no devices or external transport.""" +import http.client +import json +import socket +import threading + +import pytest + +from inspection.http import Handler, HttpError, MAX_BODY, make_server +from inspection.service import InspectionService + + +@pytest.mark.parametrize('headers,expected', [ + ('Authorization: Bearer wrong\r\n',401), + ('Authorization: Bearer fixture\r\nOrigin: https://foreign.invalid\r\n',403), +]) +def test_rejected_post_retains_error_while_draining_already_sent_body(tmp_path,monkeypatch,headers,expected): + admitted=threading.Event() + body_sent=threading.Event() + finished=threading.Event() + original_access=Handler._access + def gated_access(handler,path): + try: + return original_access(handler,path) + except HttpError: + admitted.set() + assert body_sent.wait(2),'test never sent rejected request body' + raise + monkeypatch.setattr(Handler,'_access',gated_access) + service=InspectionService(tmp_path/'jobs.db',demo=True) + server=make_server('127.0.0.1',0,service,token='fixture') + get_request=server.get_request + discarded=[] + class TrackedSocket: + def __init__(self,raw):self.raw=raw + def __getattr__(self,name):return getattr(self.raw,name) + def recv(self,size,*args): + data=self.raw.recv(size,*args) + discarded.append(data) + return data + def shutdown(self,how):return self.raw.shutdown(how) + def tracked_request(): + raw,address=get_request() + return TrackedSocket(raw),address + server.get_request=tracked_request + original_close=server.close_request + def close_request(request): + try:return original_close(request) + finally:finished.set() + server.close_request=close_request + thread=threading.Thread(target=server.serve_forever,daemon=True) + thread.start() + body=b' '*(32*1024) + try: + with socket.create_connection(server.server_address,timeout=2) as client: + request=('POST /api/jobs HTTP/1.1\r\nHost: 127.0.0.1:'+str(server.server_port)+'\r\nContent-Length: '+str(len(body))+'\r\n'+headers+'\r\n') + client.sendall(request.encode('ascii')) + assert admitted.wait(2),'server did not reach header admission' + # Header parser has already stopped: these bytes remain in TCP, + # not in a speculative BufferedReader header read. + client.sendall(body) + body_sent.set() + response=http.client.HTTPResponse(client) + response.begin() + assert response.status==expected + assert json.loads(response.read())['ok'] is False + assert finished.wait(2),'server did not finish bounded teardown' + assert b''.join(discarded)==body,'unread rejected body was closed without bounded drain' + assert service.list_jobs()['total']==0 + finally: + body_sent.set() + server.shutdown() + server.server_close() + thread.join(2) + service.close() + + +def test_discard_is_byte_bounded_even_for_unending_input(): + consumed=[] + class Socket: + def shutdown(self,how):assert how==socket.SHUT_WR + def settimeout(self,value):assert 0 Date: Thu, 15 Oct 2026 20:33:31 +0800 Subject: [PATCH 2/2] test: bound legacy HTTP helper and verify auth failure evidence --- CHANGELOG.md | 2 +- docs/VALIDATION.md | 4 ++++ tests/test_backend_fixes.py | 31 +++++++++++++++++++++++++------ 3 files changed, 30 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 075c3e1..a1e9100 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ - Apply shared response constraints to `one_of` candidates, enforce read-function-specific Modbus quantities/address spans and reject non-finite or non-integral settings. - Reject unsupported device descendants beyond one child level instead of silently reducing the expected-device count. - Append `parent_job_id` to CSV exports as column 30. Existing 29 columns keep their order and values; consumers requiring exactly 29 columns need updating. -- Add 113 automated regressions above the 100-test public baseline. The isolated Windows candidate passes 213 tests, including combined worker-start/storage failure; remote matrix outcomes are recorded separately in PR checks. These changes and tests were developed with AI assistance. +- Add 114 automated regressions above the 100-test public baseline. The isolated Windows candidate passes 214 tests, including combined worker-start/storage failure and a negative authentication-harness control; remote matrix outcomes are recorded separately in PR checks. These changes and tests were developed with AI assistance. ## 3.1 — 2026-10-08 diff --git a/docs/VALIDATION.md b/docs/VALIDATION.md index 6a47643..24e468f 100644 --- a/docs/VALIDATION.md +++ b/docs/VALIDATION.md @@ -65,6 +65,10 @@ Pre-merge review found and reproduced one additional compound failure: a worker ## Scope and limits +The first PR matrix at candidate `17b3f4a` passed three jobs but its Windows/Python 3.13 job timed out waiting eight seconds for a preexisting Node helper subprocess (212 other tests passed). The same candidate's separate push matrix passed all four jobs in [run 38734469352](https://github.com/KaiserIIII/QLDeviceCheck_Generic_WebUI_Linux/actions/runs/38734469352); the failed PR run is retained as [run 38734503505](https://github.com/KaiserIIII/QLDeviceCheck_Generic_WebUI_Linux/actions/runs/38734503505). Those logs do not establish where the delay occurred or prove a product/runtime defect. + +The test harness now gives each real fetch a three-second deadline, requires response statuses `[200,200,401]` and unauthorized JSON `ok:false`, emits request/completion evidence, and allows a separate 20-second subprocess lifecycle budget. A server-token mismatch must fail with the specific auth error and no success receipt. It uses natural Node shutdown. The final Git-tree export passed **214 tests in 36.00 seconds** (including both harness controls), with all 23 self-checks, configuration validation and compilation still passing. Runtime coverage is unchanged from the preceding 213-test result. Inspect final PR checks for the new head's cross-platform outcome; the earlier timeout is not hidden or counted as a production bug. + Coverage is collected for `inspection/`, `core.protocol_engine` and `core.transport`, not the entire legacy detector or repository. The original offline self-test adds loopback checks but does not establish field hardware compatibility. Browser tests are separate from the Python CI job and use an existing local browser runtime. This automated suite does not measure serial driver behavior, electrical performance, sustained station load, power-loss filesystems or custom vendor protocols. The earlier usage-test confirmation is separate from these automated results. OS-level device exclusivity across different station data directories is not provided. diff --git a/tests/test_backend_fixes.py b/tests/test_backend_fixes.py index 4712679..02f88a9 100644 --- a/tests/test_backend_fixes.py +++ b/tests/test_backend_fixes.py @@ -262,7 +262,8 @@ def test_nested_malformed_configuration_returns_validation_error(api, key, value assert request('POST', '/api/config/validate', {'config': config})[0] == 400 -def test_legacy_page_helper_uses_session_bearer_for_actual_http(api, monkeypatch, tmp_path): +@pytest.mark.parametrize('server_token,expected_code', [('session-secret', 0), ('different-secret', 1)]) +def test_legacy_page_helper_uses_session_bearer_for_actual_http(api, monkeypatch, tmp_path, server_token, expected_code): import legacy_web svc, server, request = api svc.demo = False @@ -270,7 +271,7 @@ def test_legacy_page_helper_uses_session_bearer_for_actual_http(api, monkeypatch monkeypatch.setattr(svc, '_snapshot', demo_config) monkeypatch.setattr(legacy_web, 'scan_catalog', lambda *args, **kwargs: {'ok': True}) monkeypatch.setattr(legacy_web, 'test_single_device', lambda *args, **kwargs: {'ok': True}) - server.access_token = 'session-secret' + server.access_token = server_token script = request('GET', '/assets/i18n.js')[1].decode() + '\n' + re.search(r'', request('GET', '/legacy')[1].decode()).group(1) node = shutil.which('node') if not node: @@ -283,9 +284,18 @@ def test_legacy_page_helper_uses_session_bearer_for_actual_http(api, monkeypatch const script = fs.readFileSync(process.argv[1], 'utf8'); const base = process.argv[2]; const values = new Map([['qldc.token', 'session-secret']]); +const statuses = []; +let unauthorized = false; const element = {addEventListener(){}, querySelectorAll(){return []}}; const document = {documentElement:{dataset:{}},querySelector:()=>element,querySelectorAll:()=>[],getElementById(){return element}, addEventListener(){}}; -const context = {document, sessionStorage:{getItem:k=>values.get(k),setItem:(k,v)=>values.set(k,v)}, fetch:(url,opts)=>{if(url.includes('secret'))throw Error('credential in URL');return fetch(base+url,opts)}}; +const context = {document, sessionStorage:{getItem:k=>values.get(k),setItem:(k,v)=>values.set(k,v)}, fetch:async(url,opts)=>{ + if(url.includes('secret'))throw Error('credential in URL'); + process.stderr.write('request '+url+'\\n'); + const response=await fetch(base+url,{...opts,signal:AbortSignal.timeout(3000)}); + statuses.push(response.status); + if(response.status===401)unauthorized=(await response.clone().json()).ok===false; + return response; +}}; vm.createContext(context); vm.runInContext(script, context); (async()=>{ @@ -294,10 +304,19 @@ def test_legacy_page_helper_uses_session_bearer_for_actual_http(api, monkeypatch let rejected=false; try{await vm.runInContext("postJson('/api/scan')",context)}catch{rejected=true} if(!rejected)throw Error('wrong token accepted'); -})().catch(e=>{console.error(e.message);process.exitCode=1}); + if(JSON.stringify(statuses)!=='[200,200,401]'||!unauthorized)throw Error('unexpected response evidence'); + // Emit proof that all awaited HTTP/helper assertions completed. Allow Node + // to release its request resources naturally rather than force-exiting. + process.stdout.write(JSON.stringify({statuses,unauthorized})+'\\n'); +})().catch(e=>{process.stderr.write(e.message+'\\n');process.exitCode=1}); """ - result = subprocess.run([node, '-e', code, str(script_path), 'http://127.0.0.1:%d' % server.server_port], capture_output=True, text=True, timeout=8) - assert result.returncode == 0, result.stderr + result = subprocess.run([node, '-e', code, str(script_path), 'http://127.0.0.1:%d' % server.server_port], capture_output=True, text=True, timeout=20) + assert result.returncode == expected_code, result.stderr + if expected_code == 0: + assert json.loads(result.stdout) == {'statuses': [200, 200, 401], 'unauthorized': True} + else: + assert 'Access token required' in result.stderr + assert not result.stdout, 'a server-token mismatch must not produce a success receipt' def test_history_sort_has_stable_ties_and_pagination_over_http(api):