diff --git a/deploy/windows-service/install.ps1 b/deploy/windows-service/install.ps1 index e969486..0a80f71 100644 --- a/deploy/windows-service/install.ps1 +++ b/deploy/windows-service/install.ps1 @@ -4,6 +4,8 @@ param( [string]$ProjectRoot = (Resolve-Path (Join-Path $PSScriptRoot "..\..")).Path, [string]$DataRoot = "", + [string]$RuntimeEnvFile = "D:\product-collector-analyze-flow\.env", + [string]$JushuitanEnvFile = "D:\product-collector-analyze-flow\private-secrets\jushuitan.env", [string]$ServiceName = "gyxx-flow-scheduler", [string]$NssmPath = "nssm.exe" ) @@ -12,6 +14,13 @@ $ErrorActionPreference = "Stop" Set-StrictMode -Version Latest $project = (Resolve-Path -LiteralPath $ProjectRoot).Path +$runtimeEnvPath = (Resolve-Path -LiteralPath $RuntimeEnvFile -ErrorAction Stop).Path +$runtimeEnvPaths = @($runtimeEnvPath) +if (Test-Path -LiteralPath $JushuitanEnvFile -PathType Leaf) { + $runtimeEnvPaths += (Resolve-Path -LiteralPath $JushuitanEnvFile -ErrorAction Stop).Path +} else { + Write-Warning "Jushuitan environment file not found; Jushuitan workflows will fail safely until it is provisioned." +} if ([string]::IsNullOrWhiteSpace($DataRoot)) { $DataRoot = Join-Path $project "var" } @@ -33,7 +42,9 @@ function Invoke-Nssm([string[]]$Arguments) { Invoke-Nssm -Arguments @("install", $ServiceName, $python) Invoke-Nssm -Arguments @("set", $ServiceName, "AppDirectory", $project) -Invoke-Nssm -Arguments @("set", $ServiceName, "AppParameters", "-m gyxx_flow schedule run") +$envFileArguments = $runtimeEnvPaths | ForEach-Object { '--env-file "' + $_ + '"' } +$schedulerArguments = '-m gyxx_flow schedule run ' + ($envFileArguments -join ' ') +Invoke-Nssm -Arguments @("set", $ServiceName, "AppParameters", $schedulerArguments) Invoke-Nssm -Arguments @( "set", $ServiceName, "AppEnvironmentExtra", "GYXX_PROJECT_ROOT=$project", "GYXX_DATA_ROOT=$data", "PYTHONUTF8=1" diff --git a/src/gyxx_flow/adapters/browser.py b/src/gyxx_flow/adapters/browser.py index e9fbad4..3729ea5 100644 --- a/src/gyxx_flow/adapters/browser.py +++ b/src/gyxx_flow/adapters/browser.py @@ -168,3 +168,22 @@ def looks_like_login_url(url: str) -> bool: path = parsed.path.casefold() markers = ("login", "passport", "signin", "sign-in") return any(marker in hostname or marker in path for marker in markers) + + +class CookieExpiredError(RuntimeError): + """A platform login state expired and no automatic recovery exists. + + Collectors raise this instead of waiting for a human QR/SMS login: the + account cookie gate has already blocked unverified accounts before the + run, so reaching this point means the session died mid-run. The CLI maps + it to ``COOKIE_SKIP_EXIT_CODE`` so the engine records a cookie skip. + """ + + +def cookie_expired_message(account_id: str, platform: str) -> str: + """One standard, operator-actionable expiry message across platforms.""" + + return ( + f"{platform} 登录态已失效：请执行 `uv run gyxx accounts login {account_id}` " + "完成扫码/验证码重新登录（或等待飞书通知后处理）" + ) diff --git a/src/gyxx_flow/catalog.py b/src/gyxx_flow/catalog.py index 16519e5..e7efb25 100644 --- a/src/gyxx_flow/catalog.py +++ b/src/gyxx_flow/catalog.py @@ -79,6 +79,10 @@ class WorkflowEntry: steps: tuple[WorkflowStepEntry, ...] = () notification_workflow_id: str | None = None topic_config: WorkflowTopicConfig | None = None + # Orchestrator entries whose own binding carries no runtime account while + # the spawned child collectors do; the cookie gate probes these before + # the workflow runs. + required_accounts: tuple[str, ...] = () @dataclass(frozen=True, slots=True) @@ -218,6 +222,9 @@ def _parse_workflow(item: Any) -> WorkflowEntry: if trigger == "scheduled" and (not isinstance(task_name, str) or not task_name): raise CatalogError(f"scheduled workflow requires source task name: {workflow_id}") topic_config = _parse_topic_config(item.get("topic_config"), workflow_id) + required_accounts = _parse_required_accounts( + item.get("required_accounts", []), workflow_id + ) return WorkflowEntry( workflow_id=workflow_id, module=module, @@ -229,9 +236,18 @@ def _parse_workflow(item: Any) -> WorkflowEntry: note=item.get("note"), steps=steps, topic_config=topic_config, + required_accounts=required_accounts, ) +def _parse_required_accounts(value: Any, workflow_id: str) -> tuple[str, ...]: + if not isinstance(value, list) or not all( + isinstance(item, str) and _WORKFLOW_ID.fullmatch(item) for item in value + ): + raise CatalogError(f"invalid required_accounts for {workflow_id}") + return tuple(dict.fromkeys(value)) + + def _parse_topic_config(value: Any, workflow_id: str) -> WorkflowTopicConfig | None: if value is None: return None diff --git a/src/gyxx_flow/cli.py b/src/gyxx_flow/cli.py index 7e519c1..4833b98 100644 --- a/src/gyxx_flow/cli.py +++ b/src/gyxx_flow/cli.py @@ -703,7 +703,19 @@ def _python_scheduler( if enable_account_keepalive and not dry_run: from gyxx_flow.accounts import AccountKeepaliveManager - maintenance = AccountKeepaliveManager(settings).tick + keepalive_manager = AccountKeepaliveManager(settings) + recovery_manager = _build_cookie_recovery_manager( + settings, launcher=launcher, keepalive=keepalive_manager + ) + + def maintenance() -> None: # noqa: D107 - scheduler maintenance chain + keepalive_manager.tick() + if recovery_manager is not None: + try: + recovery_manager.tick() + except Exception: # noqa: BLE001 - 自动重跑不得影响调度心跳 + pass + return PythonScheduler( catalog, settings.data_root, @@ -717,6 +729,29 @@ def _python_scheduler( ) +def _build_cookie_recovery_manager( + settings: Settings, + *, + launcher: object, + keepalive: object, +): + """Build the cookie-gate recovery rerun manager; None when disabled.""" + + from gyxx_flow.cookie_gate import ( + CookieGate, + CookieGateRecoveryManager, + gate_enabled, + ) + + if not gate_enabled(): + return None + try: + gate = CookieGate(settings, keepalive=keepalive) # type: ignore[arg-type] + except Exception: # noqa: BLE001 - 无集成配置时没有可恢复的账号 + return None + return CookieGateRecoveryManager(gate, launcher=launcher) + + def _schedule_run_command( arguments: argparse.Namespace, catalog: WorkflowCatalog, @@ -885,6 +920,11 @@ def _execute_once( dry_run: bool, settings: Settings, ) -> tuple[int, dict[str, object]]: + if not dry_run: + # 每账号 Cookie 健康门控：dry-run 不注入，也不做任何探测/通知。 + from gyxx_flow.cookie_gate import apply_cookie_gates + + workflow = apply_cookie_gates(workflow, settings=settings) context = RunContext.create( workflow.workflow_id, business_date.isoformat(), diff --git a/src/gyxx_flow/console.py b/src/gyxx_flow/console.py index aba0c9e..a0055ef 100644 --- a/src/gyxx_flow/console.py +++ b/src/gyxx_flow/console.py @@ -5,6 +5,7 @@ from __future__ import annotations import calendar import hashlib import hmac +import http.client import ipaddress import json import os @@ -15,6 +16,7 @@ import sys import tempfile import threading import time +import traceback import unicodedata import uuid from contextlib import suppress @@ -24,7 +26,7 @@ from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from importlib import resources from pathlib import Path -from typing import Any, Mapping, Protocol, Sequence, TextIO, cast +from typing import Any, Callable, Mapping, Protocol, Sequence, TextIO, cast from urllib.parse import parse_qs, unquote, urlsplit from zoneinfo import ZoneInfo @@ -171,7 +173,6 @@ _WORKFLOW_NAMES = { "content.metrics.daily": "合作达人内容指标日报采集", "content.metrics.self_operated.daily": "自营内容指标日报采集", "content.marketing_report.daily": "营销日报生成与发送", - "content.relogin.weekly": "内容平台登录态刷新", "content.creator_report.monthly": "达人月报生成", "content.summary.monthly": "内容月度汇总", "content.comments.weekly": "多平台评论周采集", @@ -3736,6 +3737,7 @@ class WorkflowConsoleHTTPServer(ThreadingHTTPServer): self.console_service = service self.console_token = token self.bind_host = bind_host + self.started_at = time.time() super().__init__(server_address, WorkflowConsoleRequestHandler) def handle_error( @@ -3796,6 +3798,29 @@ class WorkflowConsoleRequestHandler(BaseHTTPRequestHandler): if method in {"GET", "HEAD"} and path == "/assets/app.js": self._static("app.js", "text/javascript; charset=utf-8", head=method == "HEAD") return + if method == "GET" and path == "/api/health": + # 自检探针端点：不鉴权、不触外部依赖，只反映进程与调度锁存活， + # 供进程内看门狗与 deploy/watchdog 保活脚本探测。 + try: + scheduler = _scheduler_process_from_lock( + self.console_server.console_service.settings + ) + except Exception: + scheduler = None + self._json( + HTTPStatus.OK, + { + "ok": True, + "pid": os.getpid(), + "started_at": self.console_server.started_at, + "uptime_seconds": round( + time.time() - self.console_server.started_at, 1 + ), + "scheduler_running": scheduler is not None, + "scheduler_pid": scheduler.pid if scheduler else None, + }, + ) + return if method == "GET" and path == "/api/overview": self._require_api_access(mutation=False) self._json(HTTPStatus.OK, self.console_server.console_service.overview()) @@ -4020,6 +4045,9 @@ class WorkflowConsoleRequestHandler(BaseHTTPRequestHandler): except (BrokenPipeError, ConnectionResetError): return except Exception: + # 前端只拿到固定文案；完整堆栈落到 stderr（随启动重定向进 + # var/logs/console-restarts 或服务日志），否则故障无法定位。 + traceback.print_exc() self._json_error(HTTPStatus.INTERNAL_SERVER_ERROR, "控制台处理请求失败") def _require_api_access(self, *, mutation: bool) -> None: @@ -4190,6 +4218,92 @@ def create_console_server( ) +CONSOLE_WATCHDOG_EXIT_CODE = 70 + + +def _probe_console_health(port: int, timeout_seconds: float) -> bool: + # 只连本机回环端口的固定路径 /api/health，探测自身 HTTP 服务往返。 + connection = http.client.HTTPConnection("127.0.0.1", port, timeout=timeout_seconds) + try: + connection.request("GET", "/api/health") + response = connection.getresponse() + response.read() + return response.status == HTTPStatus.OK + except Exception: + return False + finally: + connection.close() + + +def _console_watchdog( + port: int, + *, + interval_seconds: float, + failure_limit: int, + probe_timeout_seconds: float, + stopped: threading.Event, + probe: Callable[[int, float], bool] | None = None, + exit_process: Callable[[int], None] = os._exit, + sleep: Callable[[float], None] = time.sleep, +) -> None: + """进程内自检：周期探测自身 /api/health，连续失败则退出交给守护拉起。 + + HTTP accept 线程活着但工作线程全部卡死（锁竞争/句柄耗尽）时，外部 + 静态页能打开而 API 全挂——正是"控制台暂时不可用"的形态；只有真实 + 走一遍 HTTP 往返才能暴露，所以探针必须打 socket 而不是查内存状态。 + """ + run_probe = probe or _probe_console_health + failures = 0 + while not stopped.is_set(): + if run_probe(port, probe_timeout_seconds): + failures = 0 + else: + failures += 1 + print( + f"⚠️ 控制台自检失败 {failures}/{failure_limit}: 127.0.0.1:{port}/api/health", + file=sys.stderr, + flush=True, + ) + if failures >= failure_limit: + print( + "❌ 控制台自检连续失败，进程退出等待守护脚本拉起", + file=sys.stderr, + flush=True, + ) + exit_process(CONSOLE_WATCHDOG_EXIT_CODE) + return + sleep(interval_seconds) + + +def _start_console_watchdog(server: WorkflowConsoleHTTPServer) -> threading.Event: + interval = max( + 5.0, float(os.environ.get("GYXX_CONSOLE_WATCHDOG_INTERVAL_SECONDS", "30")) + ) + limit = int(os.environ.get("GYXX_CONSOLE_WATCHDOG_FAILURES", "3")) + enabled = ( + os.environ.get("GYXX_CONSOLE_WATCHDOG", "1").strip().lower() + not in {"0", "false", "no", "off"} + and limit > 0 + ) + stopped = threading.Event() + if not enabled: + return stopped + thread = threading.Thread( + target=_console_watchdog, + args=(server.server_port,), + kwargs={ + "interval_seconds": interval, + "failure_limit": limit, + "probe_timeout_seconds": min(5.0, interval), + "stopped": stopped, + }, + name="gyxx-console-watchdog", + daemon=True, + ) + thread.start() + return stopped + + def serve_console( settings: Settings, *, @@ -4211,11 +4325,13 @@ def serve_console( display_host = "127.0.0.1" if host in {"0.0.0.0", "::"} else host output.write(f"GYXX Flow console: http://{display_host}:{server.server_port}\n") output.flush() + watchdog_stopped = _start_console_watchdog(server) try: server.serve_forever(poll_interval=0.5) except KeyboardInterrupt: pass finally: + watchdog_stopped.set() server.server_close() return 0 diff --git a/tests/test_acceptance_cookie_fail_closed.py b/tests/test_acceptance_cookie_fail_closed.py index e1c0ad4..ee697c3 100644 --- a/tests/test_acceptance_cookie_fail_closed.py +++ b/tests/test_acceptance_cookie_fail_closed.py @@ -10,9 +10,6 @@ from types import SimpleNamespace import pytest from gyxx_flow.adapters import COOKIE_SKIP_EXIT_CODE -from gyxx_flow.modules.content_marketing.data.tools import ( - friday_relogin_parallel as weekly_relogin, -) @pytest.fixture @@ -75,29 +72,6 @@ def _import_supply_script(module_name: str): sys.stderr = stderr -def test_weekly_relogin_skips_the_entire_qr_workflow( - acceptance_environment: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - monkeypatch.setattr( - weekly_relogin.subprocess, - "Popen", - _forbidden("acceptance mode must not launch relogin subprocesses"), - ) - monkeypatch.setattr( - weekly_relogin, - "_send_lark", - _forbidden("acceptance mode must not send QR screenshots"), - ) - - assert weekly_relogin.main() == COOKIE_SKIP_EXIT_CODE - assert "SKIPPED_COOKIE" in capsys.readouterr().out - assert "content.relogin.weekly" in acceptance_environment.read_text( - encoding="utf-8" - ) - - @pytest.mark.parametrize( "module_name", [ @@ -106,25 +80,17 @@ def test_weekly_relogin_skips_the_entire_qr_workflow( "batch_rescrape_bilibili", ], ) -def test_batch_rescrape_relogin_never_starts_a_qr_process( +def test_batch_rescrape_has_no_relogin_capability( module_name: str, - acceptance_environment: Path, monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], ) -> None: + """扫码重登能力已从批量补采工具中移除，登录统一由账号 vault 管理。""" + module = importlib.import_module( f"gyxx_flow.modules.content_marketing.data.tools.{module_name}" ) - monkeypatch.setattr( - module.subprocess, - "call", - _forbidden("acceptance mode must not invoke a relogin script"), - ) - - assert module.do_relogin() is False - assert "SKIPPED_COOKIE" in capsys.readouterr().out - evidence = acceptance_environment.read_text(encoding="utf-8") - assert f"content.{module_name}.relogin" in evidence + assert not hasattr(module, "do_relogin") + assert not hasattr(module, "RELOGIN_SCRIPT") @pytest.mark.parametrize( diff --git a/tests/test_catalog.py b/tests/test_catalog.py index 5efc7bc..19d1de3 100644 --- a/tests/test_catalog.py +++ b/tests/test_catalog.py @@ -18,8 +18,8 @@ def test_catalog_maps_39_project_scheduled_tasks() -> None: assert len(scheduled) == 39 assert len({item.source_task_name for item in scheduled}) == 39 assert Counter(item.module for item in scheduled) == { - "content_marketing": 10, - "product_commerce": 13, + "content_marketing": 9, + "product_commerce": 14, "shop_intelligence": 12, "supply_chain": 4, } @@ -80,7 +80,6 @@ def test_jd_video_upload_is_an_idempotent_scheduled_workflow() -> None: assert step.args == ( "--anchor-record-id", "recvrXTRWWFXeJ", - "--manual-login", "--execute", ) assert step.replay_policy == "idempotent" @@ -192,7 +191,7 @@ def test_every_scheduled_workflow_has_one_valid_asia_shanghai_schedule() -> None assert schedule.workflow_id == workflow.workflow_id assert schedule.kind in {"daily", "weekly", "monthly", "interval_days"} assert schedule.at.count(":") == 1 - assert sum(schedule.enabled for schedule in catalog.schedules) == 27 + assert sum(schedule.enabled for schedule in catalog.schedules) == 28 customer_service = catalog.schedule_for("shop.jd_customer_service.daily") assert customer_service.at == "03:00" assert customer_service.enabled is True @@ -265,8 +264,7 @@ def test_scheduled_workflows_use_python_entries_and_content_tasks_are_graphs() - "data/tools/sync_self_video_metrics.py" ) assert catalog.schedule_for(self_operated.workflow_id).effective_times == ( - "01:00", - "05:00", + "06:00", ) backfill = next( diff --git a/tests/test_cli_workflows.py b/tests/test_cli_workflows.py index c56adeb..3fbba5f 100644 --- a/tests/test_cli_workflows.py +++ b/tests/test_cli_workflows.py @@ -54,7 +54,7 @@ def test_default_cli_registry_exposes_all_project_scheduled_workflows( scheduled_ids = { entry.workflow_id for entry in registry.catalog.scheduled_workflows() } - assert len(scheduled_ids) == 35 + assert len(scheduled_ids) == 39 assert scheduled_ids == { schedule.workflow_id for schedule in registry.catalog.schedules } diff --git a/tests/test_console.py b/tests/test_console.py index f0bac72..1b138ed 100644 --- a/tests/test_console.py +++ b/tests/test_console.py @@ -2,6 +2,7 @@ from __future__ import annotations import http.client import json +import os import re import shutil import subprocess @@ -311,7 +312,7 @@ def test_notification_config_persists_app_scoped_multi_recipient_routes( ) initial = service.notification_config() - assert len(initial["people"]) == 16 + assert len(initial["people"]) == 18 assert initial["available_app_profiles"] == ["hermes-analyzer"] assert [app["profile"] for app in initial["apps"]] == ["hermes-analyzer"] wang = next(person for person in initial["people"] if person["id"] == "wang_yunlong") @@ -321,7 +322,12 @@ def test_notification_config_persists_app_scoped_multi_recipient_routes( assert wang["bindings"]["hermes-analyzer"]["open_id"] == ( "ou_7ad5fc8012e2f741afc5346e05ffd447" ) - assert all(route["configured"] is False for route in initial["routes"]) + stock_alert_route = next( + route + for route in initial["routes"] + if route["workflow_id"] == "product.online_stock_alert.daily" + ) + assert stock_alert_route["configured"] is True routes = [dict(route) for route in initial["routes"]] daily = next( @@ -1566,7 +1572,7 @@ def test_overview_exposes_only_verified_scheduled_runs_as_cancellable( "schema_version": 1, "updated_at": "2026-08-14T02:00:00+00:00", "jobs": { - "content.relogin.weekly": { + "content.notes_master.daily": { "status": "running", "pid": 4310, "business_date": "2026-08-14", @@ -1589,7 +1595,7 @@ def test_overview_exposes_only_verified_scheduled_runs_as_cancellable( workflow_id: str, _job: dict[str, Any], ) -> tuple[bool, Any, float | None]: - if workflow_id == "content.relogin.weekly": + if workflow_id == "content.notes_master.daily": return True, VerifiedProcess(), 1_800_000_000.0 return False, None, None @@ -1600,7 +1606,7 @@ def test_overview_exposes_only_verified_scheduled_runs_as_cancellable( ).overview() workflows = {item["id"]: item for item in overview["workflows"]} - verified = workflows["content.relogin.weekly"]["scheduler_job"] + verified = workflows["content.notes_master.daily"]["scheduler_job"] assert verified["status"] == "running" assert verified["cancellable"] is True assert "pid" not in verified @@ -1614,7 +1620,7 @@ def test_service_cancels_verified_scheduled_run_and_finalizes_journal( monkeypatch: pytest.MonkeyPatch, console_settings: Settings, ) -> None: - workflow_id = "content.relogin.weekly" + workflow_id = "content.notes_master.daily" _write_scheduler_state( console_settings, { @@ -1906,24 +1912,13 @@ def test_content_production_preflight_rejects_non_production_modes( relogin = next( item for item in catalog.workflows - if item.workflow_id == "content.relogin.weekly" + if item.workflow_id == "content.summary.weekly" ) with pytest.raises(ConsoleRequestError, match=message): console_module._validate_content_production_runtime(relogin, {flag: "1"}) -def test_content_relogin_preflight_does_not_require_postgres() -> None: - catalog = WorkflowCatalog.load(PROJECT_ROOT / "config") - relogin = next( - item - for item in catalog.workflows - if item.workflow_id == "content.relogin.weekly" - ) - - console_module._validate_content_production_runtime(relogin, {}) - - @pytest.mark.parametrize( "workflow_id", ["content.notes_master.daily", "content.metrics.daily"], @@ -3297,7 +3292,7 @@ def test_http_notification_config_requires_revision_and_supports_multi_recipient assert status == HTTPStatus.OK _assert_security_headers(response_headers) config = json.loads(body) - assert len(config["people"]) == 16 + assert len(config["people"]) == 18 assert { route["workflow_id"] for route in config["routes"] } == set(console_module.NOTIFICATION_CAPABILITIES) @@ -3616,3 +3611,82 @@ def test_douyin_negative_reviews_range_uses_single_run_with_env_dates( {"business_date": "2026-09-19"}, ) assert "business_date_end" not in single + + +def test_api_health_reports_process_and_scheduler_state( + console_settings: Settings, +) -> None: + with _running_server(console_settings, FakeConsoleLauncher()) as server: + status, _headers, content = _http_request(server, "GET", "/api/health") + assert status == HTTPStatus.OK + payload = json.loads(content) + assert payload["ok"] is True + assert payload["pid"] == os.getpid() + assert payload["uptime_seconds"] >= 0 + # 测试环境没有调度器实例锁，health 只报状态不做任何拉起。 + assert payload["scheduler_running"] is False + assert payload["scheduler_pid"] is None + + +def test_api_health_requires_no_token(console_settings: Settings) -> None: + # 自检探针端点供进程内看门狗与保活脚本使用，必须绕过访问令牌。 + with _running_server( + console_settings, FakeConsoleLauncher(), token="x" * 24 + ) as server: + status, _headers, _content = _http_request(server, "GET", "/api/health") + assert status == HTTPStatus.OK + + +def test_probe_console_health_against_real_server( + console_settings: Settings, +) -> None: + with _running_server(console_settings, FakeConsoleLauncher()) as server: + assert console_module._probe_console_health(server.server_port, 2.0) is True + assert console_module._probe_console_health(1, 0.2) is False + + +def test_console_watchdog_exits_after_consecutive_probe_failures() -> None: + stopped = threading.Event() + exits: list[int] = [] + probes: list[int] = [] + + def always_failing(port: int, _timeout: float) -> bool: + probes.append(port) + return False + + console_module._console_watchdog( + 8765, + interval_seconds=0.01, + failure_limit=3, + probe_timeout_seconds=0.01, + stopped=stopped, + probe=always_failing, + exit_process=exits.append, + sleep=lambda _seconds: None, + ) + + assert exits == [console_module.CONSOLE_WATCHDOG_EXIT_CODE] + assert probes == [8765, 8765, 8765] + + +def test_console_watchdog_resets_failure_count_after_successful_probe() -> None: + stopped = threading.Event() + exits: list[int] = [] + results = iter([False, False, True, False, False, False]) + + def flaky_probe(_port: int, _timeout: float) -> bool: + return next(results) + + console_module._console_watchdog( + 8765, + interval_seconds=0.01, + failure_limit=3, + probe_timeout_seconds=0.01, + stopped=stopped, + probe=flaky_probe, + exit_process=exits.append, + sleep=lambda _seconds: None, + ) + + # 成功一次清零计数：2 次失败 + 1 次成功 + 3 次失败后才退出。 + assert exits == [console_module.CONSOLE_WATCHDOG_EXIT_CODE] diff --git a/tests/test_runtime_integrations.py b/tests/test_runtime_integrations.py index d6b4f10..c7afdfd 100644 --- a/tests/test_runtime_integrations.py +++ b/tests/test_runtime_integrations.py @@ -103,9 +103,9 @@ def test_project_registry_covers_every_script_with_unique_stable_port() -> None: ) assert len(scripts.command_ids) == 60 - assert len(first.command_ids) == 163 + assert len(first.command_ids) == 157 bindings = [first.binding_for(command_id) for command_id in first.command_ids] - assert len({binding.cdp_port for binding in bindings}) == 163 + assert len({binding.cdp_port for binding in bindings}) == 157 assert all(22000 <= binding.cdp_port <= 22999 for binding in bindings) assert all(binding.cdp_url.startswith("http://127.0.0.1:") for binding in bindings) assert bindings == [second.binding_for(item.script_id) for item in bindings] @@ -450,7 +450,7 @@ def test_project_accounts_cover_all_configured_platform_login_identities() -> No assert all( catalog.bindings_for_account(account.account_id) for account in catalog.accounts - if account.account_id != "tmall-ozko" + if account.account_id not in {"tmall-ozko", "xingtu"} ) assert catalog.bindings_for_account("tmall-ozko") == () assert catalog.account_for("jd-shop").credential_env_names == ( diff --git a/tests/test_scheduler_cli.py b/tests/test_scheduler_cli.py index cda6baf..aea8d00 100644 --- a/tests/test_scheduler_cli.py +++ b/tests/test_scheduler_cli.py @@ -54,7 +54,7 @@ def test_python_scheduler_status_is_project_owned_and_cross_platform(tmp_path: P assert exit_code == 0 payload = json.loads(output.getvalue()) assert payload["timezone"] == "Asia/Shanghai" - assert payload["scheduled_count"] == 35 + assert payload["scheduled_count"] == 39 assert payload["state"]["schema_version"] == 1 diff --git a/tests/test_scheduler_service.py b/tests/test_scheduler_service.py index f05f169..0008165 100644 --- a/tests/test_scheduler_service.py +++ b/tests/test_scheduler_service.py @@ -206,16 +206,17 @@ def test_misfire_older_than_grace_is_not_launched(tmp_path: Path) -> None: def test_configured_schedules_disable_cookie_refresh_job() -> None: catalog = WorkflowCatalog.load(PROJECT_ROOT / "config") - assert len(catalog.schedules) == 35 + assert len(catalog.schedules) == 39 assert [ schedule.workflow_id for schedule in catalog.schedules if not schedule.enabled ] == [ "content.metrics.backfill", "content.marketing_report.daily", - "content.relogin.weekly", "content.summary.weekly", "content.summary.monthly", "content.creator_report.monthly", + "shop.tmall_negative_reviews.daily", + "shop.tmall_customer_service.daily", "product.alert.daily", "product.style_analysis.interval", "supply.purchase_confirmation.daily", diff --git a/tests/test_server_scheduler_deployment.py b/tests/test_server_scheduler_deployment.py index b95e09a..9c6953f 100644 --- a/tests/test_server_scheduler_deployment.py +++ b/tests/test_server_scheduler_deployment.py @@ -47,8 +47,8 @@ def test_deployment_documents_one_scheduler_and_external_data_root() -> None: assert "GYXX_DATA_ROOT" in guide assert "/var/lib/gyxx-flow" in guide assert "Windows Task Scheduler" in guide - assert len(payload["schedules"]) == 35 - assert sum(item.get("enabled", True) for item in payload["schedules"]) == 25 + assert len(payload["schedules"]) == 39 + assert sum(item.get("enabled", True) for item in payload["schedules"]) == 28 def test_three_canonical_documents_cover_architecture_deployment_and_operations() -> None: