mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 14:42:00 +00:00
ThearchyHelios
d62f921ed4
- 失败通知的「处理」行改成中性的「请在 AgentLab 的运行记录中查看原因」：服务重启时还在排队、从没开始执行的运行 没有断点，「修正后可继续运行」是错的；开始执行过的也可能续不了（执行第一个节点前就失败、系统升级后断点不兼容）， 能不能继续以记录页为准。没有开始时间的失败另写一行原因（没有可继续的进度），通用格式的 reason 字段同写；任务 还在排期上时补「下一次触发会照常发起」，和挂起（之后的触发都跳过）说清区别 - 通知扫描的 30 天窗口加两种例外：停在审批、有待审批的运行不受限制（创建 30 天以后才停到审批上的以前收不到 「等待审批」）；运行记录 30 天内有过变化的照扫（等了一个多月的审批批完、跑完，以前收不到「运行结束」）。早就 结束、多日没有变化的旧触发仍不扫 - test_loop_ticks_until_stopped 改成只看「发起了」和停下前后的条数差：建任务和第一拍之间跨过整分时会先记一条 错过的跳过，以前按总条数断言偶发失败。加一组把下一次放在上一个整分的用例，必然走到跨整分这一支
478 lines
22 KiB
Python
478 lines
22 KiB
Python
"""定时任务的调度循环（app/scheduler/loop.py、service.fire_once）。
|
||
|
||
- 到点只发起一次：条件 UPDATE 认领（WHERE next_run_at = 旧值）并推进到下一次，两拍并发也只有一拍认领得到；
|
||
触发记录 (任务, fire_at, kind) 唯一，撞上约束的不再发起
|
||
- 发起走 engine/launch.py 的 launch_run：记在「定时任务：<名字>」名下、带 schedule_id，输入按 fire_at 渲染；
|
||
正式运行用当前发布版本，没有发布版本被拒，拒绝的原话记成 failed_to_start；别的异常记通用说明、详情进日志
|
||
- 上一次的运行还没结束（排队、运行中、停在审批 / 挂起都算）记 skipped_overlap，原因按挡路的运行写清楚：仍在
|
||
进行中、正在等待审批、已挂起（要去「记录」里继续运行或放弃）；挂起的那次按失败推一次通知，不是每次触发都推
|
||
- 错过的触发：最近一次在宽限内补一次（kind=misfire），更早的积压只记一条跳过；最近一次也超出宽限就记一条
|
||
skipped_misfire，推进到未来的下一次，不补积压
|
||
- start / stop：起一个 asyncio 任务按拍执行，stop 干净地停下（不落下新起的通知扫描，接收端挂住时宽限一过就取消）；
|
||
AGENTLAB_SCHEDULER 关着时 start 不起循环
|
||
|
||
工作流、字段、任务名都是虚构的。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import time
|
||
import uuid
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
import httpx
|
||
import pytest
|
||
from sqlalchemy import delete, func, select
|
||
|
||
from app import scheduler
|
||
from app.core.config import Settings
|
||
from app.core.crypto import encrypt
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import Approval, Run, Schedule, ScheduleFire, Workflow, WorkflowVersion
|
||
from app.engine.runner import run_manager
|
||
from app.scheduler import cron, loop, service, webhook
|
||
|
||
UTC = timezone.utc
|
||
#: 上海 2026-10-05（周一）09:00
|
||
NINE = datetime(2026, 10, 5, 1, 0, tzinfo=UTC)
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def engine_up():
|
||
await run_manager.setup()
|
||
async with SessionLocal() as session:
|
||
await session.execute(delete(Schedule))
|
||
await session.commit()
|
||
yield
|
||
await run_manager.shutdown()
|
||
|
||
|
||
def node(nid, ntype, **config):
|
||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
|
||
|
||
|
||
def graph(text: str) -> dict:
|
||
return {
|
||
"nodes": [node("start", "input", fields=[{"name": "stat_date", "required": True}]),
|
||
node("out", "output", fields=[{"name": "r", "value": f"{text} {{{{ input.stat_date }}}}"}])],
|
||
"edges": [{"source": "start", "target": "out"}],
|
||
}
|
||
|
||
|
||
async def workflow(*, published: bool = True, canvas: str = "画布", snapshot: str = "发布版") -> str:
|
||
async with SessionLocal() as session:
|
||
wf = Workflow(name=f"门店周报-{uuid.uuid4().hex[:6]}", graph=graph(canvas),
|
||
status="published" if published else "draft", published_version=1 if published else None)
|
||
session.add(wf)
|
||
await session.flush()
|
||
session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=graph(snapshot)))
|
||
await session.commit()
|
||
return wf.id
|
||
|
||
|
||
async def schedule(workflow_id: str, *, cron: str = "0 9 * * *", next_run_at: datetime | None = NINE,
|
||
run_class: str = "formal", grace: int = 3600, enabled: bool = True, name: str = "周一早报") -> str:
|
||
async with SessionLocal() as session:
|
||
row = Schedule(name=name, workflow_id=workflow_id, cron=cron, timezone="Asia/Shanghai", run_class=run_class,
|
||
input_template={"stat_date": "{{ schedule.yesterday }}"}, enabled=enabled,
|
||
misfire_grace_s=grace, next_run_at=next_run_at)
|
||
session.add(row)
|
||
await session.commit()
|
||
return row.id
|
||
|
||
|
||
async def fires(schedule_id: str) -> list[ScheduleFire]:
|
||
async with SessionLocal() as session:
|
||
return list((await session.execute(
|
||
select(ScheduleFire).where(ScheduleFire.schedule_id == schedule_id).order_by(ScheduleFire.created_at)
|
||
)).scalars())
|
||
|
||
|
||
async def runs_of(schedule_id: str) -> list[Run]:
|
||
async with SessionLocal() as session:
|
||
return list((await session.execute(select(Run).where(Run.schedule_id == schedule_id))).scalars())
|
||
|
||
|
||
async def next_of(schedule_id: str) -> datetime | None:
|
||
async with SessionLocal() as session:
|
||
return (await session.get(Schedule, schedule_id)).next_run_at
|
||
|
||
|
||
async def settle(run_id: str) -> Run:
|
||
for _ in range(400):
|
||
async with SessionLocal() as session:
|
||
row = await session.get(Run, run_id)
|
||
if row.status in ("succeeded", "failed", "cancelled") and row.manifest_seq is not None:
|
||
await run_manager.wait_idle(run_id)
|
||
return row
|
||
await asyncio.sleep(0.03)
|
||
raise AssertionError(f"运行 {run_id} 停在 {row.status}")
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 到点发起
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_due_schedule_launches_once_and_moves_on():
|
||
sid = await schedule(await workflow())
|
||
await loop.fire_due(NINE + timedelta(seconds=10))
|
||
[fire] = await fires(sid)
|
||
assert (fire.kind, fire.status, fire.fire_at, fire.reason) == ("scheduled", "started", NINE, None)
|
||
assert fire.inputs == {"stat_date": "2026-10-04"}
|
||
run = await settle(fire.run_id)
|
||
assert run.schedule_id == sid and run.run_class == "formal" and run.status == "succeeded"
|
||
assert run.started_by == "定时任务：周一早报"
|
||
assert run.input == {"stat_date": "2026-10-04"}
|
||
# 正式运行用的是发布版本，不是画布
|
||
assert run.version == 1 and run.output == {"r": "发布版 2026-10-04"}
|
||
assert await next_of(sid) == NINE + timedelta(days=1)
|
||
# 同一拍再来一次：已经推进到明天，不再发起
|
||
await loop.fire_due(NINE + timedelta(seconds=20))
|
||
assert len(await fires(sid)) == 1 and len(await runs_of(sid)) == 1
|
||
|
||
|
||
async def test_exploratory_schedule_uses_the_canvas():
|
||
sid = await schedule(await workflow(published=False), run_class="exploratory")
|
||
await loop.fire_due(NINE + timedelta(seconds=5))
|
||
[fire] = await fires(sid)
|
||
run = await settle(fire.run_id)
|
||
assert run.run_class == "exploratory" and run.output == {"r": "画布 2026-10-04"}
|
||
|
||
|
||
async def test_two_concurrent_ticks_launch_only_once():
|
||
sid = await schedule(await workflow())
|
||
now = NINE + timedelta(seconds=3)
|
||
await asyncio.gather(loop.fire_due(now), loop.fire_due(now), loop.fire_due(now))
|
||
assert len(await fires(sid)) == 1
|
||
[run] = await runs_of(sid)
|
||
await settle(run.id)
|
||
|
||
|
||
async def test_concurrent_ticks_with_different_clocks_claim_once():
|
||
"""一拍准点、一拍晚了两分钟（会被当成补发、kind 不同、撞不上唯一约束）：只有认领到的那一拍处理"""
|
||
sid = await schedule(await workflow())
|
||
await asyncio.gather(loop.fire_due(NINE + timedelta(seconds=3)), loop.fire_due(NINE + timedelta(minutes=2)))
|
||
[fire] = await fires(sid)
|
||
[run] = await runs_of(sid)
|
||
assert fire.run_id == run.id
|
||
await settle(run.id)
|
||
|
||
|
||
async def test_a_fire_already_recorded_is_not_launched_again():
|
||
"""认领之外的第二道保险：同一 (任务, fire_at, kind) 已有记录，fire_once 交回那一条，不再发起"""
|
||
sid = await schedule(await workflow())
|
||
first = await service.fire_once(sid, NINE, "scheduled")
|
||
again = await service.fire_once(sid, NINE, "scheduled")
|
||
assert again.id == first.id and len(await fires(sid)) == 1
|
||
[run] = await runs_of(sid)
|
||
await settle(run.id)
|
||
|
||
|
||
async def test_disabled_schedule_is_not_fired():
|
||
sid = await schedule(await workflow(), enabled=False)
|
||
await loop.fire_due(NINE + timedelta(seconds=5))
|
||
assert await fires(sid) == [] and await runs_of(sid) == []
|
||
|
||
|
||
@pytest.mark.parametrize("status, approval, says", [
|
||
("queued", None, "仍在进行中"),
|
||
("running", None, "仍在进行中"),
|
||
("interrupted", "pending", "正在等待审批"),
|
||
# 批了、还没接着执行（审批记成 answered 到运行改回 running 之间）：仍是停在审批，不是挂起
|
||
("interrupted", "answered", "正在等待审批"),
|
||
# 服务重启挂起的：没有待审批，也不会自己接着执行，原因要说清楚该去做什么
|
||
("interrupted", None, "已挂起（通常是服务重启所致），请在「记录」中继续运行或放弃这次运行"),
|
||
])
|
||
async def test_previous_run_still_going_means_skip(status, approval, says):
|
||
wf = await workflow()
|
||
sid = await schedule(wf)
|
||
async with SessionLocal() as session:
|
||
busy = Run(workflow_id=wf, workflow_name="门店周报", status=status, schedule_id=sid, run_class="formal")
|
||
session.add(busy)
|
||
await session.flush()
|
||
if approval:
|
||
session.add(Approval(run_id=busy.id, node_id="gate", title="审核周报", status=approval))
|
||
await session.commit()
|
||
await loop.fire_due(NINE + timedelta(seconds=5))
|
||
[fire] = await fires(sid)
|
||
assert (fire.status, fire.run_id) == ("skipped_overlap", None)
|
||
# 运行编号和界面一样写成 #前 6 位；不再说「按设置」——重叠处理只有跳过一档，用户没选过什么
|
||
assert f"上一次运行 #{busy.id[:6]} " in fire.reason and says in fire.reason, fire.reason
|
||
assert fire.reason.endswith("本次触发已跳过") and "按设置" not in fire.reason
|
||
assert [r.id for r in await runs_of(sid)] == [busy.id]
|
||
assert await next_of(sid) == NINE + timedelta(days=1)
|
||
|
||
|
||
async def test_suspended_run_is_reported_once_and_later_fires_say_what_to_do(monkeypatch):
|
||
"""上一次运行在服务重启时被挂起：它不会自己接着执行，按规则之后的触发都跳过。跳过的原因写明要去继续或放弃，
|
||
订阅了失败通知的推一次（不是每次触发都推）；有人放弃了它之后，下一次照常发起"""
|
||
requests: list[httpx.Request] = []
|
||
|
||
def receive(request: httpx.Request) -> httpx.Response:
|
||
requests.append(request)
|
||
return httpx.Response(200, json={})
|
||
|
||
monkeypatch.setattr(webhook, "_transport", httpx.MockTransport(receive))
|
||
wf = await workflow()
|
||
sid = await schedule(wf)
|
||
async with SessionLocal() as session:
|
||
await session.execute(Schedule.__table__.update().where(Schedule.id == sid).values(
|
||
webhook_format="generic", webhook_url=encrypt("https://hooks.example.com/agentlab/inbox-5c1d"),
|
||
webhook_events=["finished", "failed", "awaiting_approval"]))
|
||
held = Run(workflow_id=wf, workflow_name="门店周报", status="interrupted", schedule_id=sid, run_class="formal",
|
||
error="服务重启，运行已挂起，可从断点继续运行")
|
||
session.add(held)
|
||
await session.flush()
|
||
session.add(ScheduleFire(schedule_id=sid, fire_at=NINE - timedelta(days=1), kind="scheduled",
|
||
status="started", run_id=held.id))
|
||
await session.commit()
|
||
for day in range(3):
|
||
now = NINE + timedelta(days=day, seconds=5)
|
||
await loop.fire_due(now)
|
||
await webhook.notify_due(now)
|
||
skipped = [f for f in await fires(sid) if f.status == "skipped_overlap"]
|
||
assert len(skipped) == 3
|
||
assert all(f"#{held.id[:6]} 已挂起" in f.reason and "继续运行" in f.reason for f in skipped)
|
||
[pushed] = requests
|
||
body = json.loads(pushed.content)
|
||
assert (body["event"], body["status"], body["run_id"]) == ("failed", "interrupted", held.id)
|
||
assert "挂起" in body["title"] and "继续运行" in body["text"]
|
||
# 放弃之后，下一次按时发起
|
||
async with SessionLocal() as session:
|
||
await session.execute(Run.__table__.update().where(Run.id == held.id).values(status="cancelled"))
|
||
await session.commit()
|
||
await loop.fire_due(NINE + timedelta(days=3, seconds=5))
|
||
newest = (await fires(sid))[-1]
|
||
assert (newest.status, newest.fire_at) == ("started", NINE + timedelta(days=3))
|
||
await settle(newest.run_id)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 错过的触发
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_missed_within_grace_is_caught_up_once_with_inputs_of_its_own_day():
|
||
"""服务停到 00:30（上海 10 月 6 日）才起来：补一次 9 点那一次，昨天还是 10 月 4 日"""
|
||
sid = await schedule(await workflow(), grace=86400)
|
||
late = NINE + timedelta(hours=15, minutes=30)
|
||
await loop.fire_due(late)
|
||
[fire] = await fires(sid)
|
||
assert (fire.kind, fire.status, fire.fire_at) == ("misfire", "started", NINE)
|
||
assert fire.inputs == {"stat_date": "2026-10-04"}
|
||
await settle(fire.run_id)
|
||
assert await next_of(sid) == NINE + timedelta(days=1)
|
||
|
||
|
||
async def test_backlog_is_skipped_and_only_the_latest_is_caught_up():
|
||
"""每小时一次，停了三个多小时：01–03 点的积压记一条跳过，04 点那次在宽限内补发"""
|
||
sid = await schedule(await workflow(), cron="0 * * * *", grace=3600)
|
||
await loop.fire_due(NINE + timedelta(hours=3, minutes=20))
|
||
skipped, caught_up = await fires(sid)
|
||
assert (skipped.kind, skipped.status, skipped.fire_at, skipped.run_id) == ("misfire", "skipped_misfire", NINE, None)
|
||
assert "2026-10-05 09:00" in skipped.reason and "2026-10-05 11:00" in skipped.reason
|
||
assert (caught_up.kind, caught_up.status, caught_up.fire_at) == ("misfire", "started", NINE + timedelta(hours=3))
|
||
assert len(await runs_of(sid)) == 1
|
||
await settle(caught_up.run_id)
|
||
assert await next_of(sid) == NINE + timedelta(hours=4)
|
||
|
||
|
||
async def test_missed_beyond_grace_is_skipped_without_launching():
|
||
sid = await schedule(await workflow(), grace=600)
|
||
await loop.fire_due(NINE + timedelta(hours=2))
|
||
[fire] = await fires(sid)
|
||
assert (fire.kind, fire.status, fire.fire_at, fire.run_id) == ("misfire", "skipped_misfire", NINE, None)
|
||
assert "10 分钟" in fire.reason and "2026-10-05 09:00" in fire.reason
|
||
assert await runs_of(sid) == []
|
||
assert await next_of(sid) == NINE + timedelta(days=1)
|
||
|
||
|
||
async def test_days_of_downtime_leave_one_skip_record():
|
||
sid = await schedule(await workflow(), grace=600, next_run_at=NINE - timedelta(days=4))
|
||
await loop.fire_due(NINE + timedelta(hours=2))
|
||
[fire] = await fires(sid)
|
||
assert (fire.status, fire.fire_at) == ("skipped_misfire", NINE)
|
||
assert "2026-10-01 09:00" in fire.reason and "2026-10-05 09:00" in fire.reason
|
||
assert await runs_of(sid) == [] and await next_of(sid) == NINE + timedelta(days=1)
|
||
|
||
|
||
async def test_on_time_fire_ignores_a_small_grace():
|
||
"""宽限写 0 也不影响准点的触发：拍与拍之间的十几秒不算错过"""
|
||
sid = await schedule(await workflow(), grace=0)
|
||
await loop.fire_due(NINE + timedelta(seconds=14))
|
||
[fire] = await fires(sid)
|
||
assert (fire.kind, fire.status) == ("scheduled", "started")
|
||
await settle(fire.run_id)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 发起被拒、出错
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_formal_schedule_without_published_version_records_the_refusal():
|
||
wf = await workflow(published=False)
|
||
sid = await schedule(wf)
|
||
await loop.fire_due(NINE + timedelta(seconds=5))
|
||
[fire] = await fires(sid)
|
||
async with SessionLocal() as session:
|
||
name = (await session.get(Workflow, wf)).name
|
||
assert (fire.status, fire.run_id) == ("failed_to_start", None)
|
||
assert fire.reason == f"「{name}」还没有发布版本，请先在编排页发布"
|
||
assert await runs_of(sid) == []
|
||
assert await next_of(sid) == NINE + timedelta(days=1)
|
||
|
||
|
||
async def test_deleted_workflow_records_the_refusal():
|
||
sid = await schedule(f"gone{uuid.uuid4().hex[:8]}")
|
||
await loop.fire_due(NINE + timedelta(seconds=5))
|
||
[fire] = await fires(sid)
|
||
assert (fire.status, fire.reason) == ("failed_to_start", "工作流不存在，可能已被删除")
|
||
|
||
|
||
async def test_unexpected_errors_record_a_generic_reason(monkeypatch, caplog):
|
||
from app.engine import launch
|
||
|
||
async def explode(*args, **kwargs):
|
||
raise RuntimeError("库连接断开：sqlite busy")
|
||
|
||
monkeypatch.setattr(launch, "launch_run", explode)
|
||
sid = await schedule(await workflow())
|
||
with caplog.at_level(logging.ERROR):
|
||
await loop.fire_due(NINE + timedelta(seconds=5))
|
||
[fire] = await fires(sid)
|
||
assert (fire.status, fire.reason) == ("failed_to_start", service.GENERIC_REASON)
|
||
assert "库连接断开" in caplog.text
|
||
assert await next_of(sid) == NINE + timedelta(days=1)
|
||
|
||
|
||
async def test_broken_cron_in_the_database_stops_the_schedule():
|
||
sid = await schedule(await workflow(), cron="0 25 * * *")
|
||
await loop.fire_due(NINE + timedelta(seconds=5))
|
||
[fire] = await fires(sid)
|
||
assert fire.status == "failed_to_start" and "「时」" in fire.reason
|
||
assert await next_of(sid) is None
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# start / stop
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_start_does_nothing_when_switched_off():
|
||
"""测试环境 AGENTLAB_SCHEDULER=0（conftest）：start 不起循环，stop 照样能调"""
|
||
assert await scheduler.start() is False
|
||
assert scheduler.running() is False
|
||
await scheduler.stop()
|
||
|
||
|
||
async def fire_count(schedule_id: str) -> int:
|
||
async with SessionLocal() as session:
|
||
return await session.scalar(select(func.count()).select_from(ScheduleFire)
|
||
.where(ScheduleFire.schedule_id == schedule_id))
|
||
|
||
|
||
@pytest.mark.parametrize("crossed", [False, True], ids=["on_the_minute", "minute_crossed"])
|
||
async def test_loop_ticks_until_stopped(crossed):
|
||
"""循环按拍执行、stop 之后不再有拍。只看「发起了」和停下前后的条数差，不看总条数：建任务和第一拍之间恰好跨过
|
||
整分时，第一拍会先记一条错过的跳过、再发起最近这一次——以前按总条数断言，跨整分时偶发失败。crossed 把下一次
|
||
放在上一个整分，必然走到这一种"""
|
||
due = cron.last_due("* * * * *", "UTC", datetime.now(UTC))
|
||
if crossed:
|
||
due -= timedelta(minutes=1)
|
||
sid = await schedule(await workflow(), next_run_at=due, cron="* * * * *")
|
||
on = Settings(_env_file=None, scheduler=True)
|
||
assert await scheduler.start(interval=0.05, config=on) is True
|
||
assert scheduler.running()
|
||
for _ in range(200):
|
||
if any(f.status == "started" for f in await fires(sid)):
|
||
break
|
||
await asyncio.sleep(0.02)
|
||
await scheduler.stop()
|
||
assert scheduler.running() is False
|
||
started = [f for f in await fires(sid) if f.status == "started"]
|
||
assert started
|
||
if crossed:
|
||
assert [f.status for f in await fires(sid)][0] == "skipped_misfire"
|
||
for fire in started:
|
||
await settle(fire.run_id)
|
||
# 停下之后不再有拍：把下一次拨回过去，过一会儿条数一条不多
|
||
before = await fire_count(sid)
|
||
async with SessionLocal() as session:
|
||
await session.execute(Schedule.__table__.update().where(Schedule.id == sid)
|
||
.values(next_run_at=datetime.now(UTC) - timedelta(seconds=1)))
|
||
await session.commit()
|
||
await asyncio.sleep(0.2)
|
||
assert await fire_count(sid) == before
|
||
|
||
|
||
async def test_start_twice_keeps_one_loop():
|
||
on = Settings(_env_file=None, scheduler=True)
|
||
await scheduler.start(interval=0.05, config=on)
|
||
first = loop.runner.task
|
||
await scheduler.start(interval=0.05, config=on)
|
||
assert loop.runner.task is first
|
||
await scheduler.stop()
|
||
assert first.done()
|
||
|
||
|
||
def _notify_tasks() -> list[asyncio.Task]:
|
||
return [t for t in asyncio.all_tasks() if t.get_name() == "agentlab-scheduler-notify" and not t.done()]
|
||
|
||
|
||
async def test_stop_in_the_middle_of_a_tick_leaves_no_notify_task_behind(monkeypatch):
|
||
"""stop 撞在一拍的发起当中：这一拍收尾时不再新起通知扫描。以前 stop 先把手上的通知任务清掉再等，循环收尾时
|
||
看见没有就又起了一个，它不在 stop 等的名单里，日志说「已停止」之后还在推送"""
|
||
started = asyncio.Event()
|
||
|
||
async def slow_fire_due(now=None):
|
||
started.set()
|
||
await asyncio.sleep(0.2)
|
||
return []
|
||
|
||
async def notify_due(now=None):
|
||
await asyncio.sleep(5)
|
||
return 0
|
||
|
||
monkeypatch.setattr(loop, "fire_due", slow_fire_due)
|
||
monkeypatch.setattr(webhook, "notify_due", notify_due)
|
||
runner = loop._Runner()
|
||
await runner.start(interval=0.05)
|
||
await started.wait()
|
||
await runner.stop()
|
||
assert runner.task is None and runner.notify is None
|
||
assert _notify_tasks() == []
|
||
|
||
|
||
async def test_stop_does_not_wait_out_a_hanging_webhook(monkeypatch):
|
||
"""接收端挂住、通知扫描还在重试：stop 不等满超时加重试，宽限一过就取消。关停时后面还有 run_manager 要平稳挂起
|
||
运行，docker stop 默认只给 10 秒"""
|
||
hanging = asyncio.Event()
|
||
|
||
async def fire_due(now=None):
|
||
return []
|
||
|
||
async def notify_due(now=None):
|
||
hanging.set()
|
||
await asyncio.sleep(30)
|
||
return 0
|
||
|
||
monkeypatch.setattr(loop, "fire_due", fire_due)
|
||
monkeypatch.setattr(webhook, "notify_due", notify_due)
|
||
monkeypatch.setattr(loop, "NOTIFY_GRACE_S", 0.1)
|
||
runner = loop._Runner()
|
||
await runner.start(interval=0.05)
|
||
await asyncio.wait_for(hanging.wait(), timeout=5)
|
||
began = time.monotonic()
|
||
await runner.stop()
|
||
assert time.monotonic() - began < 2
|
||
assert _notify_tasks() == []
|
||
|
||
|
||
def test_stop_fits_in_the_shutdown_budget():
|
||
"""两件事并行等，最坏也就是较长的那个宽限：给关停时的 run_manager.shutdown 留出大半的 docker stop 时间"""
|
||
assert max(loop.STOP_GRACE_S, loop.NOTIFY_GRACE_S) <= 3
|