diff --git a/.agents/skills/nexent-test-assets/references/runtime-migration.md b/.agents/skills/nexent-test-assets/references/runtime-migration.md index 4fa664a0f0..d0077074af 100644 --- a/.agents/skills/nexent-test-assets/references/runtime-migration.md +++ b/.agents/skills/nexent-test-assets/references/runtime-migration.md @@ -6,8 +6,8 @@ The repository owns test behavior and prerequisites. Put explicit asset families Machine-local `config/pipeline.yaml` owns only timeouts, notification opt-in and explicit build/deploy/preflight argv hooks. It must not become another source of Case definitions or preparation logic. Secrets, fixed resource IDs and local paths stay out of Case scripts. New settings require a schema and safe example. Existing helpers still load users, models, assets and secrets from the selected test home. -The candidate runner must account for every selected Case, distinguish blocked/retired/policy-skipped outcomes from PASS, finalize local reports after interruption, and preserve per-case evidence. A failing independent Case does not prevent unrelated Cases from running. Pending recovery stops business Cases until environment safety is verified. The next scheduled Daily may restore a validated Local configuration backup, synchronize develop and redeploy under the Daily/checkout locks; it must preserve the old evidence and pass service/mode and Local recovery checks before ordinary Cases. Active owners/locks, unsafe or damaged journals/configuration and pending D5 fault recovery remain hard stops. Targeted runs/resume do not bypass recovery. See [daily-validation.md](../../../../test-e2e/infra/tools/docs/daily-validation.md#retrying-a-failed-deployment). +The candidate runner must account for every selected Case, distinguish blocked/retired/policy-skipped outcomes from PASS, finalize local reports after interruption, and preserve per-case evidence. A failing independent Case does not prevent unrelated Cases from running. Common configuration and runtime safety may stop a batch; unavailable local toolchains and known-Case execution declarations block only affected Cases. Doctor checks common readiness by default; explicit doctor --full includes dependency diagnostics, not an all-or-nothing business prerequisite. Pending recovery stops business Cases until environment safety is verified. The next scheduled Daily may restore a validated Local configuration backup, synchronize develop and redeploy under the Daily/checkout locks; it must preserve the old evidence and pass service/mode checks before ordinary Cases. Business recovery Cases and historical Case-result files do not gate that redeploy. Active owners/locks, unsafe configuration backups and pending D5 fault recovery remain hard stops. Targeted runs/resume do not bypass recovery. See [daily-validation.md](../../../../test-e2e/infra/tools/docs/daily-validation.md#retrying-a-failed-deployment). -Daily never generates candidates, edits Excel or YAML, refreshes manifests, or promotes test assets. Update tests on the product branch. Generated registry/Excel remain read-only views during execution. Normal notifications require completed D6 and an accounted execution plan. Uncertain delivery is not automatically retried. +Daily never generates candidates, edits Excel or YAML, refreshes manifests, or promotes test assets. Update tests on the product branch. Generated registry/Excel remain read-only views during execution. Enabled terminal notifications cover success, failure and catchable interruption, including preparation failure; complete shared-latest publication is a separate decision. Uncertain delivery is not automatically retried. Before claiming runtime migration complete, verify: locked Python/Node dependency closure, controlled HTTP/MCP services, per-case preparation declarations and behavior, isolated deployment, timeout cleanup/recovery, same-version resume, D1-D5 live runs, failure evidence and D6/report/notification parity. Consult `test-e2e/infra/migration/runtime-migration-status.md` for current gaps rather than assuming they are implemented. diff --git a/test-e2e/DEVELOPER_GUIDE.zh-CN.md b/test-e2e/DEVELOPER_GUIDE.zh-CN.md index 2845c575e8..1e8b7ee348 100644 --- a/test-e2e/DEVELOPER_GUIDE.zh-CN.md +++ b/test-e2e/DEVELOPER_GUIDE.zh-CN.md @@ -87,9 +87,11 @@ Linux 的 Chromium 系统库需要另外由主机环境提供。依赖安装和 使用 GitHub Daily 时，若本机 `daily.env` 显式设置了 `NEXENT_REPO`，它必须与 `.github/workflows/nexent-daily-full-test.yml` 中的 `REPO_ROOT` 指向同一个产品 Git 仓库。迁移仓库路径时不能只改其中一个，还需核对启动入口和 SDK 引用，避免从一个仓库启动、却加载另一个仓库的代码。工作流修改只有提交并在实际运行的分支生效后才算完成入口切换；此前不要恢复可能混用新旧入口的 Runner。普通本地命令仍应从实际产品仓库根目录执行，不能仅靠修改该环境变量替代切换工作目录。 -普通部署失败留下恢复标记后，下一次 GitHub Daily 可以在持有 Daily 和仓库部署锁、确认旧进程已退出及旧批次终态、核验原始备份和 Local 配置后，保存旧证据并恢复配置，再同步最新 develop、构建部署。标记不会因允许同步而直接清除：服务预检、实际部署模式和 Local 恢复用例必须通过，才能开始普通用例；最终恢复验证完成后才解除本次标记。损坏备份、非预期配置改动、活跃锁、D5 故障恢复标记或未确认的 D5 超时/清理失败仍会阻断。普通定向执行和 resume 不具备这个自动恢复权限。详见[Daily 恢复流程](infra/tools/docs/daily-validation.md#retrying-a-failed-deployment)。 +普通部署失败留下恢复标记后，下一次 GitHub Daily 可以在持有 Daily 和仓库部署锁、确认旧进程已退出及旧批次终态、核验原始备份和 Local 配置后，保存旧证据并恢复配置，再同步最新 develop、构建部署。标记不会因允许同步而直接清除：服务预检和实际部署模式检查通过后开始普通用例；结束后同样按环境检查结果解除恢复标记。知识库等业务用例只在正常分组执行一次，失败不再阻断其他独立用例。损坏备份、非预期配置改动、活跃锁、D5 故障恢复标记或活跃故障操作锁仍会阻断。普通定向执行和 resume 不具备这个自动恢复权限。详见[Daily 恢复流程](infra/tools/docs/daily-validation.md#retrying-a-failed-deployment)。 -启用通知的 Daily 在成功、失败或可捕获中断时均发送终态报告；同步、资产校验、doctor 阻断也生成启动异常通知。报告注明未完成原因及环境恢复状态，配置还原不等于恢复用例通过。启动与执行批次通过显式关联共用发送回执，`SENDING`/`UNKNOWN` 不盲目重发。未完成批次的通知不会覆盖正式共享 latest，也不会改变用例结果或解除恢复标记。强杀、断电或发送通道不可用时不能保证进程内通知到达，需另行监控兜底。详见[终态通知](infra/tools/docs/daily-validation.md#terminal-reports-and-notifications)。 +启用通知的 Daily 在成功、失败或可捕获中断时均发送终态报告；同步、资产校验、doctor 阻断也生成启动异常通知。报告注明未完成原因及环境恢复状态，配置还原不等于环境检查通过。启动与执行批次通过显式关联共用发送回执，`SENDING`/`UNKNOWN` 不盲目重发。未完成批次的通知不会覆盖正式共享 latest，也不会改变用例结果或解除恢复标记。强杀、断电或发送通道不可用时不能保证进程内通知到达，需另行监控兜底。详见[终态通知](infra/tools/docs/daily-validation.md#terminal-reports-and-notifications)。 + +正常通知直接发送本批次的正式 `report.md`。报告生成时统一写入终态、未完成原因及环境恢复状态，并脱敏；已有报告在发送前补齐最新终态并核对脱敏。仅在正式报告缺失、为空、不可读取或无法完成发送准备时，才生成 `notification-report.md` 异常通知，并明确详细报告不可用。正常报告不再复制一份通知报告；原始结果、日志、截图及 Trace 仍独立保留。 专用 Ubuntu 测试机迁移后的目录分工：产品 Git 仓库是 `/home/jason/projects/nexent`，test-home 仍为 `/home/jason/nexent-test-suite`。旧 `/root/project/nexent` 保留用于回退，不再作为新入口；`/home/jason/nexent` 是容器工作目录，不是新的产品 Git 仓库，不能覆盖或删除。配置、历史结果、浏览器缓存及持久化数据均保留在原 test-home。专用机的本地命令应先执行 `cd /home/jason/projects/nexent`；其他开发者使用自己的仓库位置，不把上述机器路径写进通用 Case。 @@ -107,7 +109,11 @@ Windows Docker Desktop 的容器访问宿主机通常使用 `host.docker.interna uv pip check --python $TestPython ``` -遇到缺失资产或校验冲突时，按命令结果处理；不要拿空文件代替。`doctor` 按所选用例检查实际 jsdom/CSS 依赖导入及对应工具链的无头浏览器启动，`--live` 另外检查服务可达性；这些检查不代表所有账号、模型、业务流程已经通过。离线用例可省略 `--live`，并用 `--case ` 缩小检查范围。 +遇到缺失资产或校验冲突时，按命令结果处理；不要拿空文件代替。`doctor` 默认只检查公共环境，`--live` 另外检查服务可达性；需要完整诊断时加 `--full`，检查所选用例的实际 jsdom/CSS 依赖导入及对应工具链的无头浏览器启动。这些检查不代表所有账号、模型、业务流程已经通过。离线用例可省略 `--live`，并用 `--case ` 缩小检查范围。 + +完整诊断的全部检查不作为整批执行的前置阻断。运行前只检查公共配置、专用 Python 环境和必要的部署条件；浏览器或 jsdom 工具链不可用时，仅将依赖该工具链的用例标记为 BLOCKED，其他独立用例继续。账号、模型及自定义脚本依赖由实际使用它们的用例检查。已知用例的执行声明错误也只阻断该用例，严格开发校验仍要求修正。并发锁、真实 D5 故障恢复标记、危险路径和执行中源码或配置变更仍会停止整批。旧批次业务结果文件缺失不再阻断已核验备份的重新部署；旧证据不会被清除。 + +同步 develop 与执行测试使用独立 Python 进程：启动器持续持有 Daily 和仓库部署锁，新执行进程校验同步后的提交、工作区及继承锁，再重新核验恢复上下文。恢复工具不再提前导入部署执行器；异常退出仍保留恢复标记和原始报告。详见 [Daily 进程隔离](infra/tools/docs/daily-validation.md#fresh-execution-after-synchronization)。 ## 3. 收到失败结果后如何修改 diff --git a/test-e2e/infra/config/deployment-exceptions.schema.json b/test-e2e/infra/config/deployment-exceptions.schema.json index 75716a8170..ac6e49d8f8 100644 --- a/test-e2e/infra/config/deployment-exceptions.schema.json +++ b/test-e2e/infra/config/deployment-exceptions.schema.json @@ -1,7 +1,7 @@ { "$schema": "https://json-schema.org/draft/2020-12/schema", "type": "object", "additionalProperties": false, - "required": ["schema_version", "profiles", "cases", "recovery_cases"], + "required": ["schema_version", "profiles", "cases"], "properties": { "schema_version": {"const": 1}, "profiles": {"type": "object", "minProperties": 1, @@ -21,7 +21,7 @@ } }, "cases": {"type": "object", "additionalProperties": {"type": "string"}}, - "recovery_cases": {"type": "array", "minItems": 1, "uniqueItems": true, + "recovery_cases": {"type": "array", "deprecated": true, "uniqueItems": true, "items": {"type": "string", "minLength": 1}} } } diff --git a/test-e2e/infra/config/deployment-exceptions.yaml b/test-e2e/infra/config/deployment-exceptions.yaml index bd0b2cafd9..b862b56bd1 100644 --- a/test-e2e/infra/config/deployment-exceptions.yaml +++ b/test-e2e/infra/config/deployment-exceptions.yaml @@ -19,5 +19,3 @@ cases: API-AUTO-FECE331C989B771F: aidp AGT-AUTO-7A2120E630C381C7: aidp API-AUTO-7513FEF9904B9F3F: aidp -recovery_cases: - - PW-KB-LOCAL-01 diff --git a/test-e2e/infra/scripts/daily-launch.py b/test-e2e/infra/scripts/daily-launch.py index 8841fd6146..a56c720948 100644 --- a/test-e2e/infra/scripts/daily-launch.py +++ b/test-e2e/infra/scripts/daily-launch.py @@ -8,15 +8,15 @@ import os from pathlib import Path import signal -import subprocess import sys import uuid TOOLS = Path(__file__).resolve().parents[1] / "tools" sys.path.insert(0, str(TOOLS)) -from suite_runtime import BatchLock, deployment_lock, logged, now, save +from suite_runtime import BatchLock, deployment_lock, now, save from daily_git import run_git from daily_recovery import prepare_daily_retry +from daily_process import notify_terminal, run_worker def git(repo, *args, log_dir=None): @@ -44,7 +44,8 @@ def checked(*args): def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--test-home", type=Path, required=True) - parser.add_argument("--branch", required=True, choices=("djb/test-baseline", "develop")) + parser.add_argument("--branch", required=True, + help="Explicit remote branch, validated and fast-forwarded under the Daily lock") parser.add_argument("--execute", action="store_true") parser.add_argument("--publish-latest", action="store_true", help="Publish complete Daily logs to test-home/runs-ubuntu; default disabled") @@ -67,39 +68,23 @@ def main(argv=None): save(directory / "status.json", state) old_handler = signal.signal(signal.SIGTERM, lambda *_: (_ for _ in ()).throw(KeyboardInterrupt())) - def record_batch(batch): - state["batch"] = str(batch) - save(directory / "status.json", state) - try: - with BatchLock(home / "state/daily.lock"): + with BatchLock(home / "state/daily.lock") as daily_lock: try: - with deployment_lock(repo), prepare_daily_retry(repo, home, directory): + with deployment_lock(repo) as checkout_lock, prepare_daily_retry(repo, home, directory): state["head"] = sync_checkout(repo, args.branch, log_dir=directory / "git") - env = os.environ.copy() - # Daily reports local asset problems without rewriting tracked views. - command = [str(runtime), str(repo / "test-e2e/infra/tools/validate_test_assets.py"), - "--phase", "implementation", "--daily-report", str(directory / "asset-validation.json")] - if logged(command, repo, env, directory / "validate.log", 180): - raise RuntimeError("Formal asset validation failed; inspect local launcher log") - if git(repo, "status", "--porcelain", "--untracked-files=normal", log_dir=directory / "git"): - raise RuntimeError("Derived generation changed tracked assets; repair them on the product branch") - # Import only after checkout synchronization; use this exact revision's code. - import static_assets - import suite - static_assets.apply(home, repo / "test-e2e/infra/assets") - records = suite.selection(repo, argparse.Namespace(case=None, stage=None, feature=None, change=None), daily=True) - problems = suite.doctor(repo, home, records) - if problems: - save(directory / "doctor.json", {"ready": False, "problems": problems}) - raise RuntimeError("Runtime preflight failed; inspect local doctor.json") - settings = suite.config(repo, home) - state["status"] = "EXECUTING" save(directory / "status.json", state) - code = suite.execute(repo, home, records, settings, daily=True, publish_latest=args.publish_latest, - on_batch_created=record_batch) - state["status"] = "PASSED" if code == 0 else "FAILED" - return code + code = run_worker(repo, home, directory, state["head"], args.publish_latest, + [daily_lock, checkout_lock]) + result = json.loads((directory / "status.json").read_text(encoding="utf-8")) + state.update(result, worker_exit_code=code) + if (state.get("status") not in {"PASSED", "FAILED", "INCOMPLETE", "INTERRUPTED"} + or (code == 0 and state.get("status") != "PASSED") + or (code != 0 and state.get("status") == "PASSED")): + state.update(status="INCOMPLETE", + error=f"Execution process exited with code {code} without a matching terminal receipt") + return 2 + return code if code >= 0 else 2 except (Exception, KeyboardInterrupt) as exc: state.update(status="INTERRUPTED" if isinstance(exc, KeyboardInterrupt) else "INCOMPLETE", error="Interrupted" if isinstance(exc, KeyboardInterrupt) else @@ -115,11 +100,17 @@ def record_batch(batch): return 3 if isinstance(exc, RuntimeError) else 2 finally: signal.signal(signal.SIGTERM, old_handler) + # Preserve any batch link already written by an interrupted executor. + try: + persisted = json.loads((directory / "status.json").read_text(encoding="utf-8")) + if persisted.get("batch"): + state["batch"] = persisted["batch"] + except (OSError, ValueError, AttributeError): + pass state["finished_at"] = now() save(directory / "status.json", state) print(f"Daily launch: {directory}; status={state['status']}") - from suite_notify import notify_launch - notify_launch(directory, home) + notify_terminal(repo, home, directory) if __name__ == "__main__": diff --git a/test-e2e/infra/scripts/daily-worker.py b/test-e2e/infra/scripts/daily-worker.py new file mode 100644 index 0000000000..b0c8f35f5a --- /dev/null +++ b/test-e2e/infra/scripts/daily-worker.py @@ -0,0 +1,117 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Execute only synchronized code, under inherited and verified supervisor locks.""" +from __future__ import annotations + +import argparse +import os +from pathlib import Path +import signal +import subprocess +import sys +import traceback + +TOOLS = Path(__file__).resolve().parents[1] / "tools" +sys.path.insert(0, str(TOOLS)) +from daily_process import inherited_locks +from daily_recovery import prepare_daily_retry +from suite_notify import notification_environment, notification_text, notify_launch, snapshot +from suite_runtime import logged, now, save + + +def checked_revision(repo, branch, head): + """Do not fetch, switch branches, or trust a stale supervisor's source snapshot.""" + def git(*args): + return subprocess.check_output(["git", "-C", str(repo), *args], text=True).strip() + if git("branch", "--show-current") != branch or git("rev-parse", "HEAD") != head: + raise RuntimeError("Synchronized revision changed before Daily execution") + if git("status", "--porcelain", "--untracked-files=normal"): + raise RuntimeError("Daily execution requires a clean synchronized checkout") + + +def execute_locked(repo, home, directory, head, publish_latest=False): + state = snapshot(directory / "status.json") + state["worker_pid"] = os.getpid() + old_handlers = {} + for sig in (signal.SIGTERM, *([signal.SIGBREAK] if os.name == "nt" else [])): + old_handlers[sig] = signal.signal(sig, lambda *_: (_ for _ in ()).throw(KeyboardInterrupt())) + + def record_batch(batch): + state["batch"] = str(batch) + save(directory / "status.json", state) + + try: + if state.get("head") != head or state.get("status") != "PREPARING": + raise RuntimeError("Daily execution does not match the synchronized launch receipt") + checked_revision(repo, state["branch"], head) + state["worker_head"] = head + save(directory / "status.json", state) + with prepare_daily_retry(repo, home, directory, prepared=True): + command = [sys.executable, str(repo / "test-e2e/infra/tools/validate_test_assets.py"), + "--phase", "implementation", "--daily-report", str(directory / "asset-validation.json")] + if logged(command, repo, os.environ.copy(), directory / "validate.log", 180): + raise RuntimeError("Formal asset validation failed; inspect local launcher log") + checked_revision(repo, state["branch"], head) + import static_assets + import suite + static_assets.apply(home, repo / "test-e2e/infra/assets") + records = suite.selection(repo, argparse.Namespace(case=None, stage=None, feature=None, change=None), daily=True) + problems = suite.doctor(repo, home, records, common_only=True) + if problems: + save(directory / "doctor.json", {"ready": False, "problems": problems}) + raise RuntimeError("Runtime preflight failed; inspect local doctor.json") + settings = suite.config(repo, home) + state["status"] = "EXECUTING" + save(directory / "status.json", state) + code = suite.execute(repo, home, records, settings, daily=True, publish_latest=publish_latest, + on_batch_created=record_batch) + state["status"] = "PASSED" if code == 0 else "FAILED" + return code + except (Exception, KeyboardInterrupt) as exc: + env = notification_environment(home) + reason = "Interrupted" if isinstance(exc, KeyboardInterrupt) else f"{type(exc).__name__}: {exc}" + state.update(status="INTERRUPTED" if isinstance(exc, KeyboardInterrupt) else "INCOMPLETE", + error=notification_text(reason, env)) + trace = directory / "worker-error.log" + trace.write_text(notification_text(traceback.format_exc(), env), encoding="utf-8") + return 2 + finally: + for sig, handler in old_handlers.items(): + signal.signal(sig, handler) + state["finished_at"] = now() + save(directory / "status.json", state) + + +def execute(repo, home, directory, head, publish_latest=False): + with inherited_locks(repo, home): + return execute_locked(repo, home, directory, head, publish_latest) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--test-home", type=Path, required=True) + parser.add_argument("--launch", type=Path, required=True) + parser.add_argument("--head") + parser.add_argument("--publish-latest", action="store_true") + parser.add_argument("--notify-only", action="store_true") + args = parser.parse_args(argv) + repo = Path(__file__).resolve().parents[3] + home, directory = args.test_home.resolve(), args.launch.resolve() + if home.is_relative_to(repo) or repo.is_relative_to(home) or directory.parent != home / "runs/daily-launch": + parser.error("Daily execution paths must belong to an external test-home launch") + runtime = home / "runtime/test-venv" / ("Scripts/python.exe" if os.name == "nt" else "bin/python") + if os.path.normcase(os.path.abspath(sys.executable)) != os.path.normcase(os.path.abspath(runtime)): + parser.error("Invoke the worker with the dedicated test-home Python") + if args.notify_only: + notify_launch(directory, home) + return 0 + if not args.head: + parser.error("Daily execution needs a synchronized commit") + try: + return execute(repo, home, directory, args.head, args.publish_latest) + finally: + notify_launch(directory, home) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test-e2e/infra/tools/daily_process.py b/test-e2e/infra/tools/daily_process.py new file mode 100644 index 0000000000..91d8965578 --- /dev/null +++ b/test-e2e/infra/tools/daily_process.py @@ -0,0 +1,146 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Start a fresh executor while the supervisor retains the actual Daily leases.""" +from contextlib import contextmanager +import json +import os +import signal +import stat +import subprocess +import sys + +from suite_runtime import BatchLock, _DEPLOYMENT_LEASE + +LEASE_ENV = "_NEXENT_DAILY_LOCK_HANDLES" + + +@contextmanager +def worker_options(locks): + """Export exactly these file handles; retain the configured runtime environment.""" + descriptors = [lock.handle.fileno() for lock in locks] + if os.name == "nt": + import msvcrt + handles = [msvcrt.get_osfhandle(fd) for fd in descriptors] + prior = [os.get_handle_inheritable(handle) for handle in handles] + try: + for handle in handles: + os.set_handle_inheritable(handle, True) + startup = subprocess.STARTUPINFO() + startup.lpAttributeList = {"handle_list": handles} + yield {"parent": os.getpid(), "kind": "windows", "handles": handles}, { + "startupinfo": startup, "close_fds": True, + "creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} + finally: + for handle, inherited in zip(handles, prior): + os.set_handle_inheritable(handle, inherited) + else: + yield {"parent": os.getpid(), "kind": "posix", "handles": descriptors}, {"pass_fds": descriptors} + + +@contextmanager +def inherited_locks(repo, home): + """Accept OS handles, never a boolean that bypasses the deployment lock.""" + try: + lease = json.loads(os.environ.pop(LEASE_ENV)) + if not isinstance(lease, dict): + raise ValueError("invalid lease") + expected_kind = "windows" if os.name == "nt" else "posix" + parent = lease.get("parent") + if type(parent) is not int or parent <= 0 or lease.get("kind") != expected_kind: + raise ValueError("invalid supervisor") + if os.name == "nt": + # Windows venv launchers may insert an intermediate parent process. + from daily_recovery import owner_alive + if not owner_alive(parent): + raise ValueError("supervisor exited") + elif parent != os.getppid(): + raise ValueError("invalid supervisor") + handles = lease["handles"] + if len(handles) != 2 or any(type(fd) is not int or fd < 0 for fd in handles) or handles[0] == handles[1]: + raise ValueError("invalid handles") + except (KeyError, TypeError, ValueError): + raise RuntimeError("Daily execution requires inherited supervisor locks") from None + paths = [home / "state/daily.lock", repo / ".git/nexent-e2e-deployment.lock"] + streams = [] + token = None + try: + for handle, path in zip(handles, paths): + if path.is_symlink() or path.absolute() != path.resolve(): + raise RuntimeError("Daily lease path must not contain symbolic links") + if os.name == "nt": + import msvcrt + handle = msvcrt.open_osfhandle(handle, os.O_RDWR) + stream = os.fdopen(handle, "r+b", buffering=0) + streams.append(stream) + os.set_inheritable(handle, False) + actual = os.fstat(handle) + if not stat.S_ISREG(actual.st_mode) or not os.path.samestat(actual, path.stat()): + raise RuntimeError("Inherited Daily lock does not match the expected path") + if os.name != "nt": + import fcntl + # The inherited open-file description already owns this flock. + fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) + try: + with BatchLock(path): + raise ValueError("Inherited Daily lease is not locked") + except RuntimeError as exc: + if str(exc) != "another daily run is active": + raise + token = _DEPLOYMENT_LEASE.set((str(paths[1].resolve()), os.getpid())) + yield + finally: + if token is not None: + _DEPLOYMENT_LEASE.reset(token) + for stream in reversed(streams): + stream.close() + + +def fresh_command(repo, home, directory): + # A unique, unwritten cache prefix also prevents timestamp-valid old .pyc reads. + cache = home / "runtime/daily-python-cache" / directory.name + return [sys.executable, "-B", "-X", f"pycache_prefix={cache}", + str(repo / "test-e2e/infra/scripts/daily-worker.py"), + "--test-home", str(home), "--launch", str(directory)] + + +def run_worker(repo, home, directory, head, publish_latest, locks): + """Keep both supervisor locks until the executor exits, including during recovery.""" + command = [*fresh_command(repo, home, directory), "--head", head] + if publish_latest: + command.append("--publish-latest") + with worker_options(locks) as (lease, options): + env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1") + env[LEASE_ENV] = json.dumps(lease) + process = subprocess.Popen(command, cwd=repo, env=env, **options) + try: + return process.wait() + except KeyboardInterrupt: + # Send a catchable interruption only to the coordinator, which owns cleanup. + previous = {sig: signal.signal(sig, signal.SIG_IGN) for sig in (signal.SIGTERM, signal.SIGINT)} + try: + if process.poll() is None: + try: + process.send_signal(signal.CTRL_BREAK_EVENT if os.name == "nt" else signal.SIGTERM) + except OSError: + print("Could not signal Daily; retaining locks until the executor exits", flush=True) + while True: + try: + process.wait() + break + except KeyboardInterrupt: + print("Waiting for Daily execution and environment recovery to finish", flush=True) + finally: + for sig, handler in previous.items(): + signal.signal(sig, handler) + raise + + +def notify_terminal(repo, home, directory): + """Notification also loads the synchronized code in a separate interpreter.""" + command = [*fresh_command(repo, home, directory), "--notify-only"] + try: + result = subprocess.run(command, cwd=repo, timeout=150, check=False) + if result.returncode: + print("Terminal notification helper failed; inspect the launch receipt", flush=True) + except (OSError, subprocess.TimeoutExpired): + print("Terminal notification helper unavailable; original result retained", flush=True) diff --git a/test-e2e/infra/tools/daily_recovery.py b/test-e2e/infra/tools/daily_recovery.py index 255a83bbfd..0c3d880961 100644 --- a/test-e2e/infra/tools/daily_recovery.py +++ b/test-e2e/infra/tools/daily_recovery.py @@ -12,6 +12,7 @@ import yaml from local_preflight import env_file +from recovery_io import protected_write from suite_runtime import BatchLock, deployment_lock_owned, now, save MODE_MARKER = "state/deployment-exceptions/pending.json" @@ -112,15 +113,8 @@ def validate(self): previous = document(batch / "status.json") if previous.get("status") not in {"INCOMPLETE", "FAILED", "PASSED", "COMPLETE"}: raise RuntimeError("Deployment journal has no terminal owner evidence") - journal = next((p for p in (batch / "checkpoints/results.jsonl", batch / "results.journal.jsonl") - if p.exists() or p.is_symlink()), None) - if journal is not None: - for line in plain(journal).read_text(encoding="utf-8").splitlines(): - row = json.loads(line) - if row.get("stage") == "D5" and row.get("result") in {"TIMEOUT", "CLEANUP_FAILED"}: - raise RuntimeError("Previous D5 timeout/cleanup failure requires environment safety review") - elif previous.get("stages", {}).get("D5", "PENDING") != "PENDING": - raise RuntimeError("Previous D5 execution has no recovery evidence") + # Historical Case receipts are reporting evidence, not deployment recovery authority. + # The active fault lease/marker is checked by prepare_daily_retry before any writes. if not isinstance(files, list) or len(files) != 4 or document(backup / "index.json") != {"files": files}: raise ValueError("incomplete backup index") originals = {} @@ -164,29 +158,57 @@ def validate(self): raise RuntimeError("Deployment recovery backup/configuration failed safety validation") from None def prepare(self): - from deployment_exceptions import protected_write plain(self.archive) self.archive.parent.mkdir(parents=True, mode=0o700, exist_ok=True) self.archive.mkdir(mode=0o700) protected_write(self.archive / "pending.json", self.marker_bytes) # Preserve every current file before the first configuration write. + before_images = {} for index, path in enumerate(self.originals): if path.exists(): - protected_write(self.archive / str(index), path.read_bytes()) - self.record("PREPARED_FOR_REDEPLOY") + content = path.read_bytes() + protected_write(self.archive / str(index), content) + before_images[str(index)] = hashlib.sha256(content).hexdigest() + else: + before_images[str(index)] = None + self.record("PREPARED_FOR_REDEPLOY", before_images=before_images) for path, content in self.originals.items(): if content is None: path.unlink(missing_ok=True) else: protected_write(path, content) + def resume_prepared(self): + """Revalidate the supervisor's before-images without writing a second backup.""" + receipt = document(self.receipt) + if (receipt.get("status") != "PREPARED_FOR_REDEPLOY" + or receipt.get("evidence_backup") != str(self.archive) + or receipt.get("previous_batch") != self.payload["batch"] + or plain(self.archive / "pending.json").read_bytes() != self.marker_bytes): + raise RuntimeError("Prepared Daily recovery receipt differs from its journal") + before_images = receipt.get("before_images") + if not isinstance(before_images, dict) or set(before_images) != {str(i) for i in range(len(self.originals))}: + raise RuntimeError("Prepared Daily recovery before-image index is incomplete") + for index, digest in before_images.items(): + path = plain(self.archive / index) + if digest is None: + if path.exists(): + raise RuntimeError("Unexpected prepared recovery before-image") + elif (not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest) + or hashlib.sha256(path.read_bytes()).hexdigest() != digest): + raise RuntimeError("Prepared recovery before-image checksum differs") + for path, content in self.originals.items(): + actual = plain(path).read_bytes() if path.exists() else None + if actual != content: + raise RuntimeError("Prepared Local configuration changed before execution") + def record(self, status, **details): save(self.receipt, {"status": status, "at": now(), "previous_batch": self.payload["batch"], "evidence_backup": str(self.archive), **details}) @contextmanager -def prepare_daily_retry(repo, home, directory): +def prepare_daily_retry(repo, home, directory, *, prepared=False): """Require the checkout lease; retain the marker until a verified new campaign owns it.""" if not deployment_lock_owned(repo): raise RuntimeError("Daily recovery requires the checkout deployment lock") @@ -195,8 +217,11 @@ def prepare_daily_retry(repo, home, directory): if (home / FAULT_MARKER).exists() or (home / FAULT_MARKER).is_symlink(): raise RuntimeError("Pending D5 fault recovery needs container/data safety review before synchronization") retry = DailyRetry(repo, home, directory) if (home / MODE_MARKER).exists() or (home / MODE_MARKER).is_symlink() else None + if retry is None and prepared and (directory / "recovery.json").exists(): + if document(directory / "recovery.json").get("status") == "PREPARED_FOR_REDEPLOY": + raise RuntimeError("Prepared Daily recovery journal disappeared before execution") if retry is not None: - retry.prepare() + retry.resume_prepared() if prepared else retry.prepare() token = _RETRY.set(retry) try: yield retry diff --git a/test-e2e/infra/tools/deployment_exceptions.py b/test-e2e/infra/tools/deployment_exceptions.py index 684cb3412b..88fc01705a 100644 --- a/test-e2e/infra/tools/deployment_exceptions.py +++ b/test-e2e/infra/tools/deployment_exceptions.py @@ -20,6 +20,7 @@ from jsonschema import Draft202012Validator from local_preflight import env_file +from recovery_io import protected_write from daily_recovery import configuration_paths, current_retry, pending_retry_allowed from suite_runtime import deployment_lock, finalize, fingerprint, now, render_report, save @@ -54,11 +55,11 @@ def catalog(repo, records=None): value = checked_yaml(path, path.with_suffix(".schema.json")) if set(value["cases"].values()) - set(value["profiles"]): raise ValueError("Deployment exception references an unknown profile") - if set(value["recovery_cases"]) & set(value["cases"]): - raise ValueError("Local recovery cannot use deployment exception Cases") + # Kept only for readers of older catalog versions; never schedules or gates recovery. + value.setdefault("recovery_cases", []) if records is not None: available = {record["case_id"] for record in records} - missing = (set(value["cases"]) | set(value["recovery_cases"])) - available + missing = set(value["cases"]) - available if missing: raise ValueError("Deployment exceptions reference unknown Case IDs: " + ", ".join(sorted(missing))) return value @@ -119,16 +120,6 @@ def patch_env(content, updates): return text.encode("utf-8") -def protected_write(path, content): - path.parent.mkdir(parents=True, exist_ok=True) - temporary = path.with_name(path.name + ".switch-next") - with os.fdopen(os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "wb") as output: - output.write(content) - output.flush() - os.fsync(output.fileno()) - temporary.replace(path) - - def link_or_copy(source, destination): """Keep original evidence paths valid without duplicating binaries on one disk.""" try: @@ -242,7 +233,7 @@ def restore(self): path.unlink(missing_ok=True) if self.switched: self.hook("deploy", self.settings, self.repo, self.home, self.directory / "restore-local", self.env) - self.hook("preflight", self.settings, self.repo, self.home, self.directory / "restore-local", self.env) + self.hook("preflight", self.settings, self.repo, self.home, self.directory / "restore-local", self.env) self.verify(self.baseline_values, "restore-local") # Deployment can re-render ignored files; restore their exact saved bytes afterwards. for path, content in self.originals.items(): @@ -285,12 +276,12 @@ def execute_campaign(repo, home, records, settings, *, daily, publish_latest, ex old_handler = signal.signal(signal.SIGTERM, lambda *_: (_ for _ in ()).throw(KeyboardInterrupt())) print(f"Deployment campaign: {directory}", flush=True) - def child(label, group, *, recovery=False): + def child(label, group): child_dir = directory / "profiles" / label seen = set() def record_result(row): - if not recovery and row["case_id"] not in seen: + if row["case_id"] not in seen: case_id = row["case_id"] for section in ("cases", "logs"): source = child_dir / section / (case_id if section == "cases" else case_id + ".log") @@ -321,18 +312,11 @@ def record_result(row): # Finalization can add NOT_EXECUTED rows after an interruption. for row in results: record_result(row) - if not summary["execution_complete"] or (recovery and any(row["result"] != "PASS" for row in results)): + if not summary["execution_complete"]: raise RuntimeError("Incomplete or unsafe deployment group: " + label) - if any(row["result"] in {"TIMEOUT", "CLEANUP_FAILED"} and row["stage"] == "D5" for row in results): + if (home / "state/deployment-guard/pending.json").exists(): raise RuntimeError("Deployment group needs environment safety review") - def local_recovery_cases(): - recovery = [r for r in all_records if r["case_id"] in rules["recovery_cases"]] - if switch.retry is not None and (not recovery or len(recovery) != len(set(rules["recovery_cases"])) - or any(r["status"] != "active" or not r.get("execution") for r in recovery)): - raise RuntimeError("Daily recovery requires executable Local recovery Cases") - return recovery - try: # Also excludes other opted-in test homes operating on this checkout. with deployment_lock(repo): @@ -355,7 +339,6 @@ def local_recovery_cases(): hook("preflight", settings, repo, home, directory, env) switch.verify({}, "local") if switch.retry is not None: - child("recover-previous-local", local_recovery_cases(), recovery=True) switch.retry.record("REDEPLOY_VERIFIED", batch=str(directory)) state["stages"]["D0"] = "COMPLETE" for label, group in selected.items(): @@ -373,7 +356,6 @@ def local_recovery_cases(): state["profile"] = "restore-local" save(directory / "status.json", state) switch.restore() - child("restore-local", local_recovery_cases(), recovery=True) switch.complete() restoration_ok = True current = fingerprint(repo) @@ -383,12 +365,12 @@ def local_recovery_cases(): error = "Interrupted" if isinstance(exc, KeyboardInterrupt) else str(exc) if isinstance(exc, RuntimeError) else type(exc).__name__ finally: signal.signal(signal.SIGTERM, old_handler) - summary = finalize(directory, report_records, rows, error) + summary = finalize(directory, report_records, rows, error, environment=env) state.update(status=summary["status"], finished_at=now(), error=error, local_restored_and_verified=restoration_ok) state["stages"] = {f"D{i}": "COMPLETE" if summary["execution_complete"] else "INTERRUPTED" for i in range(7)} state["stages"]["D6"] = "COMPLETE" save(directory / "status.json", state) - render_report(directory, report_records, rows, summary) + render_report(directory, report_records, rows, summary, environment=env) save(home / "reports/repository-latest.json", {"batch": str(directory), "status": summary["status"]}) if daily and summary["execution_complete"] and restoration_ok: if publish_latest: @@ -402,7 +384,7 @@ def local_recovery_cases(): except OSError: print("Publication failed; could not save local publication receipt", flush=True) try: - render_report(directory, report_records, rows, summary, sharing=publication) + render_report(directory, report_records, rows, summary, sharing=publication, environment=env) except OSError: print("Could not refresh publication status in the finalized report", flush=True) if daily and settings.get("notify"): diff --git a/test-e2e/infra/tools/docs/daily-validation.md b/test-e2e/infra/tools/docs/daily-validation.md index 5a21fd1160..392b944ebf 100644 --- a/test-e2e/infra/tools/docs/daily-validation.md +++ b/test-e2e/infra/tools/docs/daily-validation.md @@ -6,13 +6,15 @@ instead of `--generate`; it never rewrites repository assets or silently shrinks ## Decisions -- Unknown findings, malformed contracts/configuration, duplicate IDs and unsafe +- Unknown findings, malformed shared contracts/configuration, duplicate IDs and unsafe script paths stop the batch. Unsafe recovery journals, active owners/locks and pending D5 fault recovery also stop before synchronization. An ordinary failed deployment can use the validated Daily redeployment path described below. - Explicit `CASE_EXECUTION_MISSING`, `CASE_SCRIPT_MISSING`, - `CASE_SELECTOR_MISSING` and `CASE_PREPARATION_MISSING` findings with a known owner retain that Case in the plan + `CASE_SELECTOR_MISSING`, `CASE_PREPARATION_MISSING` and `CASE_EXECUTION_INVALID` findings with a known owner retain that Case in the plan and produce a BLOCKED receipt without starting its worker. + Invalid execution YAML, timeout, preparation or Mock declarations are scoped to + that Case. Unsafe script paths and ambiguous script ownership remain fatal. - Missing traditional UT references and stale derived reading views are metadata warnings, not proof that formal Cases cannot run. Authoritative assets still pass all other validation. Daily never claims UT execution from reference validation. @@ -36,10 +38,22 @@ instead of `--generate`; it never rewrites repository assets or silently shrinks dependency failures retain the existing BLOCKED_BY_DEPENDENCY behavior; no dependency edges are inferred from titles, Change membership or failure keywords. -The launcher, full-registry selector, worker selection and deployment campaign use -the same typed policy. The global environment/configuration doctor, clean-source -and fingerprint gates remain required. Daily recovery is verified before business -Cases can start. +The launcher, selector, worker and deployment campaign use the same typed policy. +Execution checks only common configuration and its dedicated Python interpreter +before starting the batch. `doctor` (including existing `doctor --live` deployment +hooks) checks common environment readiness. Explicit `doctor --full` includes +complete dependency diagnostics; +an unavailable browser, jsdom dependency or unused business profile does not stop +independent Cases. Each known Node framework is checked once per execution group, +and affected Cases receive BLOCKED receipts with the actual prerequisite reason. +Opaque custom scripts and Python imports report dependency errors in their own +workers. No dependency failure is counted as PASS. + +The parent validates the full registry; each worker inspects only its requested Case. +Source/configuration fingerprints remain checked before every Case and finalization. +Dependency/cache directories are pruned before walking the tree, not filtered after +traversing their contents. This reduces repeated scanning without weakening byte +change detection. Clean-source and environment-recovery checks remain required. ## Retrying a failed deployment @@ -51,14 +65,22 @@ Local baseline, previous owner exit and absence of unrelated configuration edits Legacy journals require a terminal prior-batch receipt. It saves the old journal and current configuration privately under `runtime/daily-recovery/`, then restores only the four owned configuration files. Prior reports, original backups and -business data/volumes remain intact. +business data/volumes remain intact. Historical business-result files are not +deployment-recovery authority and may be missing or damaged without blocking a +validated redeploy. Their original evidence remains untouched. Actual fault journals, +active locks, validated configuration backups and the prior owner's terminal status +remain mandatory. The legacy `recovery_cases` catalog field is optional and no longer +participates in scheduling or Case-reference validation. The old journal remains pending through synchronization, asset validation and doctor. A new deployment campaign then adopts responsibility with its own backup and pending journal. It builds/deploys the latest synchronized revision, requires -preflight and runtime-mode verification, and executes the configured Local recovery -selection before starting ordinary Cases. Final Local restoration and recovery -must pass before clearing that campaign's unchanged journal. Any failure retains +preflight and runtime-mode verification before starting ordinary Cases. Final Local +restoration, service readiness and deployment-mode checks must pass before clearing +that campaign's unchanged journal. Business Cases are never recovery gates: they run +once in their normal group, and FAIL/TIMEOUT/CLEANUP_FAILED remains a failed result +without suppressing independent Cases. Only an unresolved environment fault marker +blocks further execution; a D5 outcome alone is not evidence of such a fault. Any failure retains recoverable evidence; a later Daily can try again after its owner exits. Targeted runs and resume do not gain this permission. A D5 fault journal or active @@ -74,8 +96,50 @@ python -m pytest -p no:nexent_formal_pytest test-e2e/infra/tools/tests/test_dail Default implementation validation checks explicit preparation on every active D2-D5 binding; no per-Case opt-in flag is needed. A self-contained proof declares empty families after dependency/cleanup analysis. Daily suppresses the worker only -for a Case missing that declaration; invalid declarations/configuration remain -fatal. Developer strict validation rejects both types of finding before delivery. +for the affected Case when that declaration is missing or invalid. Invalid shared +configuration remains fatal. Developer strict validation rejects both types of +Case finding before delivery. + +## Fresh execution after synchronization + +The supervisor retains the test-home Daily lock and checkout deployment lock +through synchronization and until a fresh `daily-worker.py` interpreter exits. +An explicitly selected validation branch is allowed through `--branch`; the +scheduled workflow remains pinned to develop. Branch/ref validation, clean +checkout, fast-forward-only fetch and exact remote HEAD checks also apply to +validation branches; selecting one does not disable any execution gate. +Only the supervisor fetches/fast-forwards. The worker checks the expected HEAD, +branch, clean checkout and inherited OS lock handles before any execution. There +is no unlocked handoff and no public skip-lock flag. Nested deployment operations +reuse the worker's verified lease. Linux shares the held flock descriptors; +Windows explicitly inherits only the selected handles while the supervisor +retains its byte locks. The supervisor forwards a catchable interruption and waits +for the coordinator's recovery/finalization before closing its own locks. + +Both execution and notification helpers use the fixed virtual-environment Python, +with a launch-specific bytecode-cache prefix and bytecode writes disabled. This +avoids reusing either in-memory modules or timestamp-valid `.pyc` files from the +previous checkout. The worker never fetches again. Recovery's `protected_write` +is in independent `recovery_io.py`, so preparing a repair cannot import a cached +deployment executor before synchronization. + +If the supervisor prepared an old Local recovery journal, the worker revalidates +that journal, receipt, private before-image path and exact restored config bytes, +then reconstructs its own recovery context without overwriting the original +backup. Old markers remain until the existing redeploy, service-readiness and +deployment-mode checks succeed. An exit without a matching terminal launch receipt becomes INCOMPLETE, +including a zero exit; it never implies successful test execution. A failed +worker also leaves a redacted traceback in `worker-error.log` when it can save one. +The fresh notification helper and the worker use the same launch/batch receipts, +so a supervisor fallback cannot blindly resend a delivered/uncertain message. + +Offline process and recovery regressions live in `tests/test_daily_process.py`. +They include pending recovery with a deliberately cached old executor, real child +process/lock inheritance, same-size/timestamp stale bytecode, invalid leases, +source drift, abnormal exit and interruption. They do not build, deploy, send +messages, clear real markers or execute business Cases. Hard host/process death +still cannot guarantee in-process notification; Windows supervision assumes the +supervisor remains alive until its child has finished. ## Accounting @@ -89,9 +153,9 @@ rewritten. D6 and the sanitized publication bundle retain the validation receipt ## Terminal reports and notifications Daily sends a terminal report when a run succeeds, fails or is interrupted. A -failed recovery probe must not suppress notification. The report states the +failed environment recovery check must not suppress notification. The report states the execution status, saved reason, result counts and recovery state, distinguishing -restored configuration from a passed Local recovery probe. Missing recovery +restored configuration from passed deployment-mode and service checks. Missing recovery evidence is reported as unverified, never as a healthy environment. Partial runs retain `NOT_EXECUTED` and the original nonzero exit status. @@ -103,9 +167,15 @@ the launcher uses that batch's notification receipt rather than an unrelated detailed report could not be rendered, the notification uses saved terminal metadata and labels the missing details explicitly. -`notification-report.md` is the notification payload; original report and result -evidence remain local and unchanged by sending. Known environment credentials are -redacted from the payload. `notification.json` records `SENDING`, `SENT`, +The formal `report.md` includes terminal metadata and known-credential redaction +when generated, and is written atomically. Notifications send that same file; +an already-current sanitized report is not rewritten and no duplicate payload is +created. Older reports may have their terminal section refreshed and credentials +redacted in place before sending. Only a missing, empty or unreadable report, or a +failed preparation of that file, produces `notification-report.md`: a clearly +labelled exception notification built from saved terminal records. Raw result +receipts, logs, screenshots and Trace evidence remain unchanged. +`notification.json` records `SENDING`, `SENT`, `UNKNOWN` or `NOT_CONFIGURED`. Concurrent sends use a per-record lock; existing or corrupt receipts are never blindly retried. Transport timeout or uncertainty does not change the test outcome. Explicit `notify: false` remains respected. @@ -136,3 +206,6 @@ python -m pytest -p no:nexent_formal_pytest test-e2e/infra/tools/tests/test_dail The formal pytest plugin is disabled only for framework UT so they do not require machine-local business credentials. It remains enabled for actual formal Cases. + +The legacy catalog `recovery_cases` list is retained for format compatibility; it no +longer schedules extra startup/finalization business tests or makes them a gate. diff --git a/test-e2e/infra/tools/local_preflight.py b/test-e2e/infra/tools/local_preflight.py index e56f9b74d2..6f6bc3c998 100644 --- a/test-e2e/infra/tools/local_preflight.py +++ b/test-e2e/infra/tools/local_preflight.py @@ -60,7 +60,7 @@ def valid_url(value: object) -> bool: return parsed.scheme in {"http", "https"} and bool(parsed.hostname) and not parsed.username and not parsed.password -def static_config_problems(home: Path, records: list[dict], *, repo: Path) -> list[str]: +def static_config_problems(home: Path, records: list[dict], *, repo: Path, common_only=False) -> list[str]: config = home / "config" daily, problems = env_file(config / "daily.env") _, errors = env_file(config / "secrets.env") @@ -79,6 +79,8 @@ def static_config_problems(home: Path, records: list[dict], *, repo: Path) -> li if (config / "secrets.env").stat().st_mode & 0o077: problems.append("secrets.env must not be readable by group or others") + if common_only: + return sorted(set(problems)) online = [record for record in records if record["stage"] != "D1" and record["status"] == "active"] for record in online: execution = record.get("execution") diff --git a/test-e2e/infra/tools/node_preflight.py b/test-e2e/infra/tools/node_preflight.py index 43851d0ade..a5178424b8 100644 --- a/test-e2e/infra/tools/node_preflight.py +++ b/test-e2e/infra/tools/node_preflight.py @@ -29,11 +29,12 @@ """ -def node_toolchain_problems(repo: Path, home: Path, frameworks: set[str]) -> list[str]: +def node_toolchain_problems(repo: Path, home: Path, frameworks: set[str], *, environment=None) -> list[str]: """Probe in child processes, without installing or changing shared dependencies.""" node = shutil.which("node") if not node: return [] # The caller reports the missing Node executable. + env = {**os.environ, **(environment or {}), "PLAYWRIGHT_BROWSERS_PATH": str(home / "runtime/playwright-browsers")} probes = [] if "vitest" in frameworks: probes.append(("D1 jsdom/CSS dependency resolution (including nested permissions)", @@ -43,8 +44,8 @@ def node_toolchain_problems(repo: Path, home: Path, frameworks: set[str]) -> lis probes.append(("product frontend Chromium/headless-shell", "frontend", BROWSER_PROBE, "")) if "playwright" in frameworks: # D4's fixed config uses the full Chromium channel for attachment downloads. - probes.append(("D4 full Chromium", "test-e2e/infra/automation/d4", BROWSER_PROBE, "chromium")) - env = dict(os.environ, PLAYWRIGHT_BROWSERS_PATH=str(home / "runtime/playwright-browsers")) + probes.append(("D4 Chromium", "test-e2e/infra/automation/d4", BROWSER_PROBE, + env.get("NEXENT_TEST_BROWSER_CHANNEL") or "chromium")) problems = [] for label, relative, script, channel in probes: try: diff --git a/test-e2e/infra/tools/recovery_io.py b/test-e2e/infra/tools/recovery_io.py new file mode 100644 index 0000000000..655f0efea2 --- /dev/null +++ b/test-e2e/infra/tools/recovery_io.py @@ -0,0 +1,14 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Private configuration writes, independent of any deployment executor.""" +import os + + +def protected_write(path, content): + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(path.name + ".switch-next") + with os.fdopen(os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "wb") as output: + output.write(content) + output.flush() + os.fsync(output.fileno()) + temporary.replace(path) diff --git a/test-e2e/infra/tools/suite.py b/test-e2e/infra/tools/suite.py index fad95fed44..f5514543d3 100644 --- a/test-e2e/infra/tools/suite.py +++ b/test-e2e/infra/tools/suite.py @@ -76,7 +76,7 @@ def config(repo, home): def selection(repo, args, *, daily=False, worker=False): - issues, registry = inspect(repo) + issues, registry = inspect(repo, case_ids=set(args.case or []) if worker else None) if issues and not (daily or worker): raise ValueError(f"Invalid test assets: {len(issues)} binding issues") records = registry["cases"] @@ -108,13 +108,17 @@ def selection(repo, args, *, daily=False, worker=False): return sorted(result, key=lambda record: (record["stage"], record["case_id"])) -def doctor(repo, home, records, *, live=False): - problems = static_config_problems(home, records, repo=repo) +def doctor(repo, home, records, *, live=False, common_only=False): + problems = static_config_problems(home, records, repo=repo, common_only=common_only) expected_runtime = (home / "runtime/test-venv").resolve() if Path(sys.prefix).resolve() != expected_runtime or not test_python(home).is_file(): problems.append("Run this suite with the dedicated /runtime/test-venv Python") if sys.version_info[:2] != (3, 11): problems.append("Product runtime requires Python 3.11") + if common_only: + if live and not problems: + problems.extend(live_service_problems(home)) + return sorted(set(problems)) for project in ("backend", "sdk"): data = tomllib.loads((repo / project / "pyproject.toml").read_text(encoding="utf-8")) for declaration in data["project"]["dependencies"]: @@ -129,26 +133,52 @@ def doctor(repo, home, records, *, live=False): except importlib.metadata.PackageNotFoundError: problems.append(f"missing {project} dependency: {req.name}") frameworks = {record["execution"]["implementations"][0]["framework"] for record in records if record["execution"]} + problems.extend(node_runtime_problems(repo, home, frameworks)) + if live and not problems: + problems.extend(live_service_problems(home)) + return sorted(set(problems)) + + +def node_runtime_problems(repo, home, frameworks, *, environment=None): + """Check one selected Node toolchain once, without installing dependencies.""" + problems = [] if frameworks & {"vitest", "playwright", "custom"}: node = shutil.which("node") if not node: problems.append("Node.js >=22 is required") else: - version = subprocess.check_output([node, "--version"], text=True).strip() - if int(version.lstrip("v").split(".")[0]) < 22: - problems.append("Node.js >=22 is required") + try: + version = subprocess.check_output([node, "--version"], text=True, timeout=10).strip() + if int(version.lstrip("v").split(".")[0]) < 22: + problems.append("Node.js >=22 is required") + except (OSError, ValueError, subprocess.SubprocessError): + problems.append("Cannot verify Node.js >=22") + if problems: + return problems for framework, relative in ( ("vitest", "d1/frontend/node_modules/vitest/vitest.mjs"), ("playwright", "d4/node_modules/playwright/cli.js"), ): if framework in frameworks and not (repo / "test-e2e/infra/automation" / relative).is_file(): problems.append(f"missing locked {framework} toolchain") - problems.extend(node_toolchain_problems(repo, home, frameworks)) - if live and not problems: - problems.extend(live_service_problems(home)) + problems.extend(node_toolchain_problems(repo, home, frameworks, environment=environment)) return sorted(set(problems)) +def runtime_blocks(repo, home, records, *, environment=None): + """Isolate known toolchain failures; opaque custom scripts check their own dependencies.""" + blocked = {} + for framework in ("vitest", "playwright"): + selected = [record for record in records if record.get("status") == "active" + and any(item.get("framework") == framework + for item in (record.get("execution") or {}).get("implementations", []))] + if selected: + problems = node_runtime_problems(repo, home, {framework}, environment=environment) + for record in selected if problems else []: + blocked.setdefault(record["case_id"], []).extend(problems) + return blocked + + def execute_hooks(name, settings, repo, home, directory, env): @@ -427,6 +457,8 @@ def _execute(repo, home, records, settings, *, daily=False, resume=None, publish for name in ("build", "deploy", "preflight"): execute_hooks(name, settings, repo, home, directory, env) state["stages"]["D0"] = "COMPLETE" + blocked_runtime = runtime_blocks(repo, home, records, environment=env) + save(directory / "d0/runtime-preflight.json", {"blocked_cases": blocked_runtime}) for stage in STAGES: state["stages"][stage] = "RUNNING" save(directory / "status.json", state) @@ -451,6 +483,11 @@ def _execute(repo, home, records, settings, *, daily=False, resume=None, publish save(directory / "cases" / case_id / "receipt.json", { "case_id": case_id, "stage": record["stage"], "result": row["result"], "cleanup_exit_code": 0, "reused_from": row["reused_from"]}) + elif case_id in blocked_runtime: + row = {"case_id": case_id, "stage": stage, "result": "BLOCKED", + "reason": "; ".join(blocked_runtime[case_id]), + "acceptance_status": "INCOMPLETE", "evidence": [f"cases/{case_id}/receipt.json"]} + save(directory / "cases" / case_id / "receipt.json", row) elif record["status"] != "active" or not record["execution"]: result = {"retired": "RETIRED", "skipped_by_policy": "SKIPPED_BY_POLICY"}.get(record["status"], "BLOCKED") row = {"case_id": case_id, "stage": stage, "result": result, "reason": record["status"], "evidence": []} @@ -486,9 +523,7 @@ def _execute(repo, home, records, settings, *, daily=False, resume=None, publish (" (reused)" if reused else ""), flush=True) if on_result is not None: on_result(dict(row)) - if (home / "state/deployment-guard/pending.json").exists() or ( - stage == "D5" and row["result"] in {"TIMEOUT", "CLEANUP_FAILED"} - ): + if (home / "state/deployment-guard/pending.json").exists(): raise RuntimeError("Environment recovery needs verification before continuing") state["stages"][stage] = "COMPLETE" current = fingerprint(repo) @@ -501,7 +536,7 @@ def _execute(repo, home, records, settings, *, daily=False, resume=None, publish except Exception as exc: error = str(exc) if isinstance(exc, RuntimeError) else type(exc).__name__ finally: - summary = finalize(directory, report_records, rows, error) + summary = finalize(directory, report_records, rows, error, environment=env) for stage, status in state["stages"].items(): if status in {"PENDING", "RUNNING"}: state["stages"][stage] = "NOT_EXECUTED" if status == "PENDING" else "INTERRUPTED" @@ -521,7 +556,7 @@ def _execute(repo, home, records, settings, *, daily=False, resume=None, publish except OSError: print("Publication failed; could not save local publication receipt", flush=True) try: - render_report(directory, report_records, rows, summary, sharing=sharing) + render_report(directory, report_records, rows, summary, sharing=sharing, environment=env) except OSError: # Keep the previously finalized report and continue the existing notification path. print("Could not refresh publication status in the finalized report", flush=True) @@ -541,6 +576,7 @@ def main(argv=None): parser.add_argument("--change") parser.add_argument("--execute", action="store_true", help="Actually run tests or install dependencies") parser.add_argument("--live", action="store_true", help="Doctor: probe configured local HTTP services") + parser.add_argument("--full", action="store_true", help="Doctor: include optional Case dependencies in diagnostics") parser.add_argument("--batch-dir", type=Path, help="Existing local batch directory for resume") args = parser.parse_args(argv) repo, home = repository_root(), args.test_home.resolve() @@ -566,12 +602,12 @@ def main(argv=None): "reused": len(resume["rows"]), "remaining": len(records) - len(resume["rows"])}, ensure_ascii=False, indent=2)) return 0 - problems = doctor(repo, home, records) + problems = doctor(repo, home, records, common_only=True) if problems: print(json.dumps({"ready": False, "problems": problems}, ensure_ascii=False, indent=2)) return 2 return execute(repo, home, records, settings, resume=resume) - records = selection(repo, args, daily=args.mode == "daily", worker=args.mode == "worker") + records = selection(repo, args, daily=args.mode in {"daily", "doctor"}, worker=args.mode == "worker") if args.mode == "worker": if not args.batch_dir or len(records) != 1: parser.error("Worker requires one Case and a batch directory") @@ -581,7 +617,7 @@ def main(argv=None): settings = json.loads((directory / "pipeline.json").read_text(encoding="utf-8")) return worker(repo, home, records[0], directory / "cases" / records[0]["case_id"], settings) if args.mode == "doctor": - problems = doctor(repo, home, records, live=args.live) + problems = doctor(repo, home, records, live=args.live, common_only=not args.full) print(json.dumps({"ready": not problems, "problems": problems}, ensure_ascii=False, indent=2)) return 2 if problems else 0 if args.mode == "plan" or not args.execute: @@ -598,7 +634,7 @@ def main(argv=None): return 0 if args.mode == "daily" and (args.case or args.stage or args.feature or args.change): parser.error("Daily must use the full registry; select cases with run") - problems = doctor(repo, home, records) + problems = doctor(repo, home, records, common_only=True) if problems: print(json.dumps({"ready": False, "problems": problems}, ensure_ascii=False, indent=2)) return 2 diff --git a/test-e2e/infra/tools/suite_notify.py b/test-e2e/infra/tools/suite_notify.py index 0c60b3ce44..984ac9cb0e 100644 --- a/test-e2e/infra/tools/suite_notify.py +++ b/test-e2e/infra/tools/suite_notify.py @@ -2,17 +2,13 @@ """Notify terminal Daily outcomes independently of publication and recovery success.""" import json -import os from pathlib import Path -import re import shutil from suite_runtime import BatchLock, logged, now, save -from suite_terminal import terminal_context +from suite_terminal import current_report, notification_environment, notification_text, terminal_context, write_report TERMINAL_STATES = frozenset({"PASSED", "FAILED", "INCOMPLETE", "INTERRUPTED", "CANCELLED"}) -ENV_LINE = re.compile(r"^(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)=(.*)$") -SECRET_KEY = re.compile(r"password|passwd|secret|token|api[_-]?key|authorization|cookie|credential|app[_-]?key|jwt", re.I) def snapshot(path): @@ -23,35 +19,6 @@ def snapshot(path): return {} -def notification_environment(home): - """Read notification credentials even when unrelated product preflight is broken.""" - env = dict(os.environ) - for name in ("daily.env", "secrets.env"): - path = Path(home) / "config" / name - try: - lines = path.read_text(encoding="utf-8").splitlines() - except (OSError, UnicodeError): - continue - for line in lines: - match = ENV_LINE.fullmatch(line.strip()) - if match: - key, value = match.groups() - value = value.strip() - if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'): - value = value[1:-1] - if value: - env[key] = value - return env - - -def notification_text(text, env): - """Keep known credentials out of new failure notifications; preserve raw local evidence.""" - for value in sorted({str(value) for key, value in env.items() - if SECRET_KEY.search(key) and len(str(value)) >= 4}, key=len, reverse=True): - text = text.replace(value, "[REDACTED]") - return re.sub(r"(?i)(Bearer\s+)[A-Za-z0-9._~-]+", r"\1[REDACTED]", text) - - def send_report(directory: Path, env): """Attempt one notification for every terminal state, including incomplete runs.""" directory = Path(directory) @@ -69,19 +36,24 @@ def send_report(directory: Path, env): if not chat or not cli: save(marker, {"status": "NOT_CONFIGURED", "at": now()}) return "NOT_CONFIGURED" - context = "\n".join(terminal_context(directory, summary)) report = directory / "report.md" try: body = report.read_text(encoding="utf-8") - except OSError: - body = "详细报告未生成；本通知依据已保存的终态记录，不能视为完整验收报告。\n" - # The current state is authoritative even if report generation failed earlier. - message = body if context in body else "# Nexent Daily 终态报告\n\n" + context + "\n" + body - (directory / "notification-report.md").write_text(notification_text(message, env), encoding="utf-8") + if not body.strip(): + raise ValueError("Empty report") + ready = notification_text(current_report(body, directory, summary, environment=env), env) + if ready != body: + write_report(report, ready) + except (OSError, UnicodeError, ValueError): + # A failed render/read cannot suppress the saved terminal outcome. + report = directory / "notification-report.md" + message = "# Nexent Daily 异常通知\n\n" + "\n".join(terminal_context(directory, summary)) + message += "\n详细报告未生成或不可读取；本通知依据已保存的终态记录，不能视为完整验收报告。\n" + write_report(report, notification_text(message, env)) save(marker, {"status": "SENDING", "at": now()}) try: code = logged([cli, "im", "+messages-send", "--as", "bot", "--chat-id", chat, - "--file", "./notification-report.md"], directory, env, + "--file", "./" + report.name], directory, env, directory / "logs/notification.log", 120) status = "SENT" if code == 0 else "UNKNOWN" except Exception: diff --git a/test-e2e/infra/tools/suite_publish.py b/test-e2e/infra/tools/suite_publish.py index a260e4c630..9842484819 100644 --- a/test-e2e/infra/tools/suite_publish.py +++ b/test-e2e/infra/tools/suite_publish.py @@ -241,7 +241,7 @@ def publish_latest(batch, home, env=None, *, ownership_file=None): sharing = {"status": "PUBLISHED", "batch_id": batch.name, "excluded_binary_count": len(excluded), "published_at": now()} render_report(staging, plan, rows, summary, metadata_directory=batch, - evidence_root=batch.resolve().as_posix(), sharing=sharing) + evidence_root=batch.resolve().as_posix(), sharing=sharing, environment=env) contents = {"report.md": redact_text((staging / "report.md").read_text(encoding="utf-8"), secrets).encode()} for name, value in (("summary.json", summary), ("status.json", state), ("provenance.json", provenance), ("report-ownership.json", ownership), ("excluded-evidence.json", excluded)): diff --git a/test-e2e/infra/tools/suite_runtime.py b/test-e2e/infra/tools/suite_runtime.py index 35dedc3286..cadb1583d4 100644 --- a/test-e2e/infra/tools/suite_runtime.py +++ b/test-e2e/infra/tools/suite_runtime.py @@ -15,7 +15,7 @@ import subprocess from report_ownership import git_label, responsibility_label -from suite_terminal import beijing_time, terminal_context +from suite_terminal import beijing_time, notification_text, terminal_context, write_report def now(): @@ -73,10 +73,10 @@ def deployment_lock(repo): yield return path = repo / ".git/nexent-e2e-deployment.lock" - with BatchLock(path): + with BatchLock(path) as lock: token = _DEPLOYMENT_LEASE.set((str(path.resolve()), os.getpid())) try: - yield + yield lock finally: _DEPLOYMENT_LEASE.reset(token) @@ -126,7 +126,11 @@ def fingerprint(repo): """Pin actual worktree bytes as well as HEAD, excluding generated runtime files.""" ignored = {"node_modules", "__pycache__", ".pytest_cache", "test-results", "runs", ".playwright-internal"} digest = hashlib.sha256() - for path in sorted((repo / "test-e2e").rglob("*")): + paths = [] + for directory, children, files in os.walk(repo / "test-e2e"): + children[:] = sorted(name for name in children if name not in ignored) + paths.extend(Path(directory) / name for name in files) + for path in sorted(paths): relative = path.relative_to(repo) if not path.is_file() or set(relative.parts) & ignored: continue @@ -172,7 +176,7 @@ def config_fingerprint(home): return digest.hexdigest() -def finalize(directory, plan, rows, error=None): +def finalize(directory, plan, rows, error=None, *, environment=None): """Account for every planned Case; interrupted work can never become PASS.""" counts = Counter(row["case_id"] for row in rows) duplicates = sorted(key for key, count in counts.items() if count > 1) @@ -209,7 +213,7 @@ def finalize(directory, plan, rows, error=None): (directory / "checkpoints").mkdir(parents=True, exist_ok=True) (directory / "checkpoints/results.jsonl").write_text( "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows), encoding="utf-8") - render_report(directory, plan, rows, summary) + render_report(directory, plan, rows, summary, environment=environment) return summary @@ -274,14 +278,15 @@ def snapshot(name): if provenance.get("deployment_campaign"): lines += ["## 部署切换与 Local 恢复", "", "- 部署例外和执行分组：`deployment-plan.json`；常规用例不重复计数。", - "- Local 配置恢复及恢复用例验证：" + + "- Local 配置恢复及部署模式、服务检查：" + ("通过" if state.get("local_restored_and_verified") else "未完成或未通过，不能认为环境已恢复可用"), - "- 恢复记录：`restoration.json`；恢复回归原始证据仅保留在执行机器：`profiles/restore-local/`。", + "- 恢复记录：`restoration.json`；部署模式检查：`restore-local/mode-verification.json`。", "- 原始配置备份仅保留在执行机器的私有 runtime 目录，不进入共享日志包。", ""] return lines -def render_report(directory, plan, rows, summary, *, metadata_directory=None, evidence_root=None, sharing=None): +def render_report(directory, plan, rows, summary, *, metadata_directory=None, evidence_root=None, sharing=None, + environment=None): """Refresh the reading view only, preserving final receipts and batch timestamps.""" metrics = report_metrics(plan, rows) def percent(value): @@ -350,4 +355,5 @@ def cell(value): lines += ["", "No non-passing cases."] if summary.get("error"): lines += ["", "Batch ended before completion. Inspect status.json and local logs."] - (directory / "report.md").write_text("\n".join(lines) + "\n", encoding="utf-8") + write_report(directory / "report.md", notification_text("\n".join(lines) + "\n", + os.environ if environment is None else environment)) diff --git a/test-e2e/infra/tools/suite_terminal.py b/test-e2e/infra/tools/suite_terminal.py index 00c104f7fb..794baf6aff 100644 --- a/test-e2e/infra/tools/suite_terminal.py +++ b/test-e2e/infra/tools/suite_terminal.py @@ -3,7 +3,71 @@ """Format terminal metadata independently of launcher modules cached before Git sync.""" from datetime import datetime, timedelta, timezone import json +import os from pathlib import Path +import re +import tempfile + +ENV_LINE = re.compile(r"^(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)=(.*)$") +SECRET_KEY = re.compile(r"password|passwd|secret|token|api[_-]?key|authorization|cookie|credential|app[_-]?key|jwt", re.I) + + +def notification_environment(home): + """Read report credentials independently of product preflight availability.""" + env = dict(os.environ) + for name in ("daily.env", "secrets.env"): + path = Path(home) / "config" / name + try: + lines = path.read_text(encoding="utf-8").splitlines() + except (OSError, UnicodeError): + continue + for line in lines: + match = ENV_LINE.fullmatch(line.strip()) + if match: + key, value = match.groups() + value = value.strip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'): + value = value[1:-1] + if value: + env[key] = value + return env + + +def notification_text(text, env): + """Sanitize reading views while leaving raw receipts and Case evidence intact.""" + for value in sorted({str(value) for key, value in env.items() + if SECRET_KEY.search(key) and len(str(value)) >= 4}, key=len, reverse=True): + text = text.replace(value, "[REDACTED]") + return re.sub(r"(?i)(Bearer\s+)[A-Za-z0-9._~-]+", r"\1[REDACTED]", text) + + +def write_report(path, text): + """Replace a report atomically so readers cannot observe a partial render.""" + path = Path(path) + descriptor, temporary = tempfile.mkstemp(prefix=".report-", suffix=".tmp", dir=path.parent) + try: + with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as output: + output.write(text) + os.replace(temporary, path) + finally: + Path(temporary).unlink(missing_ok=True) + + +def current_report(body, directory, summary, *, environment=None): + """Refresh legacy or stale terminal metadata in the same report, without duplicating it.""" + context = notification_text("\n".join(terminal_context(directory, summary)), + os.environ if environment is None else environment) + if context in body: + return body + section = re.compile(r"^## 终态与环境恢复\r?\n(?:\r?\n)*.*?(?=\r?\n\r?\n|\Z)", re.M | re.S) + if section.search(body): + body = section.sub(lambda _: context.rstrip("\n"), body, count=1) + else: + body = "# Nexent Daily 终态报告\n\n" + context + "\n" + body + status = summary.get("status") + if status: + body = re.sub(r"^Status: [^\r\n]*", lambda _: "Status: " + status, body, count=1, flags=re.M) + return body def beijing_time(value): @@ -30,6 +94,7 @@ def snapshot(name): state, provenance = snapshot("status.json"), snapshot("provenance.json") restoration = snapshot("restoration.json") + retry = snapshot("recovery.json") status = summary.get("status") or state.get("status") or "UNKNOWN" complete = summary.get("execution_complete") execution = "完整" if complete is True else "未完成" if complete is False else "未建立执行批次或未记录" @@ -42,9 +107,11 @@ def snapshot(name): else: reason = "未记录具体原因，请核查启动、执行及恢复日志" if state.get("local_restored_and_verified") is True: - recovery = "Local 配置恢复及恢复用例验证通过" + recovery = "Local 配置恢复及部署模式、服务检查通过" elif restoration.get("status") == "RESTORED": - recovery = "Local 配置已还原；恢复用例未通过或尚未完成，环境可用性未验证" + recovery = "Local 配置已还原；环境恢复检查未通过或尚未完成，环境可用性未验证" + elif retry.get("status") == "PREPARED_FOR_REDEPLOY": + recovery = "Local 配置已还原；待重新部署及恢复验证，前批恢复标记仍待处理" elif provenance.get("deployment_campaign") or (metadata / "deployment-plan.json").exists(): recovery = "Local 恢复未完成或未验证，请检查恢复记录；不能认为环境已恢复可用" else: diff --git a/test-e2e/infra/tools/test_asset_lib.py b/test-e2e/infra/tools/test_asset_lib.py index c1ce94a2fa..417a5d4799 100644 --- a/test-e2e/infra/tools/test_asset_lib.py +++ b/test-e2e/infra/tools/test_asset_lib.py @@ -74,14 +74,15 @@ def load_structured_file(path: Path) -> dict[str, Any]: return value -def discover_documents(root: Path, kind: str) -> list[AssetDocument]: +def discover_documents(root: Path, kind: str, *, case_ids=None) -> list[AssetDocument]: directory = root / ASSET_DIRECTORIES[kind] if not directory.exists(): return [] if kind == "features": paths = sorted(directory.glob("*/feature.yaml")) elif kind == "cases": - paths = sorted(directory.glob("*/case.yaml")) + paths = sorted(path for path in directory.glob("*/case.yaml") + if case_ids is None or path.parent.name in case_ids) elif kind == "changes": paths = sorted(directory.glob("*/*.yaml")) else: diff --git a/test-e2e/infra/tools/tests/test_daily_launch.py b/test-e2e/infra/tools/tests/test_daily_launch.py index 8543c25d52..219bf38824 100644 --- a/test-e2e/infra/tools/tests/test_daily_launch.py +++ b/test-e2e/infra/tools/tests/test_daily_launch.py @@ -58,6 +58,21 @@ def test_dirty_checkout_never_fetches(self): launcher.sync_checkout(Path("repo"), "develop") self.assertEqual(git.call_count, 1) + def test_explicit_validation_branch_is_plannable_without_outputs(self): + with tempfile.TemporaryDirectory() as root: + output = io.StringIO() + with redirect_stdout(output): + self.assertEqual(launcher.main(["--test-home", root, "--branch", "djb/verify-daily-process"]), 0) + self.assertEqual(json.loads(output.getvalue())["branch"], "djb/verify-daily-process") + self.assertFalse((Path(root) / "runs").exists()) + + def test_invalid_ref_is_rejected_before_fetch(self): + with patch.object(launcher, "git", side_effect=["", "invalid..branch", + subprocess.CalledProcessError(128, "check-ref-format")]) as git: + with self.assertRaises(subprocess.CalledProcessError): + launcher.sync_checkout(Path("repo"), "invalid..branch") + self.assertFalse(any("fetch" in call.args for call in git.call_args_list)) + def test_wrong_branch_never_fetches(self): with patch.object(launcher, "git", side_effect=["", "other"]) as git: with self.assertRaisesRegex(RuntimeError, "branch differs"): diff --git a/test-e2e/infra/tools/tests/test_daily_process.py b/test-e2e/infra/tools/tests/test_daily_process.py new file mode 100644 index 0000000000..a5be8542c3 --- /dev/null +++ b/test-e2e/infra/tools/tests/test_daily_process.py @@ -0,0 +1,263 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Exercise actual fresh processes and temporary leases; never deploy or send messages.""" +from contextlib import nullcontext +import importlib.util +import json +import os +from pathlib import Path +import py_compile +import signal +import subprocess +import sys +from unittest.mock import patch + +import pytest + +TOOLS = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(TOOLS)) +import daily_process as processes +import daily_recovery as recovery +from suite_runtime import BatchLock, deployment_lock, save + + +def script(name): + spec = importlib.util.spec_from_file_location(name.replace("-", "_"), TOOLS.parent / "scripts" / (name + ".py")) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +launcher, worker = script("daily-launch"), script("daily-worker") + + +def test_recovery_write_does_not_import_deployment_executor(tmp_path): + import builtins + original_import = builtins.__import__ + + def guarded(name, *args, **kwargs): + assert name != "deployment_exceptions", "Recovery must not preload the executor before Git sync" + return original_import(name, *args, **kwargs) + + target = tmp_path / "private/config" + with patch("builtins.__import__", side_effect=guarded): + recovery.protected_write(target, b"preserved local data") + assert target.read_bytes() == b"preserved local data" + + +@pytest.mark.parametrize("reverse_handles", [False, True]) +def test_real_child_loads_new_code_and_holds_both_locks(tmp_path, reverse_handles): + repo, home = tmp_path / "repo", tmp_path / "home" + (repo / ".git").mkdir(parents=True) + root = repo / "test-e2e/infra/scripts" + root.mkdir(parents=True) + module = root / "versioned_executor.py" + module.write_text("def execute_campaign(*, on_batch_created=None): return 'old-version'\n") + spec = importlib.util.spec_from_file_location("cached_old_executor", module) + old = importlib.util.module_from_spec(spec) + spec.loader.exec_module(old) + py_compile.compile(str(module), doraise=True) + previous_time = module.stat().st_mtime + module.write_text("def execute_campaign(*, on_batch_created=None): return 'new-version'\n") + os.utime(module, (previous_time, previous_time)) + child = root / "daily-worker.py" + child.write_text(f'''import json,sys +from pathlib import Path +sys.path.insert(0, {str(TOOLS)!r}) +from daily_process import inherited_locks +from suite_runtime import BatchLock, deployment_lock, deployment_lock_owned +import versioned_executor +repo,home=Path({str(repo)!r}),Path({str(home)!r}) +with inherited_locks(repo,home): + assert deployment_lock_owned(repo) + with deployment_lock(repo): + assert deployment_lock_owned(repo) + for p in (home/'state/daily.lock',repo/'.git/nexent-e2e-deployment.lock'): + try: + with BatchLock(p): raise AssertionError('Execution lost mutual exclusion') + except RuntimeError: pass + (home/'child-result.json').write_text(json.dumps({{'version':versioned_executor.execute_campaign(on_batch_created=None),'pid':__import__('os').getpid()}})) +''') + with BatchLock(home / "state/daily.lock") as daily, deployment_lock(repo) as checkout: + locks = [checkout, daily] if reverse_handles else [daily, checkout] + assert processes.run_worker(repo, home, home / "unused-launch", "head", False, locks) == (1 if reverse_handles else 0) + for path in (daily.path, checkout.path): + with pytest.raises(RuntimeError, match="another daily run"): + with BatchLock(path): + pass + if reverse_handles: + assert not (home / "child-result.json").exists() + else: + result = json.loads((home / "child-result.json").read_text()) + assert result["version"] == "new-version" and result["pid"] != os.getpid() + assert old.execute_campaign() == "old-version" + for path in (daily.path, checkout.path): + with BatchLock(path): + pass + + +@pytest.mark.parametrize("lease", [None, "broken", "[]", '{"parent":0,"handles":[],"kind":"posix"}']) +def test_missing_or_forged_lease_is_rejected(tmp_path, monkeypatch, lease): + if lease is None: + monkeypatch.delenv(processes.LEASE_ENV, raising=False) + else: + monkeypatch.setenv(processes.LEASE_ENV, lease) + with pytest.raises(RuntimeError, match="inherited supervisor locks"): + with processes.inherited_locks(tmp_path / "repo", tmp_path / "home"): + pytest.fail("Missing locks must never permit execution") + + +@pytest.mark.parametrize("outcome", ["missing", "crash", "launch-error", "interrupt", "success"]) +def test_supervisor_accounts_for_worker_exit_and_always_calls_notifier(tmp_path, monkeypatch, outcome): + home = tmp_path / "home" + runtime = home / "runtime/test-venv" / ("Scripts/python.exe" if os.name == "nt" else "bin/python") + monkeypatch.setattr(launcher.sys, "executable", str(runtime)) + monkeypatch.setattr(launcher, "deployment_lock", lambda *_: nullcontext()) + monkeypatch.setattr(launcher, "prepare_daily_retry", lambda *_: nullcontext()) + monkeypatch.setattr(launcher, "sync_checkout", lambda *a, **k: "head") + + def execute(repo, home, directory, *args): + if outcome == "launch-error": + raise OSError("offline subprocess launch failure") + if outcome == "interrupt": + raise KeyboardInterrupt() + if outcome == "success": + state = json.loads((directory / "status.json").read_text()) + save(directory / "status.json", {**state, "status": "PASSED"}) + return -9 if outcome == "crash" else 0 + + monkeypatch.setattr(launcher, "run_worker", execute) + with patch.object(launcher, "notify_terminal") as notify: + code = launcher.main(["--test-home", str(home), "--branch", "develop", "--execute"]) + notify.assert_called_once() + directory = next((home / "runs/daily-launch").iterdir()) + state = json.loads((directory / "status.json").read_text()) + assert code == (0 if outcome == "success" else 2) + assert state["status"] == ("PASSED" if outcome == "success" else "INTERRUPTED" if outcome == "interrupt" else "INCOMPLETE") + if outcome in {"crash", "missing"}: + assert "without a matching terminal receipt" in state["error"] + + +def test_worker_rejects_revision_change_before_any_execution(tmp_path): + with patch.object(worker.subprocess, "check_output", side_effect=["develop", "other-head"]): + with pytest.raises(RuntimeError, match="revision changed"): + worker.checked_revision(tmp_path, "develop", "expected-head") + + +@pytest.fixture +def abandoned_process(tmp_path): + import deployment_exceptions as modes + repo, home = tmp_path / "repo", tmp_path / "home" + (repo / ".git").mkdir(parents=True) + env = repo / "deploy/env/.env" + env.parent.mkdir(parents=True) + env.write_bytes(b"ENABLE_AIDP_KNOWLEDGE=false\nDATABASE=preserved\n") + save(home / "runs/repository-daily/old/status.json", {"status": "INCOMPLETE"}) + (home / "config").mkdir() + (home / "config/environment.yaml").write_text("features: {aidp_knowledge: false}\n") + settings = {"hooks": {"deploy": [["offline-deploy"]], "preflight": [["offline-check"]]}} + switch = modes.ModeSwitch(repo, home, home / "runs/repository-daily/old", settings, {}, None) + switch.snapshot() + marker = home / modes.MARKER + value = json.loads(marker.read_text()) + value.pop("owner_pid") + save(marker, value) + return repo, home, marker, switch + + +@pytest.mark.parametrize("failure", ["checksum", "config", "receipt", "before-image", "missing-marker"]) +def test_child_revalidates_prepared_recovery_without_replacing_original_backup(abandoned_process, failure): + import deployment_exceptions as modes + repo, home, marker, switch = abandoned_process + original = marker.read_bytes() + launch = home / "runs/daily-launch/upgrade" + with deployment_lock(repo), recovery.prepare_daily_retry(repo, home, launch) as prepared: + baseline = (switch.backup / "0").read_bytes() + with recovery.prepare_daily_retry(repo, home, launch, prepared=True) as resumed: + assert resumed.marker_bytes == original + modes.assert_recovered(home, allow_retry=True) + if failure == "checksum": + (prepared.archive / "pending.json").write_text("changed") + elif failure == "before-image": + (prepared.archive / "0").write_text("corrupted before-image") + elif failure == "missing-marker": + marker.unlink() + elif failure == "config": + env = repo / "deploy/env/.env" + env.write_bytes(env.read_bytes().replace(b"KNOWLEDGE=false", b"KNOWLEDGE=true")) + else: + save(launch / "recovery.json", {"status": "RECOVERED"}) + with pytest.raises(RuntimeError): + with recovery.prepare_daily_retry(repo, home, launch, prepared=True): + pytest.fail("Changed recovery evidence cannot be adopted") + if failure != "missing-marker": + assert marker.read_bytes() == original + assert (switch.backup / "0").read_bytes() == baseline + + +def test_pending_recovery_and_cached_old_executor_survive_upgrade_in_fresh_child(abandoned_process): + import types + repo, home, marker, switch = abandoned_process + root = repo / "test-e2e/infra/scripts" + root.mkdir(parents=True) + child = root / "daily-worker.py" + child.write_text(f'''import inspect,sys +from pathlib import Path +sys.path.insert(0,{str(TOOLS)!r}) +from daily_process import inherited_locks +from daily_recovery import prepare_daily_retry +repo,home,launch=Path({str(repo)!r}),Path({str(home)!r}),Path({str(home / 'runs/daily-launch/upgrade')!r}) +with inherited_locks(repo,home),prepare_daily_retry(repo,home,launch,prepared=True): + import deployment_exceptions + inspect.signature(deployment_exceptions.execute_campaign).bind_partial(on_batch_created=None) + deployment_exceptions.assert_recovered(home,allow_retry=True) + (home/'upgraded').write_text('fresh executor with preserved recovery') +''') + old = types.ModuleType("deployment_exceptions") + old.execute_campaign = lambda: None + before = marker.read_bytes() + with patch.dict(sys.modules, {"deployment_exceptions": old}): + with BatchLock(home / "state/daily.lock") as daily, deployment_lock(repo) as checkout: + with recovery.prepare_daily_retry(repo, home, home / "runs/daily-launch/upgrade"): + with pytest.raises(TypeError): + old.execute_campaign(on_batch_created=None) + assert processes.run_worker(repo, home, home / "runs/daily-launch/upgrade", "head", False, + [daily, checkout]) == 0 + assert sys.modules["deployment_exceptions"] is old + assert (home / "upgraded").read_text() == "fresh executor with preserved recovery" + assert marker.read_bytes() == before and (switch.backup / "0").is_file() + + +def test_worker_invalid_lease_cannot_rewrite_a_running_launch(tmp_path, monkeypatch): + launch = tmp_path / "runs/daily-launch/current" + save(launch / "status.json", {"status": "EXECUTING", "head": "head"}) + original = (launch / "status.json").read_bytes() + monkeypatch.delenv(processes.LEASE_ENV, raising=False) + with pytest.raises(RuntimeError, match="inherited supervisor locks"): + worker.execute(tmp_path / "repo", tmp_path, launch, "head") + assert (launch / "status.json").read_bytes() == original + + +def test_interrupt_forwards_to_worker_and_waits_for_recovery(): + from unittest.mock import Mock + process = Mock() + process.wait.side_effect = [KeyboardInterrupt(), KeyboardInterrupt(), None] + process.poll.return_value = None + locks = [Mock(), Mock()] + with patch.object(processes, "worker_options", return_value=nullcontext(({}, {}))), \ + patch.object(processes.subprocess, "Popen", return_value=process), \ + pytest.raises(KeyboardInterrupt): + processes.run_worker(Path("repo"), Path("home"), Path("launch"), "head", False, locks) + process.send_signal.assert_called_once_with(signal.CTRL_BREAK_EVENT if os.name == "nt" else signal.SIGTERM) + assert process.wait.call_count == 3 + + +def test_notifier_uses_fresh_interpreter_and_failure_never_relabels_result(tmp_path): + save(tmp_path / "status.json", {"status": "INCOMPLETE", "error": "worker exited"}) + original = (tmp_path / "status.json").read_bytes() + with patch.object(processes.subprocess, "run", side_effect=subprocess.TimeoutExpired("offline-helper", 150)) as send: + processes.notify_terminal(Path("repo"), Path("home"), tmp_path) + argv = send.call_args.args[0] + assert "-B" in argv and any(str(arg).startswith("pycache_prefix=") for arg in argv) + assert "--notify-only" in argv + assert (tmp_path / "status.json").read_bytes() == original diff --git a/test-e2e/infra/tools/tests/test_daily_recovery.py b/test-e2e/infra/tools/tests/test_daily_recovery.py index 009227e867..4ae1d9a3f9 100644 --- a/test-e2e/infra/tools/tests/test_daily_recovery.py +++ b/test-e2e/infra/tools/tests/test_daily_recovery.py @@ -54,18 +54,17 @@ def synchronize(*args, **kwargs): calls.append("fetch-latest") return "new-revision-with-fix" - def execute(*args, **kwargs): + def execute(repo, home, directory, *args): calls.append("redeploy") assert marker.exists() + state = json.loads((directory / "status.json").read_text()) + save(directory / "status.json", {**state, "status": "PASSED"}) return 0 with patch.object(launcher, "__file__", str(repo / "test-e2e/infra/scripts/daily-launch.py")), \ patch.object(launcher.sys, "executable", str(runtime)), \ patch.object(launcher, "sync_checkout", side_effect=synchronize), \ - patch.object(launcher, "git", return_value=""), patch.object(launcher, "logged", return_value=0), \ - patch("static_assets.apply"), patch.object(suite, "selection", return_value=[]), \ - patch.object(suite, "doctor", return_value=[]), patch.object(suite, "config", return_value={}), \ - patch.object(suite, "execute", side_effect=execute): + patch.object(launcher, "run_worker", side_effect=execute), patch.object(launcher, "notify_terminal"): assert launcher.main(["--test-home", str(home), "--branch", "develop", "--execute"]) == 0 assert calls == ["fetch-latest", "redeploy"] assert marker.read_bytes() == before @@ -165,14 +164,22 @@ def test_live_owner_and_d5_fault_journal_remain_hard_stops(abandoned): @pytest.mark.parametrize("outcome", ["TIMEOUT", "CLEANUP_FAILED"]) -def test_prior_unsafe_d5_results_block_even_without_a_fault_marker(abandoned, outcome): +def test_prior_case_outcome_does_not_replace_environment_recovery_checks(abandoned, outcome): repo, home, batch, marker, switch = abandoned path = batch / "results.journal.jsonl" path.write_text(json.dumps({"stage": "D5", "result": outcome}) + "\n") - with deployment_lock(repo), pytest.raises(RuntimeError, match="D5 timeout/cleanup"): - with recovery.prepare_daily_retry(repo, home, home / "runs/daily-launch/unsafe-d5"): - pass - assert marker.exists() + before = path.read_bytes() + with deployment_lock(repo), recovery.prepare_daily_retry(repo, home, home / "runs/daily-launch/retry-d5"): + assert marker.exists() + assert marker.exists() and path.read_bytes() == before + + +def test_prior_d5_result_file_is_not_required_for_environment_redeployment(abandoned): + repo, home, batch, marker, _ = abandoned + save(batch / "status.json", {"status": "INCOMPLETE", "stages": {"D5": "RUNNING"}}) + with deployment_lock(repo), recovery.prepare_daily_retry(repo, home, home / "runs/daily-launch/no-results"): + assert marker.exists() + assert marker.exists() and not (batch / "results.journal.jsonl").exists() def test_dead_parent_without_terminal_batch_receipt_does_not_prove_children_exited(abandoned): @@ -228,7 +235,7 @@ def test_changed_marker_cannot_be_adopted_or_cleared(abandoned): def test_partial_configuration_restore_retains_all_before_images(abandoned): repo, home, batch, marker, switch = abandoned original = marker.read_bytes() - write = modes.protected_write + write = recovery.protected_write def failing_write(path, content): if path == switch.paths[-1]: @@ -236,7 +243,7 @@ def failing_write(path, content): return write(path, content) launch = home / "runs/daily-launch/write-failure" - with deployment_lock(repo), patch.object(modes, "protected_write", side_effect=failing_write), pytest.raises(OSError): + with deployment_lock(repo), patch.object(recovery, "protected_write", side_effect=failing_write), pytest.raises(OSError): with recovery.prepare_daily_retry(repo, home, launch): pass archive = home / "runtime/daily-recovery/write-failure" @@ -299,7 +306,7 @@ def verify(controller, updates, label): def execute(repo, home, selected, settings, *, directory, on_result=None): probes.append(directory.name) assert marker.exists() # Neither old recovery nor the active campaign disappears early. - results = [{"case_id": r["case_id"], "stage": r["stage"], "result": "FAIL" if failure == "probe" else "PASS", "evidence": []} for r in selected] + results = [{"case_id": r["case_id"], "stage": r["stage"], "result": "FAIL" if failure == "case" and r["case_id"] == "LOCAL-RECOVERY" else "PASS", "evidence": []} for r in selected] directory.mkdir(parents=True) for row in results: if on_result: @@ -314,24 +321,24 @@ def execute(repo, home, selected, settings, *, directory, on_result=None): patch.object(modes.ModeSwitch, "verify", new=verify), \ patch("report_ownership.enrich", side_effect=lambda repo, rows, head: (rows, [])), \ patch("suite_notify.send_report") as notify: - code = modes.execute_campaign(repo, home, records[:2], settings, daily=True, publish_latest=False, + code = modes.execute_campaign(repo, home, records[:2] if invalid_recovery else records, settings, daily=True, publish_latest=False, execute_one=execute, all_records=records, hook=hook, machine_env=lambda *_: {"NEXENT_TEST_ASSETS_CONTAINER_HOST": "127.0.0.1"}) notify.assert_not_called() return code, calls, probes -@pytest.mark.parametrize("failure", [None, "build", "deploy", "preflight", "verify", "probe"]) -def test_redeploy_and_local_probe_must_succeed_before_business_cases(abandoned, failure): +@pytest.mark.parametrize("failure", [None, "build", "deploy", "preflight", "verify", "case"]) +def test_only_environment_failures_block_redeployment(abandoned, failure): repo, home, batch, marker, switch = abandoned old = marker.read_bytes() code, calls, probes = run_campaign(abandoned, "campaign", failure) assert code == (0 if failure is None else 1) assert (home / "runtime/daily-recovery/campaign/pending.json").read_bytes() == old - assert marker.exists() is (failure is not None) - if failure is None: + assert marker.exists() is (failure not in {None, "case"}) + if failure in {None, "case"}: assert calls[:3] == ["build", "deploy", "preflight"] - assert probes == ["recover-previous-local", "local", "modelengine", "restore-local"] + assert probes == ["local", "modelengine"] assert json.loads((home / "runs/daily-launch/campaign/recovery.json").read_text())["status"] == "RECOVERED" else: assert "local" not in probes and "modelengine" not in probes @@ -350,12 +357,12 @@ def test_another_failed_redeploy_does_not_prevent_a_later_success(abandoned): @pytest.mark.parametrize("invalid", ["empty", "blocked", "unimplemented"]) -def test_empty_or_unexecutable_recovery_is_never_a_success(abandoned, invalid): +def test_legacy_recovery_case_metadata_does_not_block_environment_checks(abandoned, invalid): repo, home, batch, marker, switch = abandoned code, calls, probes = run_campaign(abandoned, "no-recovery", invalid_recovery=invalid) - assert code == 1 and not probes and marker.exists() + assert code == 0 and probes == ["local", "modelengine"] and not marker.exists() receipt = json.loads((home / "runs/daily-launch/no-recovery/recovery.json").read_text()) - assert receipt["status"] == "ADOPTED_BY_REDEPLOY" + assert receipt["status"] == "RECOVERED" def test_exited_owner_is_detected_without_signalling_it(): @@ -372,3 +379,18 @@ def probe(pid, signal): raise exception() with patch.object(recovery.os, "name", "posix"), patch.object(recovery.os, "kill", side_effect=probe): assert recovery.owner_alive(123) is expected + + +def test_failed_local_business_case_is_recorded_once_and_other_groups_continue(abandoned): + repo, home, _, marker, _ = abandoned + code, _, groups = run_campaign(abandoned, "business-failure", "case") + assert code == 1 + assert groups == ["local", "modelengine"] + assert not marker.exists() + batch = next((home / "runs/repository-daily").glob("deployment-*")) + summary = json.loads((batch / "summary.json").read_text()) + rows = [json.loads(line) for line in (batch / "checkpoints/results.jsonl").read_text().splitlines()] + assert summary["execution_complete"] is True and summary["status"] == "FAILED" + assert summary["counts"] == {"PASS": 2, "FAIL": 1} + assert [r["result"] for r in rows if r["case_id"] == "LOCAL-RECOVERY"] == ["FAIL"] + assert json.loads((batch / "status.json").read_text())["local_restored_and_verified"] is True diff --git a/test-e2e/infra/tools/tests/test_daily_validation.py b/test-e2e/infra/tools/tests/test_daily_validation.py index c03299e31f..ef28daf59d 100644 --- a/test-e2e/infra/tools/tests/test_daily_validation.py +++ b/test-e2e/infra/tools/tests/test_daily_validation.py @@ -157,8 +157,11 @@ def test_strict_validation_still_fails_and_daily_checks_derived_views_without_wr def test_launcher_uses_daily_report_and_never_regenerates(): - script = (Path(__file__).resolve().parents[2] / "scripts/daily-launch.py").read_text(encoding="utf-8") - assert '"--daily-report"' in script and '"--generate"' not in script + scripts = Path(__file__).resolve().parents[2] / "scripts" + supervisor = (scripts / "daily-launch.py").read_text(encoding="utf-8") + executor = (scripts / "daily-worker.py").read_text(encoding="utf-8") + assert '"--daily-report"' in executor + assert '"--generate"' not in supervisor + executor def test_campaign_receives_validation_and_a_disabled_case_without_rebuilding_in_children(tmp_path): diff --git a/test-e2e/infra/tools/tests/test_direct_report_delivery.py b/test-e2e/infra/tools/tests/test_direct_report_delivery.py new file mode 100644 index 0000000000..91b39aa58a --- /dev/null +++ b/test-e2e/infra/tools/tests/test_direct_report_delivery.py @@ -0,0 +1,123 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Send the finalized report itself; only unusable reports need an exception payload.""" +import json +import os +from pathlib import Path +import sys +from unittest.mock import patch + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import suite_notify +import suite_runtime +import suite_terminal +import suite +from suite_runtime import finalize, render_report, save + +ENV = {"LARK_CLI": "offline-cli", "FEISHU_REPORT_CHAT_ID": "offline-chat", "TEST_API_KEY": "fixture-private-key"} + + +def terminal(directory): + save(directory / "status.json", {"status": "INCOMPLETE", "error": "fixture-private-key", "local_restored_and_verified": False}) + rows = [{"case_id": "CASE-A", "stage": "D1", "result": "FAIL", "reason": "Bearer fixture-bearer-value"}] + return rows, finalize(directory, rows, rows, "fixture-private-key") + + +def test_report_is_redacted_when_generated_and_sent_without_rewriting(tmp_path): + rows, summary = terminal(tmp_path) + render_report(tmp_path, rows, rows, summary, environment=ENV) + report = tmp_path / "report.md" + before, stamp = report.read_bytes(), report.stat().st_mtime_ns + assert b"fixture-private-key" not in before and b"fixture-bearer-value" not in before + assert "批次终态：INCOMPLETE" in before.decode() and "环境恢复状态：" in before.decode() + with patch.object(suite_notify, "logged", return_value=0) as send: + assert suite_notify.send_report(tmp_path, ENV) == "SENT" + assert send.call_args.args[0][-1] == "./report.md" + assert not (tmp_path / "notification-report.md").exists() + assert report.read_bytes() == before and report.stat().st_mtime_ns == stamp + assert "fixture-private-key" in (tmp_path / "summary.json").read_text() + assert "fixture-bearer-value" in (tmp_path / "checkpoints/results.jsonl").read_text() + + +@pytest.mark.parametrize("content", [None, b"", b"\xff\xfe"]) +def test_only_missing_empty_or_unreadable_report_uses_exception_payload(tmp_path, content): + _, summary = terminal(tmp_path) + report = tmp_path / "report.md" + report.unlink() + if content is not None: + report.write_bytes(content) + with patch.object(suite_notify, "logged", return_value=0) as send: + assert suite_notify.send_report(tmp_path, ENV) == "SENT" + assert send.call_args.args[0][-1] == "./notification-report.md" + text = (tmp_path / "notification-report.md").read_text(encoding="utf-8") + assert "详细报告未生成" in text and "INCOMPLETE" in text and "环境恢复状态：" in text + assert "fixture-private-key" not in text + assert (report.read_bytes() if report.exists() else None) == content + + +def test_legacy_report_is_updated_in_place_with_one_current_terminal_section(tmp_path): + rows, summary = terminal(tmp_path) + summary.update(status="PASSED", error=None, execution_complete=True) + render_report(tmp_path, rows, rows, summary) + summary.update(status="INCOMPLETE", error="fixture-private-key", execution_complete=False) + save(tmp_path / "summary.json", summary) + with patch.object(suite_notify, "logged", return_value=0) as send: + assert suite_notify.send_report(tmp_path, ENV) == "SENT" + text = (tmp_path / "report.md").read_text(encoding="utf-8") + assert text.count("## 终态与环境恢复") == 1 + assert "批次终态：INCOMPLETE" in text and "批次终态：PASSED" not in text + assert "fixture-private-key" not in text + assert "## Case results" in text + assert "| Result | Count |" in text and "Status: INCOMPLETE" in text + assert send.call_args.args[0][-1] == "./report.md" + assert not (tmp_path / "notification-report.md").exists() + + +def test_failed_atomic_render_keeps_the_previous_complete_report(tmp_path): + rows, summary = terminal(tmp_path) + before = (tmp_path / "report.md").read_bytes() + with patch.object(suite_terminal.os, "replace", side_effect=OSError("fixture disk failure")), pytest.raises(OSError): + render_report(tmp_path, rows, rows, summary, environment=ENV) + assert (tmp_path / "report.md").read_bytes() == before + assert not list(tmp_path.glob(".report-*.tmp")) + + +def test_execution_passes_machine_secrets_to_report_generation(tmp_path): + record = {"case_id": "CASE-A", "stage": "D1", "status": "active", "execution": {"test": True}} + + def run(command, cwd, env, log, timeout): + if "--case" in command: + directory = Path(command[command.index("--batch-dir") + 1]) + save(directory / "cases/CASE-A/receipt.json", { + "case_id": "CASE-A", "stage": "D1", "result": "FAIL", "reason": "fixture-private-key"}) + return 0 + + with patch.object(suite, "machine_environment", return_value={**os.environ, **ENV}), \ + patch.object(suite, "fingerprint", return_value={"head": "fixture"}), \ + patch.object(suite, "static_requirements", return_value=set()), \ + patch.object(suite, "static_inventory", return_value=[]), \ + patch.object(suite, "logged", side_effect=run): + assert suite.execute(tmp_path, tmp_path, [record], {"hooks": {}, "case_timeout_seconds": 20}) == 1 + directory = next((tmp_path / "runs/repository-daily").iterdir()) + assert "fixture-private-key" not in (directory / "report.md").read_text(encoding="utf-8") + assert "fixture-private-key" in (directory / "cases/CASE-A/receipt.json").read_text(encoding="utf-8") + + +def test_unwritable_legacy_report_does_not_send_unsanitized_original(tmp_path): + terminal(tmp_path) + original = (tmp_path / "report.md").read_bytes() + real_write = suite_notify.write_report + + def write(path, text): + if path.name == "report.md": + raise PermissionError("fixture report write failure") + return real_write(path, text) + + with patch.object(suite_notify, "write_report", side_effect=write), \ + patch.object(suite_notify, "logged", return_value=0) as send: + assert suite_notify.send_report(tmp_path, ENV) == "SENT" + assert send.call_args.args[0][-1] == "./notification-report.md" + assert (tmp_path / "report.md").read_bytes() == original + assert "fixture-private-key" not in (tmp_path / "notification-report.md").read_text(encoding="utf-8") diff --git a/test-e2e/infra/tools/tests/test_gate_scope.py b/test-e2e/infra/tools/tests/test_gate_scope.py new file mode 100644 index 0000000000..c784e2e805 --- /dev/null +++ b/test-e2e/infra/tools/tests/test_gate_scope.py @@ -0,0 +1,209 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""A local prerequisite or declaration failure must not cancel independent Cases.""" +import argparse +import json +import os +from pathlib import Path +import shutil +import sys +from unittest.mock import patch + +import pytest +import yaml + +TOOLS = Path(__file__).resolve().parents[1] +REPO = TOOLS.parents[2] +sys.path.insert(0, str(TOOLS)) +import deployment_exceptions as modes +import suite +import validate_execution +import validate_test_assets +from local_preflight import static_config_problems +from suite_runtime import fingerprint, save +from validation_policy import report, apply_blocks + + +@pytest.fixture +def case_repo(tmp_path): + case_id = "AGENT-LIST-D1-001" + target = tmp_path / "test-e2e/cases" / case_id + shutil.copytree(REPO / "test-e2e/cases" / case_id, target) + return tmp_path, target, case_id + + +@pytest.mark.parametrize("field,value", [("timeout_seconds", 0), ("required_mock_services", ["unknown-service"]), + ("preparation", {"families": "invalid"})]) +def test_invalid_case_declaration_is_scoped(case_repo, field, value): + root, target, case_id = case_repo + schema = root / "test-e2e/infra/schemas" + schema.mkdir(parents=True) + shutil.copyfile(REPO / "test-e2e/infra/schemas/preparation.schema.json", schema / "preparation.schema.json") + path = target / "execution.yaml" + data = yaml.safe_load(path.read_text(encoding="utf-8")) + data[field] = value + path.write_text(yaml.safe_dump(data), encoding="utf-8") + issues, registry = validate_execution.inspect(root) + validation = report(root, issues, registry["cases"]) + assert validation["ready"] and validation["blocked_cases"] == [case_id] + assert apply_blocks(registry["cases"], validation)[0]["asset_blocked"] + + +def test_daily_does_not_reparse_blocked_execution_via_excel(case_repo): + root, target, case_id = case_repo + (target / "execution.yaml").write_text("invalid: [", encoding="utf-8") + with patch.object(validate_test_assets.validate_features, "validate", return_value=[]), \ + patch.object(validate_test_assets.validate_cases, "validate", return_value=[]), \ + patch.object(validate_test_assets.validate_changes, "validate", return_value=[]), \ + patch.object(validate_test_assets.validate_traceability, "validate", return_value=[]), \ + patch.object(validate_test_assets.ownership_catalog, "validate", return_value=[]), \ + patch.object(validate_test_assets.generate_feature_links, "expected_pages", return_value={}), \ + patch.object(validate_test_assets.generate_excel, "write_workbook", side_effect=AssertionError("No raw reparse")): + registry = {} + issues = validate_test_assets.validate(root, "implementation", False, daily=True, registry_out=registry) + validation = report(root, issues, registry["cases"]) + assert validation["ready"] and validation["blocked_cases"] == [case_id] + assert any(issue.code == "DERIVED_VIEW_STALE" for issue in issues) + assert validate_test_assets.validate(root, "implementation", False)[0].code == "CASE_EXECUTION_INVALID" + + +def test_unsafe_case_path_remains_fatal(case_repo): + root, target, _ = case_repo + path = target / "execution.yaml" + data = yaml.safe_load(path.read_text(encoding="utf-8")) + data["implementations"][0]["file"] = "../../outside.py" + path.write_text(yaml.safe_dump(data), encoding="utf-8") + issues, registry = validate_execution.inspect(root) + assert not report(root, issues, registry["cases"])["ready"] + + +def test_worker_checks_only_requested_case_not_unrelated_broken_yaml(case_repo): + root, _, case_id = case_repo + broken = root / "test-e2e/cases/UNRELATED/case.yaml" + broken.parent.mkdir() + broken.write_text("invalid: [", encoding="utf-8") + args = argparse.Namespace(case=[case_id], change=None, stage=None, feature=None) + assert [row["case_id"] for row in suite.selection(root, args, worker=True)] == [case_id] + with pytest.raises(yaml.YAMLError): + validate_execution.inspect(root) + + +def test_legacy_recovery_case_reference_cannot_block_campaign(tmp_path): + config = tmp_path / "test-e2e/infra/config" + config.mkdir(parents=True) + shutil.copyfile(REPO / (modes.CATALOG.removesuffix(".yaml") + ".schema.json"), + config / "deployment-exceptions.schema.json") + data = yaml.safe_load((REPO / modes.CATALOG).read_text(encoding="utf-8")) + records = [{"case_id": case_id} for case_id in data["cases"]] + for legacy in (None, [], ["REMOVED-RECOVERY-CASE"]): + data.pop("recovery_cases", None) + if legacy is not None: + data["recovery_cases"] = legacy + (tmp_path / modes.CATALOG).write_text(yaml.safe_dump(data), encoding="utf-8") + assert modes.catalog(tmp_path, records)["cases"] == data["cases"] + + +def test_common_config_check_ignores_unused_business_profiles(tmp_path): + config = tmp_path / "config" + config.mkdir() + (config / "daily.env").write_text("KEY=value\n") + secret = config / "secrets.env" + secret.write_text("KEY=value\n") + secret.chmod(0o600) + (config / "environment.yaml").write_text(yaml.safe_dump({"base_url": "http://localhost:3000", "services": { + name: {"url": "http://localhost:8000"} for name in ("config", "runtime", "northbound", "data_process")}})) + rows = [{"stage": "D2", "status": "active", "execution": {"preparation": {"families": []}}}] + assert static_config_problems(tmp_path, rows, repo=REPO, common_only=True) == [] + assert any("models.yaml" in issue for issue in static_config_problems(tmp_path, rows, repo=REPO)) + (config / "environment.yaml").write_text("base_url: invalid\n") + assert static_config_problems(tmp_path, rows, repo=REPO, common_only=True) + + +def test_runtime_check_is_reused_only_for_cases_using_that_framework(tmp_path): + rows = [{"case_id": case_id, "status": "active", "execution": { + "implementations": [{"framework": framework}]}} + for case_id, framework in (("A", "vitest"), ("B", "vitest"), ("C", "pytest"), ("D", "custom"))] + with patch.object(suite, "node_runtime_problems", return_value=["unusable jsdom"]) as probe: + assert suite.runtime_blocks(REPO, tmp_path, rows) == {"A": ["unusable jsdom"], "B": ["unusable jsdom"]} + probe.assert_called_once() + + +@pytest.mark.parametrize("full", [False, True]) +def test_deploy_doctor_checks_common_environment_unless_full_diagnostics_requested(tmp_path, capsys, full): + repo, home = tmp_path / "repo", tmp_path / "home" + with patch.object(suite, "repository_root", return_value=repo), \ + patch.object(suite, "config", return_value={}), \ + patch.object(suite, "selection", return_value=[]), \ + patch.object(suite, "doctor", return_value=[]) as doctor: + assert suite.main(["doctor", "--test-home", str(home), "--live", *(["--full"] if full else [])]) == 0 + doctor.assert_called_once_with(repo, home, [], live=True, common_only=not full) + assert json.loads(capsys.readouterr().out)["ready"] + + +def test_deploy_doctor_does_not_reject_a_blocked_case_during_selection(case_repo, capsys): + repo, target, _ = case_repo + (target / "execution.yaml").write_text("invalid: [", encoding="utf-8") + home = repo.parent / (repo.name + "-home") + with patch.object(suite, "repository_root", return_value=repo), \ + patch.object(suite, "config", return_value={}), \ + patch.object(suite, "doctor", return_value=[]) as doctor: + assert suite.main(["doctor", "--test-home", str(home), "--live"]) == 0 + assert doctor.call_args.args[2][0]["asset_blocked"] + assert json.loads(capsys.readouterr().out)["ready"] + + +def test_browser_probe_uses_actual_batch_channel_and_test_home(tmp_path): + import node_preflight + from types import SimpleNamespace + with patch.object(node_preflight.shutil, "which", return_value="node"), \ + patch.object(node_preflight.subprocess, "run", return_value=SimpleNamespace(returncode=0)) as run: + assert node_preflight.node_toolchain_problems(REPO, tmp_path, {"playwright"}, environment={ + "NEXENT_TEST_BROWSER_CHANNEL": "chrome", "PLAYWRIGHT_BROWSERS_PATH": "wrong-cache"}) == [] + assert run.call_args.args[0][-1] == "chrome" + assert run.call_args.kwargs["env"]["PLAYWRIGHT_BROWSERS_PATH"] == str(tmp_path / "runtime/playwright-browsers") + + +def test_missing_browser_blocks_only_browser_case_and_batch_continues(tmp_path): + rows = [{"case_id": name, "stage": "D1", "status": "active", "execution": { + "implementations": [{"framework": framework, "file": "unused", "selector": name}]}} + for name, framework in (("BROWSER", "playwright"), ("INDEPENDENT", "pytest"))] + calls = [] + + def run(command, cwd, env, log, timeout): + if "--case" in command: + case_id = command[command.index("--case") + 1] + calls.append(case_id) + batch = Path(command[command.index("--batch-dir") + 1]) + save(batch / "cases" / case_id / "receipt.json", {"case_id": case_id, "stage": "D1", "result": "PASS"}) + return 0 + + with patch.object(suite, "machine_environment", return_value=dict(os.environ)), \ + patch.object(suite, "fingerprint", return_value={"head": "fixed"}), \ + patch.object(suite, "static_requirements", return_value=set()), \ + patch.object(suite, "static_inventory", return_value=[]), \ + patch.object(suite, "node_runtime_problems", return_value=["missing browser"]), \ + patch.object(suite, "logged", side_effect=run): + assert suite.execute(tmp_path, tmp_path, rows, {"hooks": {}, "case_timeout_seconds": 20}) == 1 + batch = next((tmp_path / "runs/repository-daily").iterdir()) + summary = json.loads((batch / "summary.json").read_text()) + assert calls == ["INDEPENDENT"] + assert summary["counts"] == {"BLOCKED": 1, "PASS": 1} and summary["execution_complete"] + assert json.loads((batch / "cases/BROWSER/receipt.json").read_text())["reason"] == "missing browser" + + +def test_fingerprint_prunes_dependency_directories_but_checks_source_bytes(tmp_path): + asset = tmp_path / "test-e2e/cases/A/test.py" + asset.parent.mkdir(parents=True) + asset.write_text("one") + dependency = tmp_path / "test-e2e/infra/node_modules/deep/huge.py" + dependency.parent.mkdir(parents=True) + dependency.write_text("ignored") + with patch("suite_runtime.subprocess.check_output", side_effect=lambda argv, **kwargs: + "head" if "rev-parse" in argv else b""): + first = fingerprint(tmp_path) + with patch.object(Path, "rglob", side_effect=AssertionError("Do not traverse node_modules per Case")): + assert fingerprint(tmp_path) == first + previous = asset.stat() + asset.write_text("two") + os.utime(asset, ns=(previous.st_atime_ns, previous.st_mtime_ns)) + assert fingerprint(tmp_path) != first diff --git a/test-e2e/infra/tools/tests/test_runtime_environment_repair.py b/test-e2e/infra/tools/tests/test_runtime_environment_repair.py index 18fde5cf30..cf07bc6554 100644 --- a/test-e2e/infra/tools/tests/test_runtime_environment_repair.py +++ b/test-e2e/infra/tools/tests/test_runtime_environment_repair.py @@ -67,7 +67,8 @@ def run(argv, **kwargs): assert "jsdom" in calls[0][0][2] assert [Path(argv[-2]).parent for argv, _ in calls[1:]] == [REPO / "frontend", REPO / "test-e2e/infra/automation/d4"] assert [argv[-1] for argv, _ in calls[1:]] == ["", "chromium"] - assert 'channel: "chromium"' in (REPO / 'test-e2e/infra/automation/d4/playwright.config.ts').read_text(encoding='utf-8') + assert 'channel: process.env.NEXENT_TEST_BROWSER_CHANNEL || "chromium"' in ( + REPO / 'test-e2e/infra/automation/d4/playwright.config.ts').read_text(encoding='utf-8') @pytest.mark.parametrize("failure", ["resolution", "timeout", "permission"]) diff --git a/test-e2e/infra/tools/tests/test_suite_runtime.py b/test-e2e/infra/tools/tests/test_suite_runtime.py index 9f54717a0a..097f7fc5cc 100644 --- a/test-e2e/infra/tools/tests/test_suite_runtime.py +++ b/test-e2e/infra/tools/tests/test_suite_runtime.py @@ -135,6 +135,37 @@ def test_notification_refuses_nonterminal(self): self.assertEqual(suite_notify.send_report(directory, {}), "NOT_TERMINAL") send.assert_not_called() + def test_case_failures_continue_unless_an_environment_fault_marker_remains(self): + for outcome, fault in (("FAIL", False), ("TIMEOUT", False), ("CLEANUP_FAILED", False), ("PASS", True)): + with self.subTest(outcome=outcome, fault=fault), tempfile.TemporaryDirectory() as temporary: + home = Path(temporary) + records = [{"case_id": case_id, "stage": "D5", "status": "active", "execution": {"test": True}} + for case_id in ("FIRST", "INDEPENDENT")] + calls = [] + + def execute(command, cwd, env, log, timeout): + if "--case" in command: + case_id = command[command.index("--case") + 1] + batch = Path(command[command.index("--batch-dir") + 1]) + calls.append(case_id) + save(batch / "cases" / case_id / "receipt.json", { + "case_id": case_id, "stage": "D5", "result": outcome if case_id == "FIRST" else "PASS"}) + if fault: + save(home / "state/deployment-guard/pending.json", {"status": "PENDING"}) + return 0 + + with patch.object(suite, "machine_environment", return_value=dict(os.environ)), \ + patch.object(suite, "fingerprint", return_value={"head": "test"}), \ + patch.object(suite, "static_requirements", return_value=set()), \ + patch.object(suite, "static_inventory", return_value=[]), \ + patch.object(suite, "logged", side_effect=execute): + self.assertEqual(suite.execute(home, home, records, {"hooks": {}, "case_timeout_seconds": 20}), 1) + batch = next((home / "runs/repository-daily").iterdir()) + summary = json.loads((batch / "summary.json").read_text()) + self.assertEqual(calls, ["FIRST"] if fault else ["FIRST", "INDEPENDENT"]) + self.assertEqual(summary["execution_complete"], not fault) + self.assertEqual(summary["counts"].get(outcome), 1) + def test_notification_deduplicates_uncertain_delivery(self): with tempfile.TemporaryDirectory() as root: directory = Path(root) @@ -284,7 +315,7 @@ def run(record, repo, home, env, *, result_dir): self.assertIn("STEP-02", receipt["reason"]) self.assertEqual(receipt["test_evidence"], ["d4/PW-TEST-01/trace.zip"]) - def test_d5_cleanup_failure_stops_remaining_cases_and_finalizes_incomplete(self): + def test_unresolved_d5_environment_fault_stops_remaining_cases_and_finalizes_incomplete(self): with tempfile.TemporaryDirectory() as root: home = Path(root) records = [{"case_id": name, "stage": "D5", "status": "active", "execution": {"test": True}} @@ -299,6 +330,7 @@ def command_result(command, cwd, env, log, timeout): save(batch / "cases" / case_id / "receipt.json", { "case_id": case_id, "stage": "D5", "result": "CLEANUP_FAILED", "test_result": "PASS", "cleanup_exit_code": 1}) + save(home / "state/deployment-guard/pending.json", {"status": "PENDING"}) return 1 return 0 diff --git a/test-e2e/infra/tools/tests/test_terminal_notifications.py b/test-e2e/infra/tools/tests/test_terminal_notifications.py index 01d4d5ea7e..f9f3eb46b9 100644 --- a/test-e2e/infra/tools/tests/test_terminal_notifications.py +++ b/test-e2e/infra/tools/tests/test_terminal_notifications.py @@ -26,6 +26,9 @@ spec = importlib.util.spec_from_file_location("terminal_daily_launcher", TOOLS.parent / "scripts/daily-launch.py") launcher = importlib.util.module_from_spec(spec) spec.loader.exec_module(launcher) +worker_spec = importlib.util.spec_from_file_location("terminal_daily_worker", TOOLS.parent / "scripts/daily-worker.py") +worker = importlib.util.module_from_spec(worker_spec) +worker_spec.loader.exec_module(worker) ENV = {"LARK_CLI": "offline-cli", "FEISHU_REPORT_CHAT_ID": "offline-chat"} @@ -47,14 +50,15 @@ def test_every_terminal_state_sends_and_preserves_results(tmp_path, status): assert suite_notify.send_report(tmp_path, ENV) == "SENT" assert suite_notify.send_report(tmp_path, ENV) == "SENT" send.assert_called_once() - text = (tmp_path / "notification-report.md").read_text(encoding="utf-8") + text = (tmp_path / "report.md").read_text(encoding="utf-8") assert f"批次终态：{status}" in text assert "环境恢复状态：" in text assert "Original report" in text if status == "INTERRUPTED": assert "Interrupted" in text and "NOT_EXECUTED=1" in text assert (tmp_path / "summary.json").read_bytes() == original - assert send.call_args.args[0][-1] == "./notification-report.md" + assert send.call_args.args[0][-1] == "./report.md" + assert not (tmp_path / "notification-report.md").exists() @pytest.mark.parametrize("verified", [False, True]) @@ -66,10 +70,10 @@ def test_restored_configuration_is_not_mistaken_for_verified_recovery(tmp_path, save(tmp_path / "status.json", state) with patch.object(suite_notify, "logged", return_value=0): assert suite_notify.send_report(tmp_path, ENV) == "SENT" - text = (tmp_path / "notification-report.md").read_text(encoding="utf-8") + text = (tmp_path / "report.md").read_text(encoding="utf-8") assert "restore-local" in text - assert ("恢复用例验证通过" in text) is verified - assert ("恢复用例未通过或尚未完成" in text) is (not verified) + assert ("部署模式、服务检查通过" in text) is verified + assert ("环境恢复检查未通过或尚未完成" in text) is (not verified) def test_missing_report_still_sends_saved_terminal_reason(tmp_path): @@ -83,6 +87,15 @@ def test_missing_report_still_sends_saved_terminal_reason(tmp_path): assert "未记录恢复验证证据" in text +def test_prepared_launch_reports_pending_redeployment_instead_of_recovered_environment(tmp_path): + receipt(tmp_path, "INCOMPLETE", error="worker could not start") + save(tmp_path / "recovery.json", {"status": "PREPARED_FOR_REDEPLOY"}) + with patch.object(suite_notify, "logged", return_value=0): + assert suite_notify.send_report(tmp_path, ENV) == "SENT" + text = (tmp_path / "report.md").read_text(encoding="utf-8") + assert "待重新部署及恢复验证" in text and "前批恢复标记仍待处理" in text + + @pytest.mark.parametrize("delivery", [124, RuntimeError("offline transport failure")]) def test_uncertain_or_failed_delivery_is_never_blindly_repeated(tmp_path, delivery): receipt(tmp_path, "INCOMPLETE", error="Interrupted") @@ -117,11 +130,11 @@ def test_notification_redacts_known_secret_without_rewriting_local_evidence(tmp_ receipt(tmp_path, "INCOMPLETE", error="Service failed with fake-secret-value") with patch.object(suite_notify, "logged", return_value=0): assert suite_notify.send_report(tmp_path, {**ENV, "TEST_API_KEY": "fake-secret-value"}) == "SENT" - assert "fake-secret-value" not in (tmp_path / "notification-report.md").read_text(encoding="utf-8") + assert "fake-secret-value" not in (tmp_path / "report.md").read_text(encoding="utf-8") assert "fake-secret-value" in (tmp_path / "summary.json").read_text(encoding="utf-8") -@pytest.mark.parametrize("failure", [None, "build", "interrupt", "recovery-probe", "restore-local", "publish"]) +@pytest.mark.parametrize("failure", [None, "build", "interrupt", "case-timeout", "restore-local", "publish"]) def test_campaign_notifies_for_failure_and_recovery_interruptions(tmp_path, failure): repo, home, env_path, original, records, settings = campaign_fixture(tmp_path) settings["hooks"]["build"] = [["offline-build"]] @@ -143,7 +156,7 @@ def execute(repo, home, selected, settings, *, directory, on_result=None): raise KeyboardInterrupt() directory.mkdir(parents=True) rows = [{"case_id": r["case_id"], "stage": r["stage"], "result": - "TIMEOUT" if failure == "recovery-probe" and directory.name == "restore-local" else "PASS"} + "TIMEOUT" if failure == "case-timeout" and r["case_id"] == "PW-KB-LOCAL-01" else "PASS"} for r in selected] for row in rows: if on_result: @@ -158,8 +171,8 @@ def execute(repo, home, selected, settings, *, directory, on_result=None): patch("report_ownership.enrich", side_effect=lambda repo, rows, head: (rows, [])), \ patch("suite_publish.publish_latest", side_effect=OSError("offline publication failure")) as publish, \ patch.object(suite_notify, "logged", return_value=0) as send: - code = modes.execute_campaign(repo, home, records[:2], settings, daily=True, - publish_latest=failure in {"publish", "recovery-probe", "restore-local"}, + code = modes.execute_campaign(repo, home, records if failure == "case-timeout" else records[:2], settings, daily=True, + publish_latest=failure in {"publish", "case-timeout", "restore-local"}, execute_one=execute, all_records=records, hook=hook, machine_env=lambda *_: env, on_batch_created=created.append) assert len(created) == 1 @@ -168,15 +181,16 @@ def execute(repo, home, selected, settings, *, directory, on_result=None): assert json.loads((directory / "notification.json").read_text())["status"] == "SENT" summary = json.loads((directory / "summary.json").read_text()) assert code == (0 if failure in {None, "publish"} else 1) - assert summary["execution_complete"] is (failure in {None, "publish"}) - text = (directory / "notification-report.md").read_text(encoding="utf-8") - if failure in {"recovery-probe", "restore-local"}: + assert summary["execution_complete"] is (failure in {None, "publish", "case-timeout"}) + text = (directory / "report.md").read_text(encoding="utf-8") + if failure == "restore-local": publish.assert_not_called() assert prior_report.read_text(encoding="utf-8") == "previous complete batch" assert "环境可用性未验证" in text or "不能认为环境已恢复可用" in text - if failure == "recovery-probe": - assert summary["counts"] == {"PASS": 2} # The separate recovery probe is not a third business case. - assert "restore-local" in text + if failure == "case-timeout": + assert summary["counts"] == {"PASS": 2, "TIMEOUT": 1} + publish.assert_called_once() + assert "部署模式、服务检查通过" in text assert env_path.read_bytes() == original assert text.count("## 终态与环境恢复") == 1 @@ -223,6 +237,7 @@ def test_launcher_reports_preflight_failures_without_product_configuration(tmp_p monkeypatch.setattr(launcher.sys, "executable", str(runtime)) monkeypatch.setattr(launcher, "deployment_lock", lambda *_: nullcontext()) monkeypatch.setattr(launcher, "prepare_daily_retry", lambda *_: nullcontext()) + monkeypatch.setattr(launcher, "notify_terminal", lambda repo, home, directory: suite_notify.notify_launch(directory, home)) def sync(*args, **kwargs): if failure == "signal": @@ -277,9 +292,15 @@ def test_launcher_uses_explicit_batch_link_or_reports_preparation_failure(tmp_pa monkeypatch.setattr(launcher.sys, "executable", str(runtime)) monkeypatch.setattr(launcher, "deployment_lock", lambda *_: nullcontext()) monkeypatch.setattr(launcher, "prepare_daily_retry", lambda *_: nullcontext()) + monkeypatch.setattr(launcher, "notify_terminal", lambda repo, home, directory: suite_notify.notify_launch(directory, home)) monkeypatch.setattr(launcher, "sync_checkout", lambda *a, **k: "fixed-revision") monkeypatch.setattr(launcher, "git", lambda *a, **k: "") - monkeypatch.setattr(launcher, "logged", lambda *a, **k: 1 if failure == "validation" else 0) + monkeypatch.setattr(worker, "logged", lambda *a, **k: 1 if failure == "validation" else 0) + monkeypatch.setattr(worker, "checked_revision", lambda *a: None) + monkeypatch.setattr(worker, "inherited_locks", lambda *a: nullcontext()) + monkeypatch.setattr(worker, "prepare_daily_retry", lambda *a, **k: nullcontext()) + monkeypatch.setattr(launcher, "run_worker", lambda repo, home, directory, head, publish, locks: + worker.execute(repo, home, directory, head, publish)) monkeypatch.setattr(static_assets, "apply", lambda *a, **k: None) monkeypatch.setattr(suite, "selection", lambda *a, **k: []) monkeypatch.setattr(suite, "doctor", lambda *a, **k: ["offline missing browser"] if failure == "doctor" else []) diff --git a/test-e2e/infra/tools/validate_execution.py b/test-e2e/infra/tools/validate_execution.py index 1dee10b6f9..d14f9a47d0 100644 --- a/test-e2e/infra/tools/validate_execution.py +++ b/test-e2e/infra/tools/validate_execution.py @@ -44,8 +44,8 @@ def _mock_service_issues(execution_path, raw) -> list[ValidationIssue]: return issues -def inspect(root: Path, phase: str = "implementation", *, require_preparation_cases=()) -> tuple[list[ValidationIssue], dict]: - cases, issues = case_index(discover_documents(root, "cases")) +def inspect(root: Path, phase: str = "implementation", *, require_preparation_cases=(), case_ids=None) -> tuple[list[ValidationIssue], dict]: + cases, issues = case_index(discover_documents(root, "cases", case_ids=case_ids)) entries = {} entrypoints: dict[str, str] = {} for case_id, (stage, case, case_path) in sorted(cases.items()): @@ -56,7 +56,11 @@ def inspect(root: Path, phase: str = "implementation", *, require_preparation_ca "contract_hash": case_contract_hash(stage, case), "execution": None, } entries[case_id] = record - issues.extend(ValidationIssue(case_path, "acceptance", problem) for problem in contract_issues(case)) + def invalid(location, message): + return ValidationIssue(execution_path, location, message, "CASE_EXECUTION_INVALID", case_id) + + issues.extend(ValidationIssue(case_path, "acceptance", problem, "CASE_EXECUTION_INVALID", case_id) + for problem in contract_issues(case)) if "acceptance" in case: record["acceptance"] = case["acceptance"] if not execution_path.is_file(): @@ -64,21 +68,25 @@ def inspect(root: Path, phase: str = "implementation", *, require_preparation_ca issues.append(ValidationIssue(case_path, case_id, "Active automated case has no execution.yaml", "CASE_EXECUTION_MISSING", case_id)) continue - raw = __import__("yaml").safe_load(execution_path.read_text(encoding="utf-8")) + try: + raw = yaml.safe_load(execution_path.read_text(encoding="utf-8")) + except (yaml.YAMLError, UnicodeError): + issues.append(invalid("", "Invalid execution YAML")) + continue if not isinstance(raw, dict) or set(raw) - EXECUTION_FIELDS or raw.get("schema_version") != "1.0": - issues.append(ValidationIssue(execution_path, "", "Invalid execution document fields or version")) + issues.append(invalid("", "Invalid execution document fields or version")) continue if raw.get("case_id") != case_id: - issues.append(ValidationIssue(execution_path, "case_id", "Execution Case ID differs from its directory")) - issues.extend(_mock_service_issues(execution_path, raw)) + issues.append(invalid("case_id", "Execution Case ID differs from its directory")) + issues.extend(invalid(issue.location, issue.message) for issue in _mock_service_issues(execution_path, raw)) required_acceptance = phase == "implementation" and case.get("automation") == "automated" - issues.extend(ValidationIssue(execution_path, "acceptance_bindings", problem) + issues.extend(invalid("acceptance_bindings", problem) for problem in binding_issues(case, raw.get("acceptance_bindings"), required=required_acceptance)) timeout_seconds = raw.get("timeout_seconds") if timeout_seconds is not None and (type(timeout_seconds) is not int or not 1 <= timeout_seconds <= 7200): - issues.append(ValidationIssue(execution_path, "timeout_seconds", "Case timeout must be an integer from 1 to 7200 seconds")) + issues.append(invalid("timeout_seconds", "Case timeout must be an integer from 1 to 7200 seconds")) if case.get("status") in {"retired", "skipped_by_policy", "manual"}: - issues.append(ValidationIssue(execution_path, "", "Inactive or manual case must not have executable metadata")) + issues.append(invalid("", "Inactive or manual case must not have executable metadata")) if (phase == "implementation" and stage != "D1" and case.get("status") == "active" and "preparation" not in raw): @@ -87,15 +95,16 @@ def inspect(root: Path, phase: str = "implementation", *, require_preparation_ca "CASE_PREPARATION_MISSING", case_id)) implementations = raw.get("implementations") if not isinstance(implementations, list) or not implementations: - issues.append(ValidationIssue(execution_path, "implementations", "At least one implementation is required")) + issues.append(invalid("implementations", "At least one implementation is required")) continue normalized = [] for index, item in enumerate(implementations): if not isinstance(item, dict) or set(item) - IMPLEMENTATION_FIELDS or not {"framework", "file", "selector"}.issubset(item): - issues.append(ValidationIssue(execution_path, f"implementations/{index}", "Invalid implementation fields")) + issues.append(invalid(f"implementations/{index}", "Invalid implementation fields")) continue - if item["framework"] not in FRAMEWORKS or not isinstance(item["file"], str) or not isinstance(item["selector"], str): - issues.append(ValidationIssue(execution_path, f"implementations/{index}", "Invalid framework, file or selector")) + if (not isinstance(item["framework"], str) or item["framework"] not in FRAMEWORKS + or not isinstance(item["file"], str) or not isinstance(item["selector"], str)): + issues.append(invalid(f"implementations/{index}", "Invalid framework, file or selector")) continue if index == 0: resolved = (execution_path.parent / item["file"]).resolve() @@ -137,13 +146,13 @@ def inspect(root: Path, phase: str = "implementation", *, require_preparation_ca schema = json.loads((root / "test-e2e/infra/schemas/preparation.schema.json").read_text(encoding="utf-8")) errors = list(Draft202012Validator(schema).iter_errors(raw["preparation"])) if errors: - issues.append(ValidationIssue(execution_path, "preparation", "Invalid preparation declaration")) + issues.append(invalid("preparation", "Invalid preparation declaration")) else: record["execution"]["preparation"] = raw["preparation"] registry = {"schema_version": "1.0", "cases": [entries[key] for key in sorted(entries)]} from deployment_exceptions import CATALOG, catalog try: - catalog(root, registry["cases"]) + catalog(root, registry["cases"] if case_ids is None else None) except (ValueError, OSError, yaml.YAMLError): issues.append(ValidationIssue(root / CATALOG, "deployment_exceptions", "Invalid deployment exception catalog or Case/profile reference")) diff --git a/test-e2e/infra/tools/validate_test_assets.py b/test-e2e/infra/tools/validate_test_assets.py index 3efb2e4077..2625e31acc 100644 --- a/test-e2e/infra/tools/validate_test_assets.py +++ b/test-e2e/infra/tools/validate_test_assets.py @@ -23,7 +23,7 @@ def validate(root: Path, phase: str, regenerate: bool, require_acceptance_cases: list[str] | None = None, require_ut_changes: list[str] | None = None, - require_preparation_cases: list[str] | None = None, *, daily=False) -> list[ValidationIssue]: + require_preparation_cases: list[str] | None = None, *, daily=False, registry_out=None) -> list[ValidationIssue]: if daily and regenerate: raise ValueError("Daily validation must not regenerate repository assets") issues: list[ValidationIssue] = [] @@ -60,9 +60,11 @@ def validate(root: Path, phase: str, regenerate: bool, issues.append(ValidationIssue(root / "test-e2e/cases", case_id, "Selected preparation Case does not exist")) execution_issues, registry = validate_execution.inspect( root, phase, require_preparation_cases=require_preparation_cases or ()) + if registry_out is not None: + registry_out.update(registry) issues.extend(execution_issues) + from validation_policy import report if issues: - from validation_policy import report if not daily or not report(root, issues, registry["cases"])["ready"]: return issues for path, expected in generate_feature_links.expected_pages(root).items(): @@ -80,6 +82,11 @@ def validate(root: Path, phase: str, regenerate: bool, excel_path = root / "test-e2e/infra/generated/Nexent_测试基线.xlsx" if regenerate: generate_excel.write_workbook(root, excel_path) + elif daily and report(root, issues, registry["cases"])["blocked_cases"]: + # The Excel reader consumes raw execution metadata. It must not turn a + # scoped parse/declaration failure back into a global launch failure. + issues.append(ValidationIssue(excel_path, "", "Excel comparison unavailable while Case execution metadata is blocked", + "DERIVED_VIEW_STALE")) else: with tempfile.TemporaryDirectory() as temporary_directory: expected = Path(temporary_directory) / "expected.xlsx" @@ -110,10 +117,12 @@ def main() -> int: from suite_runtime import save from validation_policy import report try: + registry = {} issues = validate(root, args.phase, False, args.require_acceptance_case, args.require_ut_change, - args.require_preparation_case, daily=True) - execution_issues, registry = validate_execution.inspect(root) - issues = list(dict.fromkeys([*issues, *execution_issues])) + args.require_preparation_case, daily=True, registry_out=registry) + if "cases" not in registry: + execution_issues, registry = validate_execution.inspect(root) + issues = list(dict.fromkeys([*issues, *execution_issues])) except Exception as exc: # Unknown parser/IO failures are global; do not expose arbitrary YAML values. issues = [ValidationIssue(root / "test-e2e", "", f"Asset inspection failed ({type(exc).__name__})")] diff --git a/test-e2e/infra/tools/validation_policy.py b/test-e2e/infra/tools/validation_policy.py index 434ba334af..a449f170e4 100644 --- a/test-e2e/infra/tools/validation_policy.py +++ b/test-e2e/infra/tools/validation_policy.py @@ -3,7 +3,8 @@ """Classify only explicitly typed, safely scoped Daily asset problems.""" from __future__ import annotations -CASE_CODES = {"CASE_EXECUTION_MISSING", "CASE_SCRIPT_MISSING", "CASE_SELECTOR_MISSING", "CASE_PREPARATION_MISSING"} +CASE_CODES = {"CASE_EXECUTION_MISSING", "CASE_SCRIPT_MISSING", "CASE_SELECTOR_MISSING", "CASE_PREPARATION_MISSING", + "CASE_EXECUTION_INVALID"} WARNING_CODES = {"UT_REFERENCE_MISSING", "DERIVED_VIEW_STALE", "FEATURE_RESPONSIBILITY_UNRESOLVED"} CASE_WARNING_CODES = {"CASE_RESPONSIBILITY_UNRESOLVED", "CASE_ID_FORMAT_INVALID"} @@ -39,8 +40,10 @@ def evaluate(root): from validate_execution import inspect from validate_test_assets import validate - issues = validate(root, "implementation", False, daily=True) - execution_issues, registry = inspect(root) - issues = list(dict.fromkeys([*issues, *execution_issues])) + registry = {} + issues = validate(root, "implementation", False, daily=True, registry_out=registry) + if "cases" not in registry: + execution_issues, registry = inspect(root) + issues = list(dict.fromkeys([*issues, *execution_issues])) validation = report(root, issues, registry["cases"]) return validation, apply_blocks(registry["cases"], validation)