mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
"""
|
||
进程 owner 契约和 global_vars 兼容边界测试。
|
||
"""
|
||
|
||
import ast
|
||
import threading
|
||
from pathlib import Path
|
||
|
||
from app.runtime.config import GlobalVar
|
||
from app.runtime.stop import ProcessStopState, runtime_stop_state
|
||
|
||
|
||
def test_process_stop_state_separates_repeatable_workflow_and_one_shot_transfer() -> None:
|
||
"""
|
||
工作流停止可恢复，整理路径停止只消费一次。
|
||
"""
|
||
state = ProcessStopState()
|
||
|
||
state.stop_workflow(7)
|
||
assert state.is_workflow_stopped(7)
|
||
state.resume_workflow(7)
|
||
assert not state.is_workflow_stopped(7)
|
||
|
||
state.stop_transfer("/media/a.mkv")
|
||
assert state.consume_transfer_stop("/media/a.mkv")
|
||
assert not state.consume_transfer_stop("/media/a.mkv")
|
||
|
||
|
||
def test_system_stop_propagates_to_all_cancellation_scopes() -> None:
|
||
"""
|
||
系统停止后所有工作流和整理任务都必须立即观察到停止。
|
||
"""
|
||
state = ProcessStopState()
|
||
|
||
state.stop_system()
|
||
|
||
assert state.is_system_stopped
|
||
assert state.is_workflow_stopped(99)
|
||
assert state.consume_transfer_stop("/unknown")
|
||
|
||
|
||
def test_global_vars_stop_api_delegates_to_runtime_contract(monkeypatch) -> None:
|
||
"""
|
||
旧 global_vars ABI 必须与新的显式停止状态共享同一事实源。
|
||
"""
|
||
event = threading.Event()
|
||
monkeypatch.setattr(runtime_stop_state, "_system_event", event)
|
||
legacy = GlobalVar()
|
||
|
||
legacy.stop_system()
|
||
|
||
assert event.is_set()
|
||
assert legacy.is_system_stopped
|
||
|
||
|
||
def test_host_code_no_longer_reads_stop_state_from_global_vars() -> None:
|
||
"""
|
||
除兼容实现外，宿主不得重新从 global_vars 读取任何停止信号。
|
||
"""
|
||
root = Path(__file__).resolve().parents[1]
|
||
forbidden = (
|
||
"global_vars.is_system_stopped",
|
||
"global_vars.stop_system",
|
||
"global_vars.is_workflow_stopped",
|
||
"global_vars.stop_workflow",
|
||
"global_vars.workflow_resume",
|
||
"global_vars.is_transfer_stopped",
|
||
"global_vars.stop_transfer",
|
||
)
|
||
violations = []
|
||
for path in (root / "app").rglob("*.py"):
|
||
if path == root / "app/runtime/config.py" or "app/plugins" in path.as_posix():
|
||
continue
|
||
content = path.read_text(encoding="utf-8")
|
||
for expression in forbidden:
|
||
if expression in content:
|
||
violations.append(f"{path.relative_to(root)}:{expression}")
|
||
|
||
assert violations == []
|
||
|
||
|
||
def test_host_code_no_longer_imports_global_vars() -> None:
|
||
"""
|
||
除定义与 SDK 兼容出口外，canonical 宿主不得消费 GlobalVar 门面。
|
||
"""
|
||
root = Path(__file__).resolve().parents[1]
|
||
allowed = {
|
||
root / "app/runtime/config.py",
|
||
root / "app/sdk/config.py",
|
||
}
|
||
violations: list[str] = []
|
||
for path in (root / "app").rglob("*.py"):
|
||
if path in allowed or "app/plugins" in path.as_posix():
|
||
continue
|
||
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
||
for node in ast.walk(tree):
|
||
if not isinstance(node, ast.ImportFrom):
|
||
continue
|
||
if node.module != "app.runtime.config":
|
||
continue
|
||
if any(alias.name == "global_vars" for alias in node.names):
|
||
violations.append(str(path.relative_to(root)))
|
||
|
||
assert violations == []
|