chore(config): 工作流配置、清单与文档同步

workflows/schedules/commands/notification-routing 与 source-manifests 四模块
清单同步登录统一化与新工作流(product.hk_anpei_public_inventory.daily 等);
docs 运维手册、控制台与耗时分布文档、CI 关键脚本清单、AGENTS 指南更新;
测试与模块注册对齐。
This commit is contained in:
wangyunlong committed 2026-10-09 13:45:30 +08:00
1 parent 90282ae766
commit dc8be88e60
27 files changed
+1094 -224

No files matched your search

@@ -10,7 +10,6 @@ from gyxx_flow.modules.content_marketing.data.tools import (
analyze_comments,
analyze_note,
daily_report_card,
friday_relogin_parallel,
)
@@ -138,15 +137,3 @@ def test_note_report_sends_only_to_wang_yunlong(
assert calls[0]["user_id"] == ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID
assert calls[0]["profile"] == "hermes-analyzer"
def test_relogin_notifications_resolve_only_to_wang_yunlong(
monkeypatch,
tmp_path: Path,
) -> None:
_enable_acceptance(monkeypatch, tmp_path)
assert (
friday_relogin_parallel.recipient_for_platform("pgy", "ou_other")
== ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID
)
@@ -416,11 +416,9 @@ def test_acceptance_invalid_session_skips_before_qr_login(
)
monkeypatch.setenv("GYXX_ACCEPTANCE_EVIDENCE_FILE", str(evidence_file))
monkeypatch.setattr(module, "is_logged_in", lambda _page: False)
monkeypatch.setattr(
module,
"wait_for_scan_login",
lambda *_args, **_kwargs: pytest.fail("QR login must remain disabled"),
)
# 扫码登录函数已整体移除;采集器不得重新引入交互登录。
assert not hasattr(module, "wait_for_scan_login")
assert not hasattr(module, "open_scan_login")
class Page:
def __init__(self):
+1 -1
View File
@@ -56,7 +56,6 @@ def test_catalog_composition_registers_all_workflows() -> None:
"content.metrics.self_operated.daily",
"content.metrics.backfill",
"content.marketing_report.daily",
"content.relogin.weekly",
"content.creator_report.monthly",
"content.summary.monthly",
"content.notes_master.daily",
@@ -65,6 +64,7 @@ def test_catalog_composition_registers_all_workflows() -> None:
"product.persona.daily",
"product.daily",
"product.online_stock_alert.daily",
"product.hk_anpei_public_inventory.daily",
"product.alert.daily",
"product.ecommerce_costs.daily",
"product.style_analysis.interval",
+1 -1
View File
@@ -23,7 +23,7 @@ def test_content_module_registers_only_scheduled_entries() -> None:
definitions = module.workflow_definitions()
assert tuple(item.workflow_id for item in definitions) == CONTENT_WORKFLOW_IDS
assert len(CONTENT_SCHEDULED_WORKFLOW_IDS) == 10
assert len(CONTENT_SCHEDULED_WORKFLOW_IDS) == 9
def test_content_workflows_use_project_owned_module_commands() -> None:
+1 -2
View File
@@ -44,7 +44,6 @@ REQUIRED_SOURCES = {
"chanmama_scraper.py",
"daily_creator_exposure_scope.py",
"daily_marketing_report.py",
"data/tools/friday_relogin_parallel.py",
"data/tools/schema_gyxx_super_data.sql",
"data/tools/migrations/005_style_product_profile.sql",
"data/config/db.env.example",
@@ -71,7 +70,7 @@ def test_content_source_snapshot_manifest_is_complete_and_well_formed() -> None:
assert manifest["module"] == "content_marketing"
files = manifest["files"]
assert isinstance(files, list)
assert len(files) == 66
assert len(files) == 59
sources = [entry["source_relative_path"] for entry in files]
targets = [entry["target_relative_path"] for entry in files]
+771
View File
@@ -0,0 +1,771 @@
"""Cookie gate: per-account login-state machine injected before workflows."""
from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace
from gyxx_flow.adapters.browser import BrowserCookieStore
from gyxx_flow.adapters.integration import AccountBinding
from gyxx_flow.catalog import WorkflowCatalog, WorkflowEntry
from gyxx_flow.cookie_gate import (
GATE_MODE_BLOCKING,
GATE_MODE_DEFER,
GATE_MODE_RECOVER,
CookieGate,
CookieGateRecoveryManager,
apply_cookie_gates,
gate_enabled,
)
from gyxx_flow.core.config import Settings
from gyxx_flow.core.context import RunContext
from gyxx_flow.core.layout import DataLayout
from gyxx_flow.core.locks import LockManager
from gyxx_flow.core.records import RunJournal
from gyxx_flow.workflow.engine import WorkflowEngine
from gyxx_flow.workflow.model import StepDefinition, WorkflowDefinition
from gyxx_flow.workflow.steps import StepExecution
_NOW = datetime(2026, 10, 8, 12, 0, 0, tzinfo=timezone.utc)
def _settings(tmp_path: Path) -> Settings:
return Settings(project_root=tmp_path, data_root=tmp_path / "var")
def _account(tmp_path: Path, account_id: str, **overrides: object) -> AccountBinding:
vault = tmp_path / "state" / "accounts" / account_id
values: dict[str, object] = {
"account_id": account_id,
"cdp_port": 22000,
"cdp_url": "http://127.0.0.1:22000",
"profile_dir": vault / "profile",
"cookie_file": vault / "cookies.json",
"storage_state_file": vault / "storage_state.json",
"required_cookie_domains": ("example.com",),
"required_cookie_names": ("sessionid",),
"login_identity_markers": ("示例店铺",),
"keepalive_enabled": True,
"keepalive_url": "https://example.com/home",
"keepalive_interval_seconds": 900,
}
values.update(overrides)
return AccountBinding(**values) # type: ignore[arg-type]
def _fake_catalog(*accounts: AccountBinding, bindings: dict[str, str] | None = None):
by_id = {account.account_id: account for account in accounts}
binding_accounts = bindings or {}
account_tuple = tuple(accounts)
class _Catalog:
accounts = account_tuple
def account_for(self, account_id: str) -> AccountBinding:
return by_id[account_id]
def binding_for(self, binding_id: str):
return SimpleNamespace(account=binding_accounts.get(binding_id, ""))
return _Catalog()
def _write_valid_vault(account: AccountBinding) -> None:
store = BrowserCookieStore(account.cookie_file, account.storage_state_file)
store.save_cookies(
[
{
"name": "sessionid",
"value": "live-session",
"domain": ".example.com",
"path": "/",
"expires": _NOW.timestamp() + 86_400,
}
]
)
def _write_keepalive_status(
account: AccountBinding,
payload: dict[str, object] | None,
) -> None:
path = account.cookie_file.with_name("keepalive_status.json")
if payload is None:
path.unlink(missing_ok=True)
return
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload), encoding="utf-8")
def _gate(
tmp_path: Path,
*accounts: AccountBinding,
probe: dict[str, object] | None = None,
environment: dict[str, str] | None = None,
sent: list[dict[str, object]] | None = None,
bindings: dict[str, str] | None = None,
) -> CookieGate:
probes = probe or {"status": "expired"}
keepalive = SimpleNamespace(
refresh_account=lambda account_id: dict(probes, account_id=account_id)
)
sender_calls = sent if sent is not None else []
def _sender(**kwargs: object) -> dict[str, object]:
sender_calls.append(dict(kwargs))
return {"message_id": "om_fake"}
catalog = _fake_catalog(*accounts, bindings=bindings)
gate = CookieGate(
_settings(tmp_path),
catalog=catalog,
keepalive=keepalive, # type: ignore[arg-type]
environment=environment or {},
now=lambda: _NOW,
sender=_sender,
)
gate._workflow_catalog = WorkflowCatalog(timezone="UTC", workflows=(), schedules=())
return gate
def _context(**overrides: object) -> RunContext:
values: dict[str, object] = {
"workflow_id": "shop.demo.daily",
"business_date": "2026-10-08",
"shadow": False,
"now": _NOW,
"random_suffix": "aaaa",
}
values.update(overrides)
return RunContext.create(**values) # type: ignore[arg-type]
def _noop_outcome(*, context, timeout_seconds, dry_run) -> StepExecution: # noqa: ANN001
return StepExecution(exit_code=0)
def _deny_outcome(*, context, timeout_seconds, dry_run) -> StepExecution: # noqa: ANN001
raise AssertionError("resolution must not run actions")
class _PassAction:
execute = staticmethod(_noop_outcome)
class _BoundAction:
"""Duck-typed DeferredModuleCommandStep for account resolution tests."""
def __init__(self, command_entry: str = "collect.py") -> None:
self.entry = SimpleNamespace(module="shop_intelligence", entry="fallback.py")
self.command_entry = command_entry
self.execute = _deny_outcome
class _RecordingAction:
def __init__(self, name: str, sink: list[str], command_entry: str | None = None) -> None:
def run(*, context, timeout_seconds, dry_run) -> StepExecution: # noqa: ANN001
sink.append(name)
return StepExecution(exit_code=0)
self.execute = run
if command_entry is not None:
self.entry = SimpleNamespace(
module="shop_intelligence", entry="fallback.py"
)
self.command_entry = command_entry
# ----------------------------------------------------------------------
# Mode classification
# ----------------------------------------------------------------------
def test_gate_mode_classifies_accounts(tmp_path: Path) -> None:
gate = _gate(
tmp_path,
_account(tmp_path, "qr-only"),
_account(tmp_path, "cred-scripts", credential_env_names=("A_USER", "A_PASS")),
_account(
tmp_path,
"cred-vault",
login_mode="C",
login_command_id="product_commerce:erp_login.py",
),
_account(tmp_path, "cred-mode-b", login_mode="B"),
)
assert gate.gate_mode("qr-only") == GATE_MODE_BLOCKING
assert gate.gate_mode("cred-scripts") == GATE_MODE_DEFER
assert gate.gate_mode("cred-vault") == GATE_MODE_RECOVER
assert gate.gate_mode("cred-mode-b") == GATE_MODE_DEFER
# ----------------------------------------------------------------------
# Workflow account resolution
# ----------------------------------------------------------------------
def _bound_workflow(workflow_id: str = "shop.demo.daily") -> WorkflowDefinition:
return WorkflowDefinition(
workflow_id,
(
StepDefinition("collect", _BoundAction("collect.py")),
StepDefinition("sync", _BoundAction("sync.py"), depends_on=("collect",)),
),
)
def test_collect_accounts_maps_bindings_and_skips_defer(tmp_path: Path) -> None:
gate = _gate(
tmp_path,
_account(tmp_path, "douyin-shop"),
_account(tmp_path, "jd-shop", credential_env_names=("JD_SHOP", "JD_PASSWORD")),
bindings={
"shop_intelligence:collect.py": "douyin-shop",
"shop_intelligence:other.py": "jd-shop",
},
)
mapping = gate.collect_workflow_accounts(_bound_workflow())
# sync.py 无账号绑定,jd-shop 属 defer 模式:两者都不进映射
assert mapping == {"douyin-shop": {"collect"}}
def test_collect_accounts_includes_declared_required_accounts(tmp_path: Path) -> None:
gate = _gate(tmp_path, _account(tmp_path, "douyin-shop"))
gate._workflow_catalog = WorkflowCatalog(
timezone="UTC",
workflows=(
WorkflowEntry(
workflow_id="shop.demo.daily",
module="shop_intelligence",
trigger="manual",
entry="collect.py",
required_accounts=("douyin-shop",),
),
),
schedules=(),
)
mapping = gate.collect_workflow_accounts(_bound_workflow())
assert mapping == {"douyin-shop": {"collect", "sync"}}
def test_apply_cookie_gates_rewrites_dependencies(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
gate = _gate(
tmp_path,
account,
bindings={"shop_intelligence:collect.py": "douyin-shop"},
)
gated = apply_cookie_gates(
_bound_workflow(), settings=_settings(tmp_path), gate=gate
)
ids = [step.step_id for step in gated.steps]
assert "cookie-gate.douyin-shop" in ids
collect = next(step for step in gated.steps if step.step_id == "collect")
sync = next(step for step in gated.steps if step.step_id == "sync")
assert collect.depends_on == ("cookie-gate.douyin-shop",)
assert sync.depends_on == ("collect",)
assert gated.ordered_steps() # patched DAG stays valid
def test_apply_cookie_gates_returns_workflow_without_accounts(tmp_path: Path) -> None:
gate = _gate(tmp_path, _account(tmp_path, "douyin-shop"))
workflow = _bound_workflow()
assert (
apply_cookie_gates(workflow, settings=_settings(tmp_path), gate=gate) is workflow
)
def test_gate_enabled_env_kill_switch() -> None:
assert gate_enabled({})
assert not gate_enabled({"GYXX_COOKIE_GATE": "0"})
assert not gate_enabled({"GYXX_COOKIE_GATE": "off"})
assert gate_enabled({"GYXX_COOKIE_GATE": "1"})
# ----------------------------------------------------------------------
# State machine: check()
# ----------------------------------------------------------------------
def test_check_fast_path_accepts_fresh_keepalive_without_probe(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
gate = _gate(tmp_path, account, probe={"status": "expired"})
_write_valid_vault(account)
_write_keepalive_status(
account,
{
"status": "success",
"last_success_at": (_NOW - timedelta(minutes=10)).isoformat(),
},
)
verdict = gate.check("douyin-shop", context=_context())
assert not verdict.blocked
assert "keepalive" in verdict.message
state = json.loads(
(account.cookie_file.parent / "cookie_state.json").read_text(encoding="utf-8")
)
assert state["state"] == "healthy"
def test_check_stale_status_triggers_probe(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
gate = _gate(tmp_path, account, probe={"status": "success"})
_write_valid_vault(account)
_write_keepalive_status(
account,
{
"status": "success",
"last_success_at": (_NOW - timedelta(hours=6)).isoformat(),
},
)
verdict = gate.check("douyin-shop", context=_context())
assert not verdict.blocked
assert verdict.message.endswith("探活通过)")
def test_check_expired_blocks_and_sends_feishu_alert(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
sent: list[dict[str, object]] = []
gate = _gate(
tmp_path,
account,
probe={"status": "expired"},
environment={"GYXX_LOGIN_ALERT_OPEN_IDS": "ou_alert_target"},
sent=sent,
)
_write_keepalive_status(account, None)
verdict = gate.check("douyin-shop", context=_context())
assert verdict.blocked
assert verdict.notified
assert len(sent) == 1
text = str(sent[0]["text"])
assert "uv run gyxx accounts login douyin-shop" in text
assert "shop.demo.daily" in text
assert sent[0]["profile"] == "hermes-analyzer"
state = json.loads(
(account.cookie_file.parent / "cookie_state.json").read_text(encoding="utf-8")
)
assert state["state"] == "awaiting_login"
assert state["last_notified_at"] == _NOW.isoformat(timespec="seconds")
def test_check_notification_is_throttled_per_interval(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
sent: list[dict[str, object]] = []
gate = _gate(
tmp_path,
account,
probe={"status": "expired"},
environment={"GYXX_LOGIN_ALERT_OPEN_IDS": "ou_alert_target"},
sent=sent,
)
_write_keepalive_status(account, None)
first = gate.check("douyin-shop", context=_context())
second = gate.check("douyin-shop", context=_context())
assert first.notified and second.blocked
assert not second.notified
assert len(sent) == 1
def test_check_shadow_run_never_notifies(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
sent: list[dict[str, object]] = []
gate = _gate(
tmp_path,
account,
probe={"status": "expired"},
environment={"GYXX_LOGIN_ALERT_OPEN_IDS": "ou_alert_target"},
sent=sent,
)
verdict = gate.check("douyin-shop", context=_context(shadow=True))
assert verdict.blocked
assert not verdict.notified
assert sent == []
def test_check_mode_c_recovery_success_and_failure_paths(tmp_path: Path) -> None:
account = _account(
tmp_path,
"erp-jushuitan",
login_mode="C",
login_command_id="product_commerce:erp_login.py",
)
recovered = _gate(
tmp_path, account, probe={"status": "success", "login_recovered": True}
)
_write_keepalive_status(account, None)
verdict = recovered.check("erp-jushuitan", context=_context())
assert not verdict.blocked
assert verdict.recovered
assert "自动账密重登成功" in verdict.message
sent: list[dict[str, object]] = []
failed = _gate(
tmp_path,
account,
probe={"status": "not_logged_in"},
environment={"GYXX_LOGIN_ALERT_OPEN_IDS": "ou_alert_target"},
sent=sent,
)
_write_keepalive_status(
failed.catalog.account_for("erp-jushuitan"),
{
"status": "not_logged_in",
"login_attempted": True,
"login_error_type": "TimeoutExpired",
},
)
verdict = failed.check("erp-jushuitan", context=_context())
assert verdict.blocked
assert verdict.notified
assert "自动账密重登已尝试但未成功" in str(sent[0]["text"])
def test_check_busy_falls_back_to_static_evidence(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
gate = _gate(tmp_path, account, probe={"status": "busy"})
_write_valid_vault(account)
_write_keepalive_status(account, None)
verdict = gate.check("douyin-shop", context=_context())
assert not verdict.blocked
assert "探活锁忙" in verdict.message
def test_check_probe_error_with_static_valid_vault_passes(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
gate = _gate(tmp_path, account, probe={"status": "timeout"})
_write_valid_vault(account)
_write_keepalive_status(account, None)
verdict = gate.check("douyin-shop", context=_context())
assert not verdict.blocked
assert "探活异常" in verdict.message
def test_acceptance_mode_is_static_only_and_never_alerts(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
sent: list[dict[str, object]] = []
gate = _gate(
tmp_path,
account,
probe={"status": "expired"},
environment={
"GYXX_WORKFLOW_ACCEPTANCE": "1",
"GYXX_LOGIN_ALERT_OPEN_IDS": "ou_alert_target",
},
sent=sent,
)
_write_keepalive_status(account, None)
verdict = gate.check("douyin-shop", context=_context())
assert verdict.blocked
assert sent == []
assert "验收模式" in verdict.message
def test_alert_recipient_falls_back_to_constant(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
gate = _gate(tmp_path, account)
recipients = gate._alert_recipients()
assert recipients == [
("ou_7ad5fc8012e2f741afc5346e05ffd447", "hermes-analyzer")
]
# ----------------------------------------------------------------------
# Engine integration
# ----------------------------------------------------------------------
def _engine(tmp_path: Path) -> WorkflowEngine:
return WorkflowEngine(LockManager(tmp_path / "state" / "locks"))
def test_engine_blocks_dependents_when_gate_fails(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
sent: list[dict[str, object]] = []
gate = _gate(
tmp_path,
account,
probe={"status": "expired"},
environment={"GYXX_LOGIN_ALERT_OPEN_IDS": "ou_alert_target"},
sent=sent,
bindings={"shop_intelligence:collect.py": "douyin-shop"},
)
executed: list[str] = []
workflow = WorkflowDefinition(
"shop.demo.daily",
(
StepDefinition(
"collect", _RecordingAction("collect", executed, "collect.py")
),
StepDefinition(
"sync", _RecordingAction("sync", executed), depends_on=("collect",)
),
),
)
gated = apply_cookie_gates(workflow, settings=_settings(tmp_path), gate=gate)
context = RunContext.create(
"shop.demo.daily", "2026-10-08", now=_NOW, random_suffix="dddd"
)
journal = RunJournal.create(DataLayout(_settings(tmp_path).data_root), context)
engine = _engine(tmp_path)
run_workflow = engine.execute
result = run_workflow(gated, context=context, journal=journal)
assert result.status == "failed"
assert result.exit_code == 1
assert result.steps["cookie-gate.douyin-shop"].status == "failed"
assert result.steps["collect"].reason == "dependency-failed"
assert result.steps["sync"].reason == "dependency-failed"
assert executed == []
assert len(sent) == 1
def test_engine_runs_collect_steps_when_gate_passes(tmp_path: Path) -> None:
account = _account(tmp_path, "douyin-shop")
gate = _gate(
tmp_path,
account,
probe={"status": "success"},
bindings={"shop_intelligence:collect.py": "douyin-shop"},
)
_write_valid_vault(account)
_write_keepalive_status(account, None)
executed: list[str] = []
workflow = WorkflowDefinition(
"shop.demo.daily",
(
StepDefinition(
"collect", _RecordingAction("collect", executed, "collect.py")
),
),
)
gated = apply_cookie_gates(workflow, settings=_settings(tmp_path), gate=gate)
context = RunContext.create(
"shop.demo.daily", "2026-10-08", now=_NOW, random_suffix="eeee"
)
journal = RunJournal.create(DataLayout(_settings(tmp_path).data_root), context)
engine = _engine(tmp_path)
run_workflow = engine.execute
result = run_workflow(gated, context=context, journal=journal)
assert result.status == "success"
assert result.steps["cookie-gate.douyin-shop"].status == "success"
assert result.steps["collect"].status == "success"
# ----------------------------------------------------------------------
# Recovery: automatic rerun after the login recovers
# ----------------------------------------------------------------------
class _FakeLauncher:
def __init__(self, fail: bool = False) -> None:
self.launched: list[tuple[str, str]] = []
self.closed: list[int] = []
self.fail = fail
def launch(self, workflow_id, slot, business_date): # noqa: ANN001
if self.fail:
raise RuntimeError("spawn denied")
process = SimpleNamespace(pid=9000 + len(self.launched), poll=lambda: None)
self.launched.append((workflow_id, business_date.isoformat()))
return process
def close(self, process) -> None: # noqa: ANN001
self.closed.append(process.pid)
def _state_file(account: AccountBinding) -> Path:
return account.cookie_file.parent / "cookie_state.json"
def _read_state(account: AccountBinding) -> dict:
return json.loads(_state_file(account).read_text(encoding="utf-8"))
def _recovery_workflow_catalog() -> WorkflowCatalog:
return WorkflowCatalog(
timezone="UTC",
workflows=(
WorkflowEntry(
workflow_id="shop.demo.daily",
module="shop_intelligence",
trigger="scheduled",
entry="collect.py",
),
),
schedules=(),
)
def _blocking_gate(
tmp_path: Path,
account: AccountBinding,
sent: list[dict[str, object]],
) -> CookieGate:
gate = _gate(
tmp_path,
account,
probe={"status": "expired"},
environment={"GYXX_LOGIN_ALERT_OPEN_IDS": "ou_alert_target"},
sent=sent,
)
return gate
def _healthy_gate(
tmp_path: Path,
account: AccountBinding,
sent: list[dict[str, object]],
) -> CookieGate:
gate = _gate(
tmp_path,
account,
probe={"status": "success"},
environment={"GYXX_LOGIN_ALERT_OPEN_IDS": "ou_alert_target"},
sent=sent,
)
gate._workflow_catalog = _recovery_workflow_catalog()
return gate
def _simulate_manual_login(account: AccountBinding) -> None:
_write_valid_vault(account)
_write_keepalive_status(
account,
{
"status": "success",
"last_success_at": (_NOW - timedelta(seconds=60)).isoformat(),
},
)
def test_expire_records_pending_reruns(tmp_path: Path) -> None:
sent: list[dict[str, object]] = []
account = _account(tmp_path, "douyin-shop")
gate = _blocking_gate(tmp_path, account, sent)
gate.check("douyin-shop", context=_context())
gate.check("douyin-shop", context=_context()) # 同工作流同日期去重
gate.check("douyin-shop", context=_context(business_date="2026-10-09"))
state = _read_state(account)
assert [
(item["workflow_id"], item["business_date"])
for item in state["pending_reruns"]
] == [
("shop.demo.daily", "2026-10-08"),
("shop.demo.daily", "2026-10-09"),
]
# shadow 运行的阻断不进入待重跑清单
gate.check(
"douyin-shop", context=_context(business_date="2026-10-10", shadow=True)
)
state = _read_state(account)
assert all(
item["business_date"] != "2026-10-10" for item in state["pending_reruns"]
)
def test_recovery_reruns_blocked_workflows_after_login(tmp_path: Path) -> None:
sent: list[dict[str, object]] = []
account = _account(tmp_path, "douyin-shop")
blocked = _blocking_gate(tmp_path, account, sent)
blocked.check("douyin-shop", context=_context())
_simulate_manual_login(account)
healthy = _healthy_gate(tmp_path, account, sent)
launcher = _FakeLauncher()
outcomes = CookieGateRecoveryManager(healthy, launcher=launcher).tick()
assert launcher.launched == [("shop.demo.daily", "2026-10-08")]
assert launcher.closed # 父进程日志句柄立即关闭
assert outcomes and outcomes[0]["launched"]
state = _read_state(account)
assert state["pending_reruns"] == []
assert state["state"] == "healthy"
assert state["last_rerun_launch_at"]
assert state["rerun_history"][0]["workflow_id"] == "shop.demo.daily"
assert any("登录态已恢复" in str(item["text"]) for item in sent)
def test_recovery_skips_while_login_still_expired(tmp_path: Path) -> None:
sent: list[dict[str, object]] = []
account = _account(tmp_path, "douyin-shop")
blocked = _blocking_gate(tmp_path, account, sent)
blocked.check("douyin-shop", context=_context())
_write_keepalive_status(account, None)
still_expired = _gate(
tmp_path,
account,
probe={"status": "expired"},
environment={"GYXX_LOGIN_ALERT_OPEN_IDS": "ou_alert_target"},
sent=sent,
)
still_expired._workflow_catalog = _recovery_workflow_catalog()
launcher = _FakeLauncher()
outcomes = CookieGateRecoveryManager(still_expired, launcher=launcher).tick()
assert launcher.launched == []
assert outcomes == []
state = _read_state(account)
assert len(state["pending_reruns"]) == 1
assert state["last_recovery_probe_at"]
def test_recovery_drops_adhoc_script_ids(tmp_path: Path) -> None:
sent: list[dict[str, object]] = []
account = _account(tmp_path, "douyin-shop")
blocked = _blocking_gate(tmp_path, account, sent)
blocked.check(
"douyin-shop", context=_context(workflow_id="script.shop.abcdef123456")
)
assert (
_read_state(account)["pending_reruns"][0]["workflow_id"]
== "script.shop.abcdef123456"
)
_simulate_manual_login(account)
healthy = _healthy_gate(tmp_path, account, sent)
launcher = _FakeLauncher()
CookieGateRecoveryManager(healthy, launcher=launcher).tick()
assert launcher.launched == []
assert _read_state(account)["pending_reruns"] == []
def test_recovery_launch_failure_keeps_pending_for_retry(tmp_path: Path) -> None:
sent: list[dict[str, object]] = []
account = _account(tmp_path, "douyin-shop")
blocked = _blocking_gate(tmp_path, account, sent)
blocked.check("douyin-shop", context=_context())
_simulate_manual_login(account)
healthy = _healthy_gate(tmp_path, account, sent)
launcher = _FakeLauncher(fail=True)
CookieGateRecoveryManager(healthy, launcher=launcher).tick()
state = _read_state(account)
assert len(state["pending_reruns"]) == 1
assert "error" in state["pending_reruns"][0]
assert state["state"] == "awaiting_login"
assert not any("已恢复" in str(item["text"]) for item in sent)
def test_recovery_respects_launch_cooldown(tmp_path: Path) -> None:
sent: list[dict[str, object]] = []
account = _account(tmp_path, "douyin-shop")
blocked = _blocking_gate(tmp_path, account, sent)
blocked.check("douyin-shop", context=_context())
_simulate_manual_login(account)
state = _read_state(account)
state["last_rerun_launch_at"] = _NOW.isoformat(timespec="seconds")
_state_file(account).write_text(json.dumps(state), encoding="utf-8")
healthy = _healthy_gate(tmp_path, account, sent)
launcher = _FakeLauncher()
CookieGateRecoveryManager(healthy, launcher=launcher).tick()
assert launcher.launched == []
+16 -5
View File
@@ -67,15 +67,21 @@ def test_baseline_contains_all_provided_people_and_routable_capabilities(
) -> None:
payload, revision = NotificationRoutingStore(routing_settings).snapshot()
assert len(payload["people"]) == 16
assert len(payload["people"]) == 18
assert payload["app_profile"] == "hermes-analyzer"
assert (
payload["app_profiles"]["hermes-analyzer"]["app_id"]
== ANALYZER_NOTIFICATION_APP_ID
)
assert all(person["mobile"] is None for person in payload["people"].values())
mobiles = {
person_id: person["mobile"]
for person_id, person in payload["people"].items()
}
assert mobiles["wang_yunlong"] == "+8613663785145"
assert mobiles["stock_alert_user_19832771651"] == "+8619832771651"
assert set(payload["capabilities"]) == set(NOTIFICATION_CAPABILITIES)
assert payload["routes"] == {}
assert set(payload["routes"]) == {"product.online_stock_alert.daily"}
assert payload["routes"]["product.online_stock_alert.daily"]["enabled"] is True
assert len(revision) == 64
wang = payload["people"]["wang_yunlong"]
@@ -85,10 +91,15 @@ def test_baseline_contains_all_provided_people_and_routable_capabilities(
assert wang["bindings"]["hermes-analyzer"]["open_id"] == (
"ou_7ad5fc8012e2f741afc5346e05ffd447"
)
_mobile_lookup_people = {
"wang_yunlong",
"online_stock_alert_recipient",
"stock_alert_user_19832771651",
}
provided = {
person["name"]: binding["open_id"]
for person_id, person in payload["people"].items()
if person_id != "wang_yunlong"
if person_id not in _mobile_lookup_people
for profile, binding in person["bindings"].items()
if profile == "hermes-analyzer"
}
@@ -112,7 +123,7 @@ def test_baseline_contains_all_provided_people_and_routable_capabilities(
assert all(
binding["verified"] is False and binding["source"] == "provided"
for person_id, person in payload["people"].items()
if person_id != "wang_yunlong"
if person_id not in _mobile_lookup_people
for profile, binding in person["bindings"].items()
if profile == "hermes-analyzer"
)
+6 -3
View File
@@ -24,7 +24,7 @@ def test_product_module_registers_scheduled_and_manual_entries() -> None:
WorkflowCatalog.load(PROJECT_ROOT / "config")
)
assert tuple(item.workflow_id for item in module.workflow_definitions()) == PRODUCT_WORKFLOW_IDS
assert len(PRODUCT_SCHEDULED_WORKFLOW_IDS) == 12
assert len(PRODUCT_SCHEDULED_WORKFLOW_IDS) == 14
assert PRODUCT_MANUAL_WORKFLOW_IDS == ()
@@ -78,7 +78,10 @@ def test_product_workflows_use_project_owned_module_commands() -> None:
assert definition.steps
for step in definition.steps:
assert isinstance(step.action, DeferredModuleCommandStep)
assert step.timeout_seconds == PRODUCT_TIMEOUT_SECONDS == 6 * 60 * 60
assert step.timeout_seconds == PRODUCT_TIMEOUT_SECONDS == 6 * 60 * 60 or (
definition.workflow_id == "product.hk_anpei_public_inventory.daily"
and step.timeout_seconds == 900.0
)
assert step.max_attempts == 1
if definition.workflow_id == "product.ecommerce_costs.daily":
assert step.resources == (
@@ -93,6 +96,7 @@ def test_product_workflows_use_project_owned_module_commands() -> None:
"product.video_upload",
"product.jd_video_upload",
"product.tmall_baibu_apply",
"product.hk_anpei_public_inventory.daily",
}:
assert step.resources == (
f"{PRODUCT_RESOURCE}:{definition.workflow_id}",
@@ -219,7 +223,6 @@ def test_jd_video_upload_workflow_allows_idempotent_rescans() -> None:
assert step.action.command_args == (
"--anchor-record-id",
"recvrXTRWWFXeJ",
"--manual-login",
"--execute",
)
+3 -3
View File
@@ -53,10 +53,10 @@ def test_catalog_and_scripts_work_after_project_and_data_roots_move(
assert workflows.returncode == 0, workflows.stderr
assert scripts.returncode == 0, scripts.stderr
assert len(json.loads(workflows.stdout)) == 35
assert len(json.loads(workflows.stdout)) == 39
commands = json.loads(scripts.stdout)
assert len(commands) == 55
assert len({row["command_id"] for row in commands}) == 55
assert len(commands) == 60
assert len({row["command_id"] for row in commands}) == 60
assert all(row["legacy_script_id"].startswith(f"{row['module']}:") for row in commands)
+4 -8
View File
@@ -80,10 +80,10 @@ def test_default_registry_covers_every_executable_workflow_entry() -> None:
for entry in entries:
workflow_entries.add(f"{workflow['module']}:{entry}")
assert scheduled_workflows == 35
assert scheduled_workflows == 39
assert manual_workflows == 0
assert len(workflow_entries) == 50
assert len(catalog.scripts) == 55
assert len(workflow_entries) == 55
assert len(catalog.scripts) == 60
assert workflow_entries < set(catalog.script_ids)
assert {
"content.mapping.rebuild",
@@ -179,10 +179,6 @@ def test_default_registry_covers_every_executable_workflow_entry() -> None:
catalog.get("content.marketing_report.generate").notification_workflow_id
== "content.marketing_report.daily"
)
assert (
catalog.get("content.login.refresh").notification_workflow_id
== "content.relogin.weekly"
)
@pytest.mark.parametrize(
@@ -284,7 +280,7 @@ def test_cli_lists_and_dry_runs_registered_commands(tmp_path: Path) -> None:
assert main(["scripts", "list", "--json"], settings=settings, stdout=output) == 0
rows = json.loads(output.getvalue())
assert len(rows) == 55
assert len(rows) == 60
assert {row["module"] for row in rows} == {
"content_marketing",
"product_commerce",
+1 -1
View File
@@ -57,6 +57,6 @@ def test_every_workflow_and_manual_script_resolves_inside_new_project() -> None:
assert target.is_relative_to(PROJECT_ROOT)
commands = ScriptCatalog.load_default()
assert len(commands.scripts) == 55
assert len(commands.scripts) == 60
assert len(commands.command_ids) == len(set(commands.command_ids))
assert all(command.path.is_relative_to(PROJECT_ROOT) for command in commands.scripts)