418 lines
18 KiB
Python
418 lines
18 KiB
Python
# ============================================================
|
||
# 自动化网关驱动黄金测试（round-40 方向 S · 矩阵 76）
|
||
# 覆盖：网关定时驱动（POST /api/automation/tick 到期触发 + 状态落盘）、
|
||
# 调度器持久化往返（规则/到期基线/暂停/运行记录宽容恢复）、
|
||
# 真实业务动作接线（G2 只读建议 / G3 出确认卡 / G4 受控自动执行，
|
||
# 未登记动作仍被 P3 门禁拦截）。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from datetime import UTC, datetime, timedelta
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from server.agent_core.automation import (
|
||
AutomationGate,
|
||
AutomationScheduler,
|
||
Rule,
|
||
RuleExecutor,
|
||
RuleRegistry,
|
||
automation_authorization,
|
||
)
|
||
from server.aps_domain import workflow as wf
|
||
from server.auth.context import IdentityContext, bind_identity, reset_identity
|
||
from server.contracts import IntentResult
|
||
|
||
|
||
class _MemStore:
|
||
"""极简内存 store：world + next_id + save（黄金测试隔离）。"""
|
||
|
||
def __init__(self, world: dict | None = None) -> None:
|
||
self.data: dict = world if world is not None else {"auditEvents": []}
|
||
self._seq = 0
|
||
|
||
def next_id(self, _kind: str) -> int:
|
||
self._seq += 1
|
||
return self._seq
|
||
|
||
def save(self) -> None:
|
||
pass
|
||
|
||
|
||
class _CheckpointStore:
|
||
"""极简检查点仓（避免 G4 自动执行时写真实快照文件）。"""
|
||
|
||
def create(self, *_args, **_kwargs) -> dict:
|
||
return {"pairId": "pair-auto"}
|
||
|
||
|
||
def _approve_escalation_twice(gate: AutomationGate, confirm_id: str) -> dict:
|
||
"""P3 升权确认卡需两重审批且由不同用户完成（SOD），返回最终决策记录。"""
|
||
first = gate.approve(confirm_id)
|
||
assert first is not None and first["needsSecondConfirm"] is True
|
||
token = bind_identity(IdentityContext(
|
||
2002, "collaborator", "Collaborator", "platform", roles=("planner",),
|
||
))
|
||
try:
|
||
second = gate.approve(confirm_id)
|
||
finally:
|
||
reset_identity(token)
|
||
assert second is not None and second["needsSecondConfirm"] is False
|
||
return second
|
||
|
||
|
||
def _scheduled_world_store() -> _MemStore:
|
||
"""种子世界 + 规则引擎试排一版：scheduleVersions 含 DRAFT 版本（发布动作可用）。"""
|
||
from server.aps_domain.rush import _engine_params, _sandbox_counter
|
||
from server.engines import get_engine
|
||
from server.state.seed import seed_world
|
||
world = seed_world()
|
||
get_engine("RULE").solve(world, _engine_params(world, "DELIVERY_FIRST"), _sandbox_counter())
|
||
assert world.get("scheduleVersions"), "种子世界必须含 DRAFT 版本"
|
||
# 演示种子按约束配置会产出硬约束冲突（发布门禁 fail-closed）；
|
||
# 测试关注自动化接线，因此把演示冲突标记为已解决，使发布路径可达。
|
||
for _conflict in world.get("conflicts") or []:
|
||
_conflict["isResolved"] = True
|
||
return _MemStore(world)
|
||
|
||
|
||
def _business_bridge(store):
|
||
"""与网关 AutomationRuntime 相同的业务动作桥（rule action → 真实意图）。"""
|
||
def bridge(binding, params, ctx):
|
||
payload = dict(binding.params or {})
|
||
payload.update(dict((params or {}).get("payload") or {}))
|
||
intent = IntentResult(intent=binding.intent, params=payload,
|
||
confidence=1.0, source="RULE_FAST")
|
||
return wf.run_automation_intent(
|
||
store, str(ctx.get("session_id") or "s"), intent,
|
||
actor=str(ctx.get("actor") or "automation"),
|
||
auto=bool(ctx.get("auto")),
|
||
)
|
||
return bridge
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _cleanup(monkeypatch, tmp_path):
|
||
"""自动化运行时/门禁队列隔离：状态文件指向临时目录，运行时每次重建。"""
|
||
import server.gateway.app as gw
|
||
from server.agent_core import harness
|
||
harness._approval_store.clear()
|
||
monkeypatch.setenv("APS_AUTOMATION_STATE_PATH", str(tmp_path / "automation_state.json"))
|
||
gw._reset_automation_runtime()
|
||
yield
|
||
gw._reset_automation_runtime()
|
||
harness._approval_store.clear()
|
||
|
||
|
||
# ---------------- P1：调度器持久化往返（规则/到期基线/暂停/运行记录） ----------------
|
||
|
||
def test_scheduler_persistence_roundtrip_rules_and_due_state(tmp_path):
|
||
"""规则 + lastRun 到期基线 + 暂停 + 运行记录落盘，重启宽容恢复且到期连续。"""
|
||
base = datetime(2026, 8, 2, 0, 0, 0, tzinfo=UTC)
|
||
clock = {"now": base}
|
||
store = _MemStore()
|
||
registry = RuleRegistry()
|
||
executor = RuleExecutor(registry=registry)
|
||
registry.register(Rule("r-every", "room-1", {"on": "schedule", "every": 60},
|
||
gear="G2", action="notify"))
|
||
registry.register(Rule("r-cron", "room-1", {"on": "schedule", "cron": "0 8 * * *"},
|
||
gear="G2", action="notify"))
|
||
registry.register(Rule("r-paused", "room-1", {"on": "schedule", "every": 120},
|
||
gear="G2", action="notify"))
|
||
scheduler = AutomationScheduler(executor, now=lambda: clock["now"])
|
||
scheduler.pause_rule("r-paused")
|
||
scheduler.pause() # 整体暂停也要持久化
|
||
# 首个 tick：间隔规则到期触发；cron 仅建基线；暂停规则跳过
|
||
first = scheduler.tick(world=store.data, next_id=store.next_id, save=store.save)
|
||
assert [r.rule_id for r in first] == []
|
||
scheduler.resume()
|
||
first = scheduler.tick(world=store.data, next_id=store.next_id, save=store.save)
|
||
assert [r.rule_id for r in first] == ["r-every"]
|
||
|
||
state_path = tmp_path / "automation_state.json"
|
||
scheduler.save_state(state_path)
|
||
assert state_path.exists()
|
||
|
||
# 重启：新调度器宽容加载（规则/暂停/到期基线/运行记录）
|
||
clock["now"] = base + timedelta(seconds=61)
|
||
loaded = AutomationScheduler.load_state(state_path)
|
||
assert [r.rule_id for r in loaded.executor.registry.list()] == ["r-every", "r-cron", "r-paused"]
|
||
assert loaded.paused_rules() == ["r-paused"]
|
||
assert loaded.paused is False # 整体暂停状态恢复为 resume 后
|
||
# 恢复的 lastRun 基线：+61s 后 r-every 再到期触发；r-cron 未到期；r-paused 仍暂停
|
||
runs = loaded.tick(world=store.data, next_id=store.next_id, save=store.save,
|
||
now=clock["now"])
|
||
assert [r.rule_id for r in runs] == ["r-every"]
|
||
# 运行记录恢复（可回放）
|
||
assert loaded.executor.runs and any(r.rule_id == "r-every" for r in loaded.executor.runs.values())
|
||
|
||
# 宽容：损坏 JSON / 缺失文件 → 空调度不抛错
|
||
bad = tmp_path / "bad.json"
|
||
bad.write_text("{not json", encoding="utf-8")
|
||
assert AutomationScheduler.load_state(bad).executor.registry.list() == []
|
||
assert AutomationScheduler.load_state(tmp_path / "missing.json").executor.registry.list() == []
|
||
|
||
|
||
def test_rule_params_roundtrip_through_persistence():
|
||
"""业务动作入参（如重排 level）随规则序列化往返。"""
|
||
rule = Rule("r-x", "room-1", {"on": "schedule", "every": 30}, gear="G2",
|
||
action="reschedule", params={"level": "L3"})
|
||
restored = Rule.from_dict(rule.as_dict())
|
||
assert restored == rule
|
||
assert restored.params == {"level": "L3"}
|
||
|
||
|
||
# ---------------- P2：网关定时驱动（tick 到期触发 + 审计 + 落盘） ----------------
|
||
|
||
def test_gateway_tick_endpoint_triggers_due_rule_and_persists(monkeypatch):
|
||
"""POST /api/automation/tick：到期规则触发（真实业务动作出卡）+ AUTOMATION 审计 + 状态落盘。"""
|
||
import server.gateway.app as gw
|
||
from tests.auth_provider import install_test_auth
|
||
|
||
install_test_auth(monkeypatch, "tenant-batch-test")
|
||
store = _scheduled_world_store()
|
||
store.world_key = "personal-1001"
|
||
store.tenant_uuid = "tenant-batch-test"
|
||
monkeypatch.setattr(gw, "get_store", lambda: store)
|
||
|
||
client = TestClient(gw.create_app())
|
||
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
||
assert login.status_code == 200, login.text
|
||
|
||
runtime = gw._get_automation_runtime()
|
||
runtime.executor.registry.register(
|
||
Rule("r-pub", "room-1", {"on": "schedule", "every": 3600},
|
||
gear="G2", action="commit")
|
||
)
|
||
resp = client.post("/api/automation/tick", json={})
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["ok"] is True and body["triggered"] == 1
|
||
run = body["runs"][0]
|
||
assert run["ruleId"] == "r-pub"
|
||
assert run["status"] == "STAGED" # G2：P2 真实业务动作只出确认卡
|
||
assert run["payload"]["confirmId"]
|
||
# 既有 AUTOMATION 审计已写（含 gear/rule/trigger）
|
||
run_events = [e for e in store.data["auditEvents"] if e["action"] == "automation.run"]
|
||
assert run_events and run_events[-1]["rationale"]["runId"] == run["runId"]
|
||
assert run_events[-1]["rationale"]["gear"] == "G2"
|
||
# 门禁出卡审计（schedule.publish.stage，GATE 类）
|
||
assert any(e["action"] == "schedule.publish.stage" for e in store.data["auditEvents"])
|
||
# 状态已落盘（规则可跨重启恢复）
|
||
state_path = Path(os.environ["APS_AUTOMATION_STATE_PATH"])
|
||
assert state_path.exists()
|
||
import json as _json
|
||
state = _json.loads(state_path.read_text(encoding="utf-8"))
|
||
assert any(rule.get("ruleId") == "r-pub" for rule in state["rules"])
|
||
# 状态端点只读投影
|
||
status = client.get("/api/automation/status")
|
||
assert status.status_code == 200
|
||
assert status.json()["ruleCount"] >= 1
|
||
|
||
|
||
def test_gateway_confirm_mints_bound_g4_grant_and_executes_real_bridge(monkeypatch):
|
||
import server.aps_domain.workflow as wf_module
|
||
import server.gateway.app as gw
|
||
from tests.auth_provider import install_test_auth
|
||
|
||
class ScopedProjectStore:
|
||
def active_world_key(self):
|
||
return "p1"
|
||
|
||
def require_active_write(self):
|
||
return None
|
||
|
||
def snapshot(self, include_messages=False):
|
||
data = {
|
||
"projects": [{"id": "p1", "name": "Automation", "workDir": ""}],
|
||
"sessions": [{"id": "s-g4", "projectId": "p1"}],
|
||
"files": [],
|
||
"activeProjectId": "p1",
|
||
"activeSessionId": "s-g4",
|
||
}
|
||
if include_messages:
|
||
data["messages"] = {"s-g4": []}
|
||
return data
|
||
|
||
install_test_auth(monkeypatch, "tenant-g4-gateway")
|
||
store = _scheduled_world_store()
|
||
store.world_key = "p1"
|
||
store.tenant_uuid = "tenant-g4-gateway"
|
||
monkeypatch.setattr(gw, "get_store", lambda: store)
|
||
monkeypatch.setattr("server.state.projects.get_project_store", lambda: ScopedProjectStore())
|
||
monkeypatch.setattr(wf_module, "get_checkpoints", lambda: _CheckpointStore())
|
||
|
||
client = TestClient(gw.create_app())
|
||
assert client.post(
|
||
"/api/auth/login", json={"username": "planner", "password": "test"},
|
||
).status_code == 200
|
||
runtime = gw._get_automation_runtime()
|
||
rule = Rule(
|
||
"r-g4-api", "room-1", {"on": "schedule", "every": 60},
|
||
gear="G4", action="commit", params={"track": "fixed"},
|
||
)
|
||
runtime.executor.registry.register(rule)
|
||
token = bind_identity(IdentityContext(
|
||
1001, "planner", "计划员", "tenant-g4-gateway", roles=("planner",),
|
||
))
|
||
try:
|
||
staged = runtime.executor.execute(
|
||
rule,
|
||
rule.trigger,
|
||
world=store.data,
|
||
next_id=store.next_id,
|
||
session_id="s-g4",
|
||
save=store.save,
|
||
)
|
||
finally:
|
||
reset_identity(token)
|
||
assert staged.status == "GATE_REQUIRED"
|
||
confirm_id = staged.payload["escalateConfirmId"]
|
||
|
||
first = client.post(
|
||
"/api/actions/confirm", json={"confirmId": confirm_id, "approve": True},
|
||
)
|
||
assert first.status_code == 200, first.text
|
||
assert first.json()["secondConfirmRequired"] is True, first.json()
|
||
|
||
assert client.post(
|
||
"/api/auth/login", json={"username": "collaborator", "password": "test"},
|
||
).status_code == 200
|
||
second = client.post(
|
||
"/api/actions/confirm", json={"confirmId": confirm_id, "approve": True},
|
||
)
|
||
assert second.status_code == 200, second.text
|
||
assert second.json()["secondConfirmRequired"] is False
|
||
assert "已获 G4" in second.json()["message"]
|
||
grant = runtime.executor.gate.grant_for(rule.rule_id)
|
||
assert grant is not None and grant.authorization_digest
|
||
|
||
token = bind_identity(IdentityContext(
|
||
1002, "collaborator", "协作成员", "tenant-g4-gateway", roles=("planner",),
|
||
))
|
||
try:
|
||
executed = runtime.executor.execute(
|
||
rule,
|
||
rule.trigger,
|
||
world=store.data,
|
||
next_id=store.next_id,
|
||
session_id="s-g4",
|
||
save=store.save,
|
||
)
|
||
finally:
|
||
reset_identity(token)
|
||
assert executed.status == "EXECUTED" and executed.gear == "G4"
|
||
assert store.data["scheduleVersions"][-1]["status"] == "PUBLISHED"
|
||
assert any(
|
||
event.get("action") == "automation.gear.escalate.granted"
|
||
for event in store.data["auditEvents"]
|
||
)
|
||
|
||
|
||
# ---------------- P3：真实业务动作接线（G2/G3/G4 档位语义） ----------------
|
||
|
||
def test_business_action_wiring_gear_semantics(monkeypatch):
|
||
"""G2 只读建议 / G2 出卡 / G3 出确认卡 / G4 受控自动执行 / 未登记动作 P3 拦截。"""
|
||
import server.aps_domain.workflow as wf_module
|
||
store = _scheduled_world_store()
|
||
monkeypatch.setattr(wf_module, "get_checkpoints", lambda: _CheckpointStore())
|
||
|
||
registry = RuleRegistry()
|
||
gate = AutomationGate()
|
||
executor = RuleExecutor(registry=registry, gate=gate, business_bridge=_business_bridge(store))
|
||
|
||
# 动作注册表：commit → schedule.publish；reschedule → flex.reschedule；未登记 → None
|
||
assert executor.binding_for("commit").intent == "schedule.publish"
|
||
assert executor.binding_for("reschedule").intent == "flex.reschedule"
|
||
assert executor.binding_for("reschedule").params == {"level": "L2"}
|
||
assert executor.binding_for("notify") is None
|
||
|
||
# G2 只读建议：suggest 动作不产生任何写入
|
||
s = Rule("r-s", "room-1", {"on": "schedule", "every": 60}, gear="G2", action="suggest")
|
||
registry.register(s)
|
||
run = executor.execute(s, s.trigger, world=store.data, next_id=store.next_id,
|
||
session_id="s", save=store.save)
|
||
assert run.status == "PROPOSED" and run.payload["summary"]
|
||
|
||
# G2 沙盒：commit（P2 真实业务动作）只出确认卡，绝不自动执行
|
||
g2 = Rule("r-g2", "room-1", {"on": "schedule", "every": 60}, gear="G2", action="commit")
|
||
registry.register(g2)
|
||
run = executor.execute(g2, g2.trigger, world=store.data, next_id=store.next_id,
|
||
session_id="s", save=store.save)
|
||
assert run.status == "STAGED"
|
||
assert run.payload["confirmId"]
|
||
assert store.data["scheduleVersions"][-1]["status"] == "DRAFT" # 未批准 → 不发布
|
||
|
||
# G3 监督：升权凭据到手后出确认卡（P2 卡）
|
||
g3 = Rule("r-g3", "room-1", {"on": "schedule", "every": 60}, gear="G3", action="commit")
|
||
registry.register(g3)
|
||
card = gate.request("s", "r-g3", "G3", room_id="room-1")
|
||
gate.grant_from_decision(card["confirmId"], _approve_escalation_twice(gate, card["confirmId"]),
|
||
rule_id="r-g3", params=card["params"])
|
||
run = executor.execute(g3, g3.trigger, world=store.data, next_id=store.next_id,
|
||
session_id="s", save=store.save)
|
||
assert run.status == "STAGED" and run.gear == "G3"
|
||
assert run.payload["confirmId"]
|
||
assert store.data["scheduleVersions"][-1]["status"] == "DRAFT"
|
||
|
||
# G4 全自治：升权凭据 + 业务桥 → 自动批准执行真实发布（版本 PUBLISHED，全审计）
|
||
g4 = Rule("r-g4", "room-1", {"on": "schedule", "every": 60}, gear="G4", action="commit")
|
||
registry.register(g4)
|
||
card = gate.request(
|
||
"s",
|
||
"r-g4",
|
||
"G4",
|
||
room_id="room-1",
|
||
authorization=automation_authorization(g4, executor.binding_for(g4.action)),
|
||
)
|
||
gate.grant_from_decision(card["confirmId"], _approve_escalation_twice(gate, card["confirmId"]),
|
||
rule_id="r-g4", params=card["params"])
|
||
run = executor.execute(g4, g4.trigger, world=store.data, next_id=store.next_id,
|
||
session_id="s", save=store.save)
|
||
assert run.status == "EXECUTED" and run.gear == "G4"
|
||
assert store.data["scheduleVersions"][-1]["status"] == "PUBLISHED" # 真实业务动作已执行
|
||
g4_events = [e for e in store.data["auditEvents"] if e["action"] == "automation.run"
|
||
and e["rationale"]["runId"] == run.run_id]
|
||
assert g4_events and g4_events[0]["rationale"]["gear"] == "G4"
|
||
|
||
|
||
def test_unregistered_action_blocked_by_p3_gate_even_at_g4(monkeypatch):
|
||
"""未登记真实业务动作（无 binding / 未接线业务桥）→ G4 受控自动仍被 P3 门禁拦截。"""
|
||
import server.aps_domain.workflow as wf_module
|
||
store = _scheduled_world_store()
|
||
monkeypatch.setattr(wf_module, "get_checkpoints", lambda: _CheckpointStore())
|
||
|
||
gate = AutomationGate()
|
||
# 未注入业务桥、未注册处理器 → commit 无可执行的真实业务实现
|
||
bare = RuleExecutor(registry=RuleRegistry(), gate=gate)
|
||
bare._business_bindings.pop("commit", None) # 显式模拟「未登记动作」
|
||
rule = Rule("r-g4x", "room-1", {"on": "schedule", "every": 60}, gear="G4", action="commit")
|
||
bare.registry.register(rule)
|
||
with pytest.raises(ValueError, match="BusinessActionBinding"):
|
||
gate.request(
|
||
"s",
|
||
"r-g4x",
|
||
"G4",
|
||
room_id="room-1",
|
||
authorization=automation_authorization(rule, bare.binding_for(rule.action)),
|
||
)
|
||
|
||
run = bare.execute(rule, rule.trigger, world=store.data, next_id=store.next_id,
|
||
session_id="s", save=store.save, escalate_on_demand=False)
|
||
assert run.status == "DENIED"
|
||
assert run.payload.get("unregisteredAction") is True
|
||
denied = [e for e in store.data["auditEvents"] if e["action"] == "automation.run"][-1]
|
||
assert denied["power"] == "P3" # fail-closed：未登记默认 P3
|
||
assert store.data["scheduleVersions"][-1]["status"] == "DRAFT" # 未执行任何写入
|
||
|
||
|
||
def test_automation_intent_unbound_fails_closed():
|
||
"""未接线的真实业务意图 → ValueError（fail closed，不产生任何写入）。"""
|
||
store = _scheduled_world_store()
|
||
with pytest.raises(ValueError):
|
||
wf.run_automation_intent(
|
||
store, "s", IntentResult(intent="order.cancel", params={}), auto=False)
|