mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-28 19:06:36 +08:00
1157 lines
35 KiB
Python
1157 lines
35 KiB
Python
"""
|
|
Scheduler 任务句柄、generation 与 AgentTask reservation 回归。
|
|
"""
|
|
|
|
import asyncio
|
|
import gc
|
|
import inspect
|
|
import threading
|
|
import warnings
|
|
from collections.abc import Generator
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.runtime.loop import main_loop_registry
|
|
from app.scheduler import bridge as scheduler_bridge
|
|
from app.scheduler import execution as scheduler_execution
|
|
from app.scheduler import lifecycle as scheduler_lifecycle
|
|
from app.scheduler import progress as scheduler_progress
|
|
from app.scheduler.facade import Scheduler
|
|
from app.scheduler.registry import ExecutionRegistry
|
|
|
|
|
|
class _ProgressStub:
|
|
"""
|
|
隔离 scheduler 生命周期测试的同步进度后端。
|
|
"""
|
|
|
|
def __init__(self, _key: str) -> None:
|
|
"""
|
|
接收进度键但不连接外部后端。
|
|
"""
|
|
|
|
def start(self) -> None:
|
|
"""
|
|
记录进度开始。
|
|
"""
|
|
|
|
def update(self, **_kwargs) -> None:
|
|
"""
|
|
忽略中间进度。
|
|
"""
|
|
|
|
def get(self):
|
|
"""
|
|
返回空的历史进度。
|
|
"""
|
|
return None
|
|
|
|
|
|
class _AsyncProgressStub:
|
|
"""
|
|
隔离 scheduler 生命周期测试的异步进度后端。
|
|
"""
|
|
|
|
def __init__(self, _key: str) -> None:
|
|
"""
|
|
接收进度键但不连接外部后端。
|
|
"""
|
|
|
|
async def get(self):
|
|
"""
|
|
返回空的历史进度。
|
|
"""
|
|
return None
|
|
|
|
async def update(self, **_kwargs) -> None:
|
|
"""
|
|
忽略中间进度。
|
|
"""
|
|
|
|
async def end(self, **_kwargs) -> None:
|
|
"""
|
|
记录终态但不访问外部缓存。
|
|
"""
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolate_scheduler_main_loop() -> Generator[None, None, None]:
|
|
"""
|
|
隔离主循环登记，避免前序兼容层假循环改变 Scheduler 投递路径。
|
|
"""
|
|
previous = main_loop_registry.current
|
|
main_loop_registry.replace_compat(None)
|
|
try:
|
|
yield
|
|
finally:
|
|
main_loop_registry.replace_compat(previous)
|
|
|
|
|
|
def _patch_progress(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
async_progress: type = _AsyncProgressStub,
|
|
) -> None:
|
|
"""
|
|
替换执行启动与进度 owner 各自持有的缓存边界。
|
|
"""
|
|
monkeypatch.setattr(scheduler_execution, "ProgressHelper", _ProgressStub)
|
|
monkeypatch.setattr(scheduler_progress, "ProgressHelper", _ProgressStub)
|
|
monkeypatch.setattr(scheduler_progress, "AsyncProgressHelper", async_progress)
|
|
|
|
|
|
def _patch_main_loop(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
loop: asyncio.AbstractEventLoop,
|
|
) -> None:
|
|
"""
|
|
让执行与桥接 owner 共享同一个测试事件循环登记。
|
|
"""
|
|
registry = SimpleNamespace(current=loop)
|
|
monkeypatch.setattr(scheduler_execution, "main_loop_registry", registry)
|
|
monkeypatch.setattr(scheduler_bridge, "main_loop_registry", registry)
|
|
|
|
|
|
def _scheduler(job_id: str, func) -> Scheduler:
|
|
"""
|
|
构造已启动但不拥有 APScheduler 线程的实例。
|
|
"""
|
|
scheduler = object.__new__(Scheduler)
|
|
scheduler._scheduler = None
|
|
scheduler._event = threading.Event()
|
|
scheduler._lock = threading.RLock()
|
|
scheduler._jobs = {
|
|
job_id: {
|
|
"name": "生命周期测试",
|
|
"provider_name": "测试",
|
|
"func": func,
|
|
"running": False,
|
|
"_generation": 1,
|
|
}
|
|
}
|
|
scheduler._lifecycle_state = "running"
|
|
scheduler._registry = ExecutionRegistry(scheduler._lock)
|
|
scheduler._registry.assign_generation(job_id, scheduler._jobs[job_id])
|
|
return scheduler
|
|
|
|
|
|
def test_internal_loop_submission_requires_job_owner() -> None:
|
|
"""
|
|
Scheduler 内部协程桥接不得再提供省略 job owner 的游离提交分支。
|
|
"""
|
|
parameter = inspect.signature(Scheduler._submit_to_loop).parameters["job_id"]
|
|
|
|
assert parameter.default is inspect.Parameter.empty
|
|
assert parameter.annotation is str
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_stop_async_cancels_and_awaits_scheduler_owned_job(monkeypatch) -> None:
|
|
"""
|
|
关闭后已投递协程必须取消并完成收尾，不得遗留 owner 句柄。
|
|
"""
|
|
started = asyncio.Event()
|
|
cleaned = asyncio.Event()
|
|
|
|
async def job():
|
|
started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
finally:
|
|
cleaned.set()
|
|
|
|
_patch_progress(monkeypatch)
|
|
scheduler = _scheduler("lifecycle-job", job)
|
|
|
|
assert scheduler.start("lifecycle-job") is True
|
|
await asyncio.wait_for(started.wait(), timeout=1)
|
|
assert scheduler._registry.handles()
|
|
|
|
await scheduler.stop_async()
|
|
|
|
assert cleaned.is_set()
|
|
assert scheduler._jobs["lifecycle-job"]["running"] is False
|
|
assert scheduler._jobs["lifecycle-job"]["last_error"] == "任务已取消"
|
|
assert scheduler._registry.handles() == ()
|
|
assert scheduler._lifecycle_state == "stopped"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_stop_during_final_progress_does_not_mark_completed_job_unsubmitted(
|
|
monkeypatch,
|
|
) -> None:
|
|
"""
|
|
业务协程已完成后，取消最终进度写入不得改写任务执行结果。
|
|
"""
|
|
finish_started = asyncio.Event()
|
|
|
|
class BlockingFinishProgress(_AsyncProgressStub):
|
|
"""
|
|
把任务停在最终进度读取阶段。
|
|
"""
|
|
|
|
async def get(self):
|
|
"""
|
|
通知收尾读取已经开始，并保持等待以验证关闭取消能够传递。
|
|
"""
|
|
finish_started.set()
|
|
await asyncio.Event().wait()
|
|
|
|
async def job() -> None:
|
|
return None
|
|
|
|
_patch_progress(monkeypatch, BlockingFinishProgress)
|
|
scheduler = _scheduler("final-progress-stop", job)
|
|
|
|
assert scheduler.start("final-progress-stop") is True
|
|
await asyncio.wait_for(finish_started.wait(), timeout=1)
|
|
|
|
await scheduler.stop_async()
|
|
|
|
assert scheduler._jobs["final-progress-stop"]["running"] is False
|
|
assert scheduler._jobs["final-progress-stop"]["last_error"] is None
|
|
assert scheduler._registry.handles() == ()
|
|
assert scheduler._registry.is_active("final-progress-stop") is False
|
|
|
|
_patch_progress(monkeypatch)
|
|
scheduler._lifecycle_state = "running"
|
|
assert scheduler.start("final-progress-stop") is True
|
|
|
|
async def wait_until_finished() -> None:
|
|
while (
|
|
scheduler._registry.handles()
|
|
or scheduler._registry.is_active("final-progress-stop")
|
|
):
|
|
await asyncio.sleep(0)
|
|
|
|
await asyncio.wait_for(wait_until_finished(), timeout=1)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_foreign_loop_submission_runs_on_main_loop_and_finishes_before_stop(
|
|
monkeypatch,
|
|
) -> None:
|
|
"""
|
|
自建事件循环提交的任务仍由应用主循环拥有并完成取消收尾。
|
|
"""
|
|
main_loop = asyncio.get_running_loop()
|
|
started = asyncio.Event()
|
|
cancelling = asyncio.Event()
|
|
cleanup_release = asyncio.Event()
|
|
execution_loop = None
|
|
|
|
async def job() -> None:
|
|
nonlocal execution_loop
|
|
execution_loop = asyncio.get_running_loop()
|
|
started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
except asyncio.CancelledError:
|
|
cancelling.set()
|
|
await cleanup_release.wait()
|
|
raise
|
|
|
|
_patch_progress(monkeypatch)
|
|
_patch_main_loop(monkeypatch, main_loop)
|
|
scheduler = _scheduler("foreign-loop-job", job)
|
|
|
|
def submit_from_foreign_loop() -> bool:
|
|
async def submit() -> bool:
|
|
return scheduler.start("foreign-loop-job")
|
|
|
|
return asyncio.run(submit())
|
|
|
|
assert await asyncio.to_thread(submit_from_foreign_loop) is True
|
|
await asyncio.wait_for(started.wait(), timeout=1)
|
|
assert execution_loop is main_loop
|
|
|
|
stop_task = asyncio.create_task(scheduler.stop_async())
|
|
await asyncio.wait_for(cancelling.wait(), timeout=1)
|
|
await asyncio.sleep(0)
|
|
assert not stop_task.done()
|
|
assert scheduler._lifecycle_state == "stopping"
|
|
|
|
stop_task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await stop_task
|
|
assert scheduler._registry.handles()
|
|
assert scheduler._lifecycle_state == "stopping"
|
|
|
|
cleanup_release.set()
|
|
|
|
async def wait_until_released() -> None:
|
|
while scheduler._registry.handles():
|
|
await asyncio.sleep(0)
|
|
|
|
await asyncio.wait_for(wait_until_released(), timeout=1)
|
|
await scheduler.stop_async()
|
|
assert scheduler._registry.handles() == ()
|
|
assert scheduler._lifecycle_state == "stopped"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_cross_thread_submission_is_registered_before_stop_snapshot(
|
|
monkeypatch,
|
|
) -> None:
|
|
"""
|
|
跨线程提交与 owner 登记必须对关闭快照表现为同一原子操作。
|
|
"""
|
|
main_loop = asyncio.get_running_loop()
|
|
registration_entered = threading.Event()
|
|
registration_release = threading.Event()
|
|
|
|
async def job() -> None:
|
|
await asyncio.Event().wait()
|
|
|
|
_patch_progress(monkeypatch)
|
|
_patch_main_loop(monkeypatch, main_loop)
|
|
scheduler = _scheduler("atomic-submit", job)
|
|
register_handle = scheduler._register_handle
|
|
|
|
def delayed_register(**kwargs) -> bool:
|
|
registration_entered.set()
|
|
registration_release.wait(timeout=1)
|
|
return register_handle(**kwargs)
|
|
|
|
monkeypatch.setattr(scheduler, "_register_handle", delayed_register)
|
|
submit_thread = threading.Thread(target=scheduler.start, args=("atomic-submit",))
|
|
submit_thread.start()
|
|
assert await asyncio.to_thread(registration_entered.wait, 1)
|
|
|
|
stop_result = []
|
|
stop_thread = threading.Thread(target=lambda: stop_result.append(scheduler._begin_stop()))
|
|
stop_thread.start()
|
|
await asyncio.sleep(0.02)
|
|
assert stop_thread.is_alive()
|
|
|
|
registration_release.set()
|
|
await asyncio.to_thread(submit_thread.join, 1)
|
|
await asyncio.to_thread(stop_thread.join, 1)
|
|
assert not submit_thread.is_alive()
|
|
assert not stop_thread.is_alive()
|
|
assert len(stop_result[0][1]) == 1
|
|
|
|
for handle in stop_result[0][1]:
|
|
scheduler._cancel_handle(handle)
|
|
await scheduler._await_cancelled_handles(stop_result[0][1])
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_submit_to_loop_tracks_internal_progress_or_finish_tasks() -> None:
|
|
"""
|
|
进度和收尾协程也必须归 Scheduler 所有并可在关闭时收口。
|
|
"""
|
|
started = asyncio.Event()
|
|
cancelled = asyncio.Event()
|
|
|
|
async def pending() -> None:
|
|
started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
except asyncio.CancelledError:
|
|
cancelled.set()
|
|
raise
|
|
|
|
scheduler = _scheduler("internal-task", lambda: None)
|
|
scheduler._submit_to_loop(
|
|
pending(),
|
|
job_id="internal-task",
|
|
generation=1,
|
|
)
|
|
|
|
await asyncio.wait_for(started.wait(), timeout=1)
|
|
assert len(scheduler._registry.handles()) == 1
|
|
|
|
await scheduler.stop_async()
|
|
|
|
assert cancelled.is_set()
|
|
assert scheduler._registry.handles() == ()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sync_job_callback_and_finish_handles_are_owned(monkeypatch) -> None:
|
|
"""
|
|
同步任务回投的进度与收尾句柄都必须纳入关闭收口。
|
|
"""
|
|
update_started = asyncio.Event()
|
|
finish_started = asyncio.Event()
|
|
gate = asyncio.Event()
|
|
cancelled = 0
|
|
|
|
class BlockingProgress:
|
|
"""
|
|
让进度和收尾停在异步后端，便于验证 owner registry。
|
|
"""
|
|
|
|
def __init__(self, _key: str) -> None:
|
|
pass
|
|
|
|
async def update(self, **_kwargs) -> None:
|
|
"""
|
|
阻塞进度更新并记录取消次数，随后继续传播取消异常。
|
|
"""
|
|
nonlocal cancelled
|
|
update_started.set()
|
|
try:
|
|
await gate.wait()
|
|
except asyncio.CancelledError:
|
|
cancelled += 1
|
|
raise
|
|
|
|
async def get(self):
|
|
"""
|
|
阻塞收尾读取并记录取消次数；正常释放时返回空进度。
|
|
"""
|
|
nonlocal cancelled
|
|
finish_started.set()
|
|
try:
|
|
await gate.wait()
|
|
except asyncio.CancelledError:
|
|
cancelled += 1
|
|
raise
|
|
return None
|
|
|
|
async def end(self, **_kwargs) -> None:
|
|
"""
|
|
提供无副作用的结束入口，使测试聚焦取消与读取过程。
|
|
"""
|
|
pass
|
|
|
|
_patch_progress(monkeypatch, BlockingProgress)
|
|
_patch_main_loop(monkeypatch, asyncio.get_running_loop())
|
|
|
|
def job(progress_callback) -> None:
|
|
progress_callback(value=50)
|
|
|
|
scheduler = _scheduler("callback-handles", job)
|
|
await asyncio.to_thread(scheduler.start, "callback-handles")
|
|
await asyncio.wait_for(update_started.wait(), timeout=1)
|
|
await asyncio.sleep(0)
|
|
|
|
assert len(scheduler._registry.handles()) == 2
|
|
assert not finish_started.is_set()
|
|
|
|
await scheduler.stop_async()
|
|
|
|
assert cancelled == 1
|
|
assert scheduler._registry.handles() == ()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_stale_progress_cannot_update_replaced_job(monkeypatch) -> None:
|
|
"""
|
|
旧 generation 的延迟进度不得写入新注册的同 ID 任务。
|
|
"""
|
|
updates = []
|
|
|
|
class RecordingProgress:
|
|
"""
|
|
在内存中记录调度进度，验证任务状态的更新顺序。
|
|
"""
|
|
def __init__(self, _key: str) -> None:
|
|
pass
|
|
|
|
async def update(self, **kwargs) -> None:
|
|
"""
|
|
记录进度更新参数，供任务状态断言使用。
|
|
"""
|
|
updates.append(kwargs)
|
|
|
|
monkeypatch.setattr(scheduler_progress, "AsyncProgressHelper", RecordingProgress)
|
|
scheduler = _scheduler("generation-progress", lambda: None)
|
|
old_job = scheduler._jobs["generation-progress"]
|
|
callback = scheduler._build_progress_callback(
|
|
"generation-progress",
|
|
old_job,
|
|
)
|
|
scheduler._jobs["generation-progress"] = {
|
|
"name": "新一代",
|
|
"provider_name": "测试",
|
|
"running": True,
|
|
"_generation": 2,
|
|
}
|
|
|
|
callback(value=42, text="旧进度")
|
|
await asyncio.sleep(0)
|
|
await asyncio.sleep(0)
|
|
|
|
assert updates == []
|
|
assert scheduler._registry.handles() == ()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_final_progress_waits_for_pending_update(monkeypatch) -> None:
|
|
"""
|
|
任务终态必须等待已提交的进度回调，避免 running 快照迟到覆盖。
|
|
"""
|
|
update_started = asyncio.Event()
|
|
allow_update = asyncio.Event()
|
|
finished = asyncio.Event()
|
|
writes = []
|
|
|
|
class BlockingProgress(_AsyncProgressStub):
|
|
"""
|
|
把中间进度写停在终态收尾之前。
|
|
"""
|
|
|
|
async def update(self, **_kwargs) -> None:
|
|
"""
|
|
等待测试允许后记录进度写入，验证关闭不会提前结束在途更新。
|
|
"""
|
|
update_started.set()
|
|
await allow_update.wait()
|
|
writes.append("update")
|
|
|
|
async def end(self, **_kwargs) -> None:
|
|
"""
|
|
记录结束写入并通知测试收尾已完成。
|
|
"""
|
|
writes.append("end")
|
|
finished.set()
|
|
|
|
async def job(progress_callback) -> None:
|
|
progress_callback(value=100, text="业务处理完成")
|
|
|
|
_patch_progress(monkeypatch, BlockingProgress)
|
|
scheduler = _scheduler("progress-order", job)
|
|
|
|
assert scheduler.start("progress-order") is True
|
|
await asyncio.wait_for(update_started.wait(), timeout=1)
|
|
await asyncio.sleep(0)
|
|
|
|
assert writes == []
|
|
assert not finished.is_set()
|
|
|
|
allow_update.set()
|
|
await asyncio.wait_for(finished.wait(), timeout=1)
|
|
|
|
assert writes == ["update", "end"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_replaced_job_keeps_active_state_without_stale_progress(monkeypatch) -> None:
|
|
"""
|
|
同 ID 新 generation 显示真实运行态，但不继承旧任务进度详情。
|
|
"""
|
|
detail = {}
|
|
|
|
class RecordingProgress:
|
|
"""
|
|
在内存中记录调度进度，验证任务状态的更新顺序。
|
|
"""
|
|
def __init__(self, _key: str) -> None:
|
|
pass
|
|
|
|
def start(self) -> None:
|
|
"""
|
|
提供无副作用的进度启动入口。
|
|
"""
|
|
pass
|
|
|
|
def update(self, **kwargs) -> None:
|
|
"""
|
|
将进度更新合入共享详情，模拟同步进度存储。
|
|
"""
|
|
detail.update(kwargs)
|
|
|
|
def get(self):
|
|
"""
|
|
返回当前进度详情，供同步收尾路径读取。
|
|
"""
|
|
return detail
|
|
|
|
class RecordingAsyncProgress:
|
|
"""
|
|
提供异步任务使用的进度读取接口，隔离真实进度存储。
|
|
"""
|
|
def __init__(self, _key: str) -> None:
|
|
pass
|
|
|
|
async def get(self):
|
|
"""
|
|
返回当前进度详情，供异步收尾路径读取。
|
|
"""
|
|
return detail
|
|
|
|
monkeypatch.setattr(scheduler_execution, "ProgressHelper", RecordingProgress)
|
|
monkeypatch.setattr(scheduler_progress, "ProgressHelper", RecordingProgress)
|
|
monkeypatch.setattr(
|
|
scheduler_progress,
|
|
"AsyncProgressHelper",
|
|
RecordingAsyncProgress,
|
|
)
|
|
scheduler = _scheduler("generation-cache", lambda: None)
|
|
old_job = scheduler._prepare_job("generation-cache")
|
|
assert old_job is not None
|
|
assert detail["data"]["_generation"] == 1
|
|
|
|
scheduler._jobs["generation-cache"] = {
|
|
"name": "新一代",
|
|
"provider_name": "测试",
|
|
"running": False,
|
|
"_generation": 2,
|
|
}
|
|
|
|
progress = scheduler.get_progress("generation-cache")
|
|
assert progress is not None
|
|
assert progress.status == "running"
|
|
assert progress.enable is True
|
|
assert progress.value == 0
|
|
assert "_generation" not in progress.data
|
|
async_progress = await scheduler.aget_progress("generation-cache")
|
|
assert async_progress is not None
|
|
assert async_progress.status == "running"
|
|
assert async_progress.enable is True
|
|
assert async_progress.value == 0
|
|
assert "_generation" not in async_progress.data
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_stale_generation_cannot_finish_replaced_job(monkeypatch) -> None:
|
|
"""
|
|
旧 generation 收尾不得改写同 ID 的新任务状态或进度。
|
|
"""
|
|
monkeypatch.setattr(scheduler_progress, "AsyncProgressHelper", _AsyncProgressStub)
|
|
scheduler = _scheduler("generation-job", lambda: None)
|
|
old_job = scheduler._jobs["generation-job"]
|
|
old_job["running"] = True
|
|
new_job = {
|
|
"name": "新一代",
|
|
"provider_name": "测试",
|
|
"running": True,
|
|
"_generation": 2,
|
|
}
|
|
scheduler._jobs["generation-job"] = new_job
|
|
|
|
await scheduler._finish_job(
|
|
job_id="generation-job",
|
|
job=old_job,
|
|
generation=1,
|
|
success=True,
|
|
)
|
|
|
|
assert new_job["running"] is True
|
|
assert "last_finished_at" not in new_job
|
|
assert old_job["running"] is True
|
|
|
|
|
|
def test_agent_task_manual_start_has_single_reservation() -> None:
|
|
"""
|
|
并发手动触发同一 AgentTask 时只能有一个调用获得 reservation。
|
|
"""
|
|
scheduler = _scheduler("agent-task-1", lambda: None)
|
|
scheduler._jobs["agent-task-1"].update(
|
|
name="AgentTask",
|
|
owner="agent",
|
|
)
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
results = []
|
|
|
|
def start(*_args, **_kwargs):
|
|
entered.set()
|
|
release.wait(timeout=1)
|
|
return True
|
|
|
|
scheduler.start = start
|
|
|
|
first = threading.Thread(
|
|
target=lambda: results.append(scheduler.start_agent_task(1)),
|
|
)
|
|
first.start()
|
|
assert entered.wait(timeout=1)
|
|
second = scheduler.start_agent_task(1)
|
|
release.set()
|
|
first.join(timeout=1)
|
|
|
|
assert second is False
|
|
assert results == [True]
|
|
assert scheduler._registry.reservation_owner("agent-task-1") is None
|
|
|
|
|
|
def test_scheduler_rejects_new_submission_after_stop() -> None:
|
|
"""
|
|
进入 stopping/stopped 后不得再从旧 scheduler 提交任务。
|
|
"""
|
|
scheduler = _scheduler("stopped-job", lambda: None)
|
|
scheduler._lifecycle_state = "stopping"
|
|
|
|
assert scheduler.start("stopped-job") is False
|
|
assert scheduler._jobs["stopped-job"]["running"] is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_config_reload_does_not_restart_scheduler_during_shutdown() -> None:
|
|
"""
|
|
系统关闭开始后到达的配置事件不得重新打开调度入口。
|
|
"""
|
|
scheduler = _scheduler("shutdown-reload", lambda: None)
|
|
scheduler._lifecycle_state = "stopping"
|
|
scheduler.init = lambda **_kwargs: pytest.fail("关闭阶段不得重新初始化调度器")
|
|
|
|
await scheduler.on_config_changed()
|
|
|
|
assert scheduler._lifecycle_state == "stopping"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_config_reload_clears_wallpaper_cache_before_reinitializing() -> None:
|
|
"""
|
|
壁纸配置热重载应先清理旧缓存，再建立新任务目录。
|
|
"""
|
|
scheduler = _scheduler("reload-wallpaper", lambda: None)
|
|
calls = []
|
|
scheduler._services = SimpleNamespace(
|
|
clear_wallpaper_cache=lambda: calls.append("clear"),
|
|
)
|
|
|
|
def init(**_kwargs) -> None:
|
|
calls.append("init")
|
|
scheduler._lifecycle_state = "running"
|
|
|
|
scheduler.init = init
|
|
|
|
await scheduler.on_config_changed()
|
|
|
|
assert calls == ["clear", "init"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_scheduler_list_keeps_previous_catalog_until_reload_is_ready(monkeypatch) -> None:
|
|
"""
|
|
重载期间返回旧任务快照，新调度器启动后切回实时目录。
|
|
"""
|
|
_patch_progress(monkeypatch)
|
|
scheduler = _scheduler("reload-list", lambda: None)
|
|
shutdown_started = threading.Event()
|
|
shutdown_release = threading.Event()
|
|
|
|
class BlockingScheduler:
|
|
"""
|
|
提供可控关闭的旧调度器。
|
|
"""
|
|
|
|
running = True
|
|
|
|
def __init__(self) -> None:
|
|
"""
|
|
准备一项重载前可见的计划任务。
|
|
"""
|
|
self.jobs = [SimpleNamespace(
|
|
id="reload-list",
|
|
name="生命周期测试",
|
|
next_run_time=None,
|
|
)]
|
|
|
|
def get_jobs(self) -> list:
|
|
"""
|
|
返回当前注册作业。
|
|
"""
|
|
return list(self.jobs)
|
|
|
|
def remove_all_jobs(self) -> None:
|
|
"""
|
|
模拟热重载时清除旧调度器的作业。
|
|
"""
|
|
self.jobs.clear()
|
|
|
|
def shutdown(self) -> None:
|
|
"""
|
|
在测试信号处暂停关闭，暴露重载期间的列表状态。
|
|
"""
|
|
shutdown_started.set()
|
|
shutdown_release.wait(timeout=1)
|
|
|
|
class ReplacementScheduler:
|
|
"""
|
|
提供初始化目录完成后的最小 Scheduler 接口。
|
|
"""
|
|
|
|
running = False
|
|
|
|
def get_jobs(self) -> list:
|
|
"""
|
|
返回新目录尚未启动时的作业。
|
|
"""
|
|
return [SimpleNamespace(
|
|
id="reload-list",
|
|
name="生命周期测试",
|
|
next_run_time=None,
|
|
)]
|
|
|
|
def start(self) -> None:
|
|
"""
|
|
标记新调度器已经启动。
|
|
"""
|
|
self.running = True
|
|
|
|
old_scheduler = BlockingScheduler()
|
|
scheduler._scheduler = old_scheduler
|
|
previous = scheduler.list()
|
|
assert [job.id for job in previous] == ["reload-list"]
|
|
|
|
monkeypatch.setattr(
|
|
scheduler_lifecycle,
|
|
"get_scheduler_runtime_config",
|
|
lambda: SimpleNamespace(dev=False),
|
|
)
|
|
replacement = ReplacementScheduler()
|
|
scheduler._reconcile_agent_task_interruptions = lambda: None
|
|
scheduler._initialize_catalog = lambda _config: setattr(scheduler, "_scheduler", replacement)
|
|
|
|
reload_task = asyncio.create_task(scheduler.on_config_changed())
|
|
assert await asyncio.to_thread(shutdown_started.wait, 1)
|
|
|
|
assert scheduler._lifecycle_state == "reloading"
|
|
assert scheduler._scheduler is None
|
|
assert scheduler.list() == previous
|
|
|
|
shutdown_release.set()
|
|
await asyncio.wait_for(reload_task, timeout=1)
|
|
|
|
assert scheduler._lifecycle_state == "running"
|
|
assert scheduler._reload_schedule_snapshot is None
|
|
assert [job.id for job in scheduler.list()] == ["reload-list"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_concurrent_config_reload_waits_for_old_scheduler_shutdown(
|
|
monkeypatch,
|
|
) -> None:
|
|
"""
|
|
并发配置事件合并为一次重建，旧调度线程池结束前不得启动新实例。
|
|
"""
|
|
shutdown_started = threading.Event()
|
|
shutdown_release = threading.Event()
|
|
|
|
class BlockingScheduler:
|
|
"""
|
|
提供可控关闭的旧调度器。
|
|
"""
|
|
|
|
running = True
|
|
|
|
@staticmethod
|
|
def get_jobs() -> list:
|
|
"""
|
|
该并发用例不注册计划任务。
|
|
"""
|
|
return []
|
|
|
|
@staticmethod
|
|
def remove_all_jobs() -> None:
|
|
"""
|
|
并发用例不注册计划任务，无需清理。
|
|
"""
|
|
pass
|
|
|
|
@staticmethod
|
|
def shutdown() -> None:
|
|
"""
|
|
在测试信号处阻塞，以验证重复重载被合并。
|
|
"""
|
|
shutdown_started.set()
|
|
shutdown_release.wait(timeout=1)
|
|
|
|
scheduler = _scheduler("reload-once", lambda: None)
|
|
scheduler._scheduler = BlockingScheduler()
|
|
init_calls = 0
|
|
|
|
def init(**_kwargs) -> None:
|
|
nonlocal init_calls
|
|
init_calls += 1
|
|
scheduler._lifecycle_state = "running"
|
|
|
|
monkeypatch.setattr(scheduler, "init", init)
|
|
first = asyncio.create_task(scheduler.on_config_changed())
|
|
assert await asyncio.to_thread(shutdown_started.wait, 1)
|
|
|
|
await scheduler.on_config_changed()
|
|
assert init_calls == 0
|
|
assert scheduler._lifecycle_state == "reloading"
|
|
|
|
shutdown_release.set()
|
|
await asyncio.wait_for(first, timeout=1)
|
|
assert init_calls == 1
|
|
assert scheduler._lifecycle_state == "running"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_config_reload_preserves_overlap_guard_across_job_generations(
|
|
monkeypatch,
|
|
) -> None:
|
|
"""
|
|
热重载替换任务定义后，同 ID 旧任务结束前不得启动新 generation。
|
|
"""
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
finished = asyncio.Event()
|
|
run_count = 0
|
|
|
|
async def job() -> None:
|
|
nonlocal run_count
|
|
run_count += 1
|
|
if run_count == 1:
|
|
started.set()
|
|
await release.wait()
|
|
finished.set()
|
|
|
|
_patch_progress(monkeypatch)
|
|
scheduler = _scheduler("reload-overlap", job)
|
|
|
|
class ActiveScheduler:
|
|
"""
|
|
提供列表接口所需的最小 APScheduler 状态。
|
|
"""
|
|
|
|
running = True
|
|
|
|
@staticmethod
|
|
def get_jobs() -> list:
|
|
"""
|
|
当前用例只关注正在运行任务，不提供后续计划。
|
|
"""
|
|
return []
|
|
|
|
def init(**_kwargs) -> None:
|
|
replacement = {
|
|
"name": "生命周期测试",
|
|
"provider_name": "测试",
|
|
"func": job,
|
|
"running": False,
|
|
}
|
|
scheduler._assign_job_generation("reload-overlap", replacement)
|
|
scheduler._jobs = {"reload-overlap": replacement}
|
|
scheduler._scheduler = ActiveScheduler()
|
|
scheduler._lifecycle_state = "running"
|
|
|
|
monkeypatch.setattr(scheduler, "init", init)
|
|
|
|
assert scheduler.start("reload-overlap") is True
|
|
await asyncio.wait_for(started.wait(), timeout=1)
|
|
await scheduler.on_config_changed()
|
|
|
|
progress = scheduler.get_progress("reload-overlap")
|
|
assert progress is not None
|
|
assert progress.status == "running"
|
|
assert progress.enable is True
|
|
listed = scheduler.list()
|
|
assert len(listed) == 1
|
|
assert listed[0].id == "reload-overlap"
|
|
assert listed[0].status == "正在运行"
|
|
assert scheduler.start("reload-overlap") is False
|
|
assert run_count == 1
|
|
assert len(scheduler._registry.handles()) == 1
|
|
|
|
release.set()
|
|
await asyncio.wait_for(finished.wait(), timeout=1)
|
|
|
|
async def wait_until_released() -> None:
|
|
while (
|
|
scheduler._registry.is_active("reload-overlap")
|
|
or scheduler._registry.handles()
|
|
):
|
|
await asyncio.sleep(0)
|
|
|
|
await asyncio.wait_for(wait_until_released(), timeout=1)
|
|
assert scheduler.start("reload-overlap") is True
|
|
|
|
async def wait_until_second_run_finishes() -> None:
|
|
while (
|
|
scheduler._registry.is_active("reload-overlap")
|
|
or scheduler._registry.handles()
|
|
):
|
|
await asyncio.sleep(0)
|
|
|
|
await asyncio.wait_for(wait_until_second_run_finishes(), timeout=1)
|
|
assert run_count == 2
|
|
|
|
|
|
def test_stop_between_prepare_and_submission_releases_active_generation(
|
|
monkeypatch,
|
|
) -> None:
|
|
"""
|
|
关闭插入准备与提交之间时，不得遗留未实际运行的 generation。
|
|
"""
|
|
calls = 0
|
|
scheduler = _scheduler("stop-race", None)
|
|
|
|
async def job() -> None:
|
|
nonlocal calls
|
|
calls += 1
|
|
|
|
scheduler._jobs["stop-race"]["func"] = job
|
|
original_prepare = scheduler._prepare_job
|
|
|
|
def prepare_then_stop(job_id: str):
|
|
prepared = original_prepare(job_id)
|
|
scheduler._begin_stop()
|
|
return prepared
|
|
|
|
monkeypatch.setattr(scheduler, "_prepare_job", prepare_then_stop)
|
|
|
|
assert scheduler.start("stop-race") is False
|
|
assert calls == 0
|
|
assert scheduler._registry.handles() == ()
|
|
assert scheduler._registry.is_active("stop-race") is False
|
|
assert scheduler._jobs["stop-race"]["running"] is False
|
|
assert scheduler._jobs["stop-race"]["last_error"] == "任务未提交"
|
|
|
|
monkeypatch.setattr(scheduler, "_prepare_job", original_prepare)
|
|
scheduler._lifecycle_state = "running"
|
|
assert scheduler.start("stop-race") is True
|
|
assert calls == 1
|
|
assert scheduler._registry.is_active("stop-race") is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_cross_thread_rejection_closes_unstarted_business_coroutine(
|
|
monkeypatch,
|
|
) -> None:
|
|
"""
|
|
跨线程提交被关闭门禁拒绝时，包装与业务协程都必须释放。
|
|
"""
|
|
main_loop = asyncio.get_running_loop()
|
|
prepared = threading.Event()
|
|
release = threading.Event()
|
|
calls = 0
|
|
scheduler = _scheduler("cross-thread-stop-race", None)
|
|
|
|
async def job() -> None:
|
|
nonlocal calls
|
|
calls += 1
|
|
|
|
scheduler._jobs["cross-thread-stop-race"]["func"] = job
|
|
_patch_main_loop(monkeypatch, main_loop)
|
|
original_prepare = scheduler._prepare_job
|
|
|
|
def prepare_then_wait(job_id: str):
|
|
result = original_prepare(job_id)
|
|
prepared.set()
|
|
release.wait(timeout=1)
|
|
return result
|
|
|
|
monkeypatch.setattr(scheduler, "_prepare_job", prepare_then_wait)
|
|
|
|
with warnings.catch_warnings(record=True) as captured:
|
|
warnings.simplefilter("always", RuntimeWarning)
|
|
start_task = asyncio.create_task(
|
|
asyncio.to_thread(scheduler.start, "cross-thread-stop-race")
|
|
)
|
|
assert await asyncio.to_thread(prepared.wait, 1)
|
|
scheduler._begin_stop()
|
|
release.set()
|
|
assert await asyncio.wait_for(start_task, timeout=1) is False
|
|
gc.collect()
|
|
|
|
assert calls == 0
|
|
assert scheduler._registry.is_active("cross-thread-stop-race") is False
|
|
assert scheduler._registry.handles() == ()
|
|
assert not any("was never awaited" in str(item.message) for item in captured)
|
|
|
|
|
|
def test_cancelled_cross_thread_proxy_waits_for_target_loop_cleanup(
|
|
monkeypatch,
|
|
) -> None:
|
|
"""
|
|
跨线程代理提前取消后，真实完成信号必须等待目标循环清理。
|
|
"""
|
|
target_loop = asyncio.new_event_loop()
|
|
loop_blocked = threading.Event()
|
|
loop_release = threading.Event()
|
|
loop_drained = threading.Event()
|
|
loop_errors = []
|
|
loop_thread = threading.Thread(target=target_loop.run_forever)
|
|
loop_thread.start()
|
|
|
|
def block_target_loop() -> None:
|
|
loop_blocked.set()
|
|
loop_release.wait(timeout=1)
|
|
|
|
target_loop.set_exception_handler(
|
|
lambda _loop, context: loop_errors.append(context)
|
|
)
|
|
target_loop.call_soon_threadsafe(block_target_loop)
|
|
assert loop_blocked.wait(timeout=1)
|
|
|
|
scheduler = _scheduler("cancel-before-start", None)
|
|
calls = 0
|
|
|
|
async def business() -> None:
|
|
nonlocal calls
|
|
calls += 1
|
|
|
|
scheduler._jobs["cancel-before-start"]["func"] = business
|
|
_patch_main_loop(monkeypatch, target_loop)
|
|
|
|
with warnings.catch_warnings(record=True) as captured:
|
|
warnings.simplefilter("always", RuntimeWarning)
|
|
try:
|
|
assert scheduler.start("cancel-before-start") is True
|
|
scheduler_handle = scheduler._registry.handles()[0]
|
|
scheduler._cancel_handle(scheduler_handle)
|
|
assert not scheduler_handle.completion.done()
|
|
|
|
loop_release.set()
|
|
target_loop.call_soon_threadsafe(loop_drained.set)
|
|
assert loop_drained.wait(timeout=1)
|
|
assert scheduler_handle.completion.done()
|
|
gc.collect()
|
|
finally:
|
|
target_loop.call_soon_threadsafe(target_loop.stop)
|
|
loop_thread.join(timeout=1)
|
|
target_loop.close()
|
|
|
|
assert calls == 0
|
|
assert loop_errors == []
|
|
assert scheduler._registry.is_active("cancel-before-start") is False
|
|
assert scheduler._registry.handles() == ()
|
|
assert not any("was never awaited" in str(item.message) for item in captured)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_subscription_queue_poll_returns_while_owned_search_is_running(monkeypatch) -> None:
|
|
"""
|
|
长搜索不占用轮询调用，重复轮询不重入且取消必须等待同步 worker 结束。
|
|
"""
|
|
from app.chain.subscribe.facade import SubscribeChain
|
|
|
|
_patch_progress(monkeypatch)
|
|
chain = object.__new__(SubscribeChain)
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
calls = []
|
|
|
|
def consume(**kwargs):
|
|
"""
|
|
模拟慢搜索并捕获手工搜索参数，禁止真实业务调用。
|
|
"""
|
|
calls.append(kwargs)
|
|
started.set()
|
|
assert release.wait(timeout=5)
|
|
|
|
chain.resume_search_queue = consume
|
|
job_id = "subscribe_search_queue"
|
|
scheduler = _scheduler(job_id, chain.async_resume_search_queue)
|
|
scheduler._jobs[job_id]["kwargs"] = {"limit": 2, "manual_sids": (1, 2)}
|
|
scheduler._poll_subscription_search_queue()
|
|
try:
|
|
assert await asyncio.to_thread(started.wait, 2)
|
|
assert scheduler._is_job_active(job_id)
|
|
for _ in range(3):
|
|
scheduler._poll_subscription_search_queue()
|
|
assert len(calls) == 1
|
|
assert calls[0]["limit"] == 2
|
|
assert calls[0]["manual_sids"] == (1, 2)
|
|
handles = scheduler._registry.handles()
|
|
assert handles
|
|
stopping = asyncio.create_task(scheduler.stop_async())
|
|
await asyncio.sleep(0)
|
|
assert not stopping.done()
|
|
assert scheduler._is_job_active(job_id)
|
|
finally:
|
|
release.set()
|
|
await asyncio.wait_for(stopping, timeout=3)
|
|
assert not scheduler._is_job_active(job_id)
|
|
assert scheduler._registry.handles() == ()
|