mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-15 12:50:20 +08:00
837 lines
30 KiB
Python
837 lines
30 KiB
Python
"""
|
|
验证宿主、SDK 与旧插件共用实例名额，以及启动、关闭和取消的真实时序。
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from contextvars import copy_context
|
|
from types import ModuleType, SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from app.adapters.network import browser
|
|
from app.adapters.system.host import OwnedProcessTree
|
|
from app.agent.terminal.ownership import TerminalScope
|
|
from app.agent.tools.impl.browse_webpage import BrowseWebpageTool
|
|
from app.runtime.browser import BrowserCapacity, BrowserQueueCancelledError, browser_request
|
|
from app.startup.initializers import modules as modules_initializer
|
|
from app.startup.initializers import plugins as plugins_initializer
|
|
|
|
|
|
class _Browser:
|
|
"""
|
|
模拟独立浏览器进程，关闭失败时继续保持连接和名额。
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
"""
|
|
初始化连接事实、断开监听器和可控关闭失败次数。
|
|
"""
|
|
self.closed = False
|
|
self.close_failures = 0
|
|
self.listeners = []
|
|
self.close_threads = []
|
|
|
|
def on(self, event, callback):
|
|
"""
|
|
只登记被测实例需要的进程断开通知。
|
|
"""
|
|
assert event == "disconnected"
|
|
self.listeners.append(callback)
|
|
|
|
def is_connected(self):
|
|
"""
|
|
返回模拟进程是否仍然存活。
|
|
"""
|
|
return not self.closed
|
|
|
|
def disconnect(self):
|
|
"""
|
|
模拟进程退出，向全部持有者发布真实终态。
|
|
"""
|
|
self.closed = True
|
|
for listener in self.listeners:
|
|
listener()
|
|
|
|
def close(self, **_kwargs):
|
|
"""
|
|
仅成功关闭才发布断开事件，支持在原句柄上重试。
|
|
"""
|
|
self.close_threads.append(threading.get_ident())
|
|
if self.close_failures:
|
|
self.close_failures -= 1
|
|
raise RuntimeError("close failed")
|
|
self.disconnect()
|
|
|
|
|
|
class _AsyncBrowser(_Browser):
|
|
"""
|
|
支持暂停关闭过程的异步进程替身。
|
|
"""
|
|
|
|
def __init__(self, provider):
|
|
"""
|
|
复用同步进程状态，并读取测试设置的异步关闭屏障。
|
|
"""
|
|
super().__init__()
|
|
self.provider = provider
|
|
|
|
async def close(self, **kwargs):
|
|
"""
|
|
等到测试放行后才确认进程已退出。
|
|
"""
|
|
if self.provider.close_started is not None:
|
|
self.provider.close_started.set()
|
|
await self.provider.finish_close.wait()
|
|
super().close(**kwargs)
|
|
|
|
|
|
class _WatchTree:
|
|
"""
|
|
可控的专属进程树替身，用于验证监控线程确认退出后才归还名额。
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
"""
|
|
初始化未退出状态。
|
|
"""
|
|
self.terminated = False
|
|
|
|
@property
|
|
def stopped(self) -> bool:
|
|
"""
|
|
返回测试替身是否已经被强制回收。
|
|
"""
|
|
return self.terminated
|
|
|
|
def terminate(self, _timeout: float) -> bool:
|
|
"""
|
|
记录强制回收并报告已经确认退出。
|
|
"""
|
|
self.terminated = True
|
|
return True
|
|
|
|
|
|
@pytest.fixture
|
|
def provider(monkeypatch):
|
|
"""
|
|
用同一组六类 SDK 入口模拟嵌套启动，不下载内核或访问授权服务器。
|
|
"""
|
|
capacity = BrowserCapacity()
|
|
module = ModuleType("cloakbrowser")
|
|
implementation = ModuleType("cloakbrowser.browser")
|
|
state = SimpleNamespace(
|
|
module=module, implementation=implementation, capacity=capacity, created=[],
|
|
settings={"CLOAKBROWSER_MAX_CONCURRENCY": 1, "CLOAKBROWSER_QUEUE_TIMEOUT": 0.5},
|
|
fail_launch=False, fail_context=False, fail_page=False,
|
|
launch_started=None, finish_launch=None, close_started=None, finish_close=None,
|
|
)
|
|
|
|
def make_browser(asynchronous=False):
|
|
"""
|
|
创建一份独立实例，并记录是否出现超出容量的真实存活进程。
|
|
"""
|
|
if state.fail_launch:
|
|
raise RuntimeError("launch failed")
|
|
value = _AsyncBrowser(state) if asynchronous else _Browser()
|
|
state.created.append(value)
|
|
assert sum(not item.closed for item in state.created) <= state.settings["CLOAKBROWSER_MAX_CONCURRENCY"]
|
|
return value
|
|
|
|
def make_context(value):
|
|
"""
|
|
保留 SDK 的上下文关闭会一并关闭浏览器的合同。
|
|
"""
|
|
if state.fail_context:
|
|
raise RuntimeError("context failed")
|
|
context = MagicMock()
|
|
context.browser = value
|
|
context.close = value.close
|
|
context.new_page.return_value.title.return_value = "Ready"
|
|
if state.fail_page:
|
|
context.new_page.side_effect = RuntimeError("page failed")
|
|
return context
|
|
|
|
def launch(*_args, **_kwargs):
|
|
"""
|
|
模拟公开的同步浏览器启动。
|
|
"""
|
|
return make_browser()
|
|
|
|
def launch_context(*args, **kwargs):
|
|
"""
|
|
通过同一子模块的 launch 嵌套启动，不能二次取得名额。
|
|
"""
|
|
return make_context(implementation.launch(*args, **kwargs))
|
|
|
|
def launch_persistent_context(*_args, **_kwargs):
|
|
"""
|
|
模拟直接返回持久化上下文的入口。
|
|
"""
|
|
return make_context(make_browser())
|
|
|
|
async def launch_async(*_args, **_kwargs):
|
|
"""
|
|
可在浏览器创建前暂停，重现调用方取消而供应方仍在启动的情况。
|
|
"""
|
|
if state.launch_started is not None:
|
|
state.launch_started.set()
|
|
await state.finish_launch.wait()
|
|
return make_browser(asynchronous=True)
|
|
|
|
async def launch_context_async(*args, **kwargs):
|
|
"""
|
|
异步上下文内部也经同一组已包装的浏览器入口启动。
|
|
"""
|
|
return make_context(await implementation.launch_async(*args, **kwargs))
|
|
|
|
async def launch_persistent_context_async(*_args, **_kwargs):
|
|
"""
|
|
模拟异步持久化上下文的直接启动入口。
|
|
"""
|
|
return make_context(make_browser(asynchronous=True))
|
|
|
|
for entrypoint in (
|
|
launch, launch_context, launch_persistent_context,
|
|
launch_async, launch_context_async, launch_persistent_context_async,
|
|
):
|
|
setattr(module, entrypoint.__name__, entrypoint)
|
|
setattr(implementation, entrypoint.__name__, entrypoint)
|
|
monkeypatch.setitem(sys.modules, "cloakbrowser", module)
|
|
monkeypatch.setitem(sys.modules, "cloakbrowser.browser", implementation)
|
|
monkeypatch.setattr(browser, "browser_capacity", capacity)
|
|
monkeypatch.setattr(browser, "_require_browser_license", lambda *_args: None)
|
|
monkeypatch.setattr(browser, "get_runtime_setting", lambda key, default=None: state.settings.get(key, default))
|
|
browser.configure_browser_queue()
|
|
yield state
|
|
browser.BrowserSessionHelper.close_all_sessions()
|
|
assert capacity.waiting == 0
|
|
assert capacity.active == 0
|
|
|
|
|
|
def _until(predicate):
|
|
"""
|
|
有界等待一个可观察状态，不用固定休眠判断竞态是否完成。
|
|
"""
|
|
deadline = time.monotonic() + 3
|
|
while not predicate():
|
|
assert time.monotonic() < deadline
|
|
time.sleep(0.001)
|
|
|
|
|
|
async def _until_async(predicate):
|
|
"""
|
|
在保持事件循环可运行的情况下等待明确状态。
|
|
"""
|
|
async with asyncio.timeout(3):
|
|
while not predicate():
|
|
await asyncio.sleep(0.001)
|
|
|
|
|
|
def test_owned_process_tree_rejects_foreign_root():
|
|
"""
|
|
进程树只接受当前宿主直接创建的驱动，避免误杀其它浏览器。
|
|
"""
|
|
with pytest.raises(RuntimeError, match="不能接管"):
|
|
OwnedProcessTree(os.getpid(), "--moviepilot-browser-owner=foreign")
|
|
|
|
|
|
def test_owned_process_tree_kills_marked_reparented_child_without_touching_neighbor():
|
|
"""
|
|
驱动退出后按唯一标记找回后代并结束，旁边的同类进程继续存活。
|
|
"""
|
|
marker = "--moviepilot-browser-owner=test-tree"
|
|
child_code = "import sys,time; time.sleep(60)"
|
|
driver_code = (
|
|
"import subprocess,sys; subprocess.Popen([sys.executable, '-c', sys.argv[1], sys.argv[2]]);"
|
|
)
|
|
neighbor = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
|
|
driver = subprocess.Popen(
|
|
[sys.executable, "-c", driver_code, child_code, marker],
|
|
)
|
|
tree = OwnedProcessTree(driver.pid, marker)
|
|
try:
|
|
assert driver.wait(timeout=3) == 0
|
|
assert tree.terminate(2)
|
|
assert tree.stopped
|
|
assert neighbor.poll() is None
|
|
finally:
|
|
if driver.poll() is None:
|
|
driver.kill()
|
|
if neighbor.poll() is None:
|
|
neighbor.kill()
|
|
neighbor.wait(timeout=3)
|
|
|
|
|
|
def test_browser_watchdog_reclaims_owned_tree_after_operation_cancel(monkeypatch):
|
|
"""
|
|
原生操作取消后由 owner 监控线程回收专属进程，并在确认后释放容量。
|
|
"""
|
|
capacity = BrowserCapacity()
|
|
lease = capacity.acquire(0.1)
|
|
settings = {
|
|
"CLOAKBROWSER_OPERATION_TIMEOUT": 1,
|
|
"CLOAKBROWSER_CLOSE_TIMEOUT": 0.2,
|
|
}
|
|
monkeypatch.setattr(
|
|
browser,
|
|
"get_runtime_setting",
|
|
lambda key, default=None: settings.get(key, default),
|
|
)
|
|
owner = browser._BrowserLaunchOwner(lease)
|
|
tree = _WatchTree()
|
|
owner._tree = tree
|
|
owner.finish()
|
|
cancelled = threading.Event()
|
|
with browser_request(cancelled.is_set):
|
|
owner.begin_operation()
|
|
cancelled.set()
|
|
owner._wake.set()
|
|
assert owner.wait_stopped(2)
|
|
assert tree.terminated
|
|
assert capacity.active == 0
|
|
|
|
|
|
def test_delayed_native_start_is_killed_after_owner_expiry(monkeypatch):
|
|
"""
|
|
启动超时后的迟到驱动会被新建的专属进程树立即结束，不能重新占用名额。
|
|
"""
|
|
capacity = BrowserCapacity()
|
|
lease = capacity.acquire(0.1)
|
|
owner = browser._BrowserLaunchOwner(lease)
|
|
owner._tree = _WatchTree()
|
|
owner._tree.terminated = True
|
|
owner._expired = True
|
|
owner._reason = "启动已超时"
|
|
candidate = _WatchTree()
|
|
monkeypatch.setattr(browser, "OwnedProcessTree", lambda _pid, _marker: candidate)
|
|
connection = MagicMock()
|
|
connection._transport._proc.pid = 1234
|
|
browser_type = SimpleNamespace(_impl_obj=SimpleNamespace(_connection=connection))
|
|
with pytest.raises(browser.BrowserOperationTimeoutError, match="启动已超时"):
|
|
owner.attach(browser_type, {})
|
|
assert candidate.terminated
|
|
assert owner._stopped.is_set()
|
|
assert capacity.active == 0
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["launch", "launch_context", "launch_persistent_context"])
|
|
def test_sync_entrypoints_hold_one_seat_until_closed(provider, name):
|
|
"""
|
|
三种同步入口及内部嵌套均只计一个实例，重复关闭不会扩大容量。
|
|
"""
|
|
value = getattr(provider.module, name)()
|
|
assert provider.capacity.active == 1
|
|
assert len(provider.created) == 1
|
|
assert getattr(provider.module, name) is getattr(provider.implementation, name)
|
|
value.close()
|
|
value.close()
|
|
assert provider.capacity.active == 0
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["launch_async", "launch_context_async", "launch_persistent_context_async"])
|
|
@pytest.mark.asyncio
|
|
async def test_async_entrypoints_hold_one_seat_until_closed(provider, name):
|
|
"""
|
|
异步入口与内部嵌套沿用同一实例名额和关闭合同。
|
|
"""
|
|
value = await getattr(provider.module, name)()
|
|
assert provider.capacity.active == 1
|
|
assert len(provider.created) == 1
|
|
await value.close()
|
|
await value.close()
|
|
assert provider.capacity.active == 0
|
|
|
|
|
|
def test_host_sdk_and_legacy_requests_share_fifo_queue(provider):
|
|
"""
|
|
宿主和直接 SDK 调用按入队顺序启动，持有者开新标签不需要新名额。
|
|
"""
|
|
first = browser.launch_browser_context()
|
|
first.new_page()
|
|
first.new_page()
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
second = executor.submit(browser.launch_browser_context)
|
|
_until(lambda: provider.capacity.waiting == 1)
|
|
third = executor.submit(provider.implementation.launch)
|
|
_until(lambda: provider.capacity.waiting == 2)
|
|
assert len(provider.created) == 1
|
|
first.close()
|
|
second_context = second.result(timeout=3)
|
|
assert not third.done()
|
|
second_context.close()
|
|
third.result(timeout=3).close()
|
|
assert len(provider.created) == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_and_async_requests_share_capacity_without_blocking_loop(provider):
|
|
"""
|
|
异步等待同步持有者释放时，事件循环和其他调用继续运行。
|
|
"""
|
|
first = provider.module.launch()
|
|
waiting = asyncio.create_task(browser.launch_browser_context_async())
|
|
await _until_async(lambda: provider.capacity.waiting == 1)
|
|
assert not waiting.done()
|
|
assert len(provider.created) == 1
|
|
first.close()
|
|
second = await asyncio.wait_for(waiting, 3)
|
|
await second.close()
|
|
|
|
|
|
def test_queue_timeout_does_not_start_or_release_other_browser(provider):
|
|
"""
|
|
排队超时只撤销等待者，持有者和其名额都保持有效。
|
|
"""
|
|
first = provider.module.launch()
|
|
provider.settings["CLOAKBROWSER_QUEUE_TIMEOUT"] = 0.01
|
|
try:
|
|
with pytest.raises(TimeoutError, match="等待浏览器名额超时"):
|
|
provider.module.launch()
|
|
assert provider.capacity.active == 1
|
|
assert provider.capacity.waiting == 0
|
|
assert len(provider.created) == 1
|
|
finally:
|
|
first.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancelled_async_waiter_never_launches_after_capacity_returns(provider):
|
|
"""
|
|
异步等待取消后立即出队，之后归还名额也不能补启动已取消的请求。
|
|
"""
|
|
first = provider.module.launch()
|
|
waiting = asyncio.create_task(provider.module.launch_async())
|
|
await _until_async(lambda: provider.capacity.waiting == 1)
|
|
waiting.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await waiting
|
|
assert provider.capacity.waiting == 0
|
|
assert provider.capacity.active == 1
|
|
first.close()
|
|
await asyncio.sleep(0)
|
|
assert len(provider.created) == 1
|
|
|
|
|
|
def test_cancelled_sync_request_is_removed_before_launch(provider):
|
|
"""
|
|
线程内等待能观察调用取消，无需等占用中的另一个浏览器退出。
|
|
"""
|
|
first = provider.module.launch()
|
|
cancelled = threading.Event()
|
|
|
|
def request():
|
|
"""
|
|
为本次请求绑定可跨线程设置的取消信号。
|
|
"""
|
|
with browser_request(cancelled.is_set):
|
|
provider.module.launch()
|
|
|
|
with ThreadPoolExecutor(max_workers=1) as executor:
|
|
waiting = executor.submit(request)
|
|
_until(lambda: provider.capacity.waiting == 1)
|
|
cancelled.set()
|
|
with pytest.raises(BrowserQueueCancelledError):
|
|
waiting.result(timeout=3)
|
|
assert len(provider.created) == 1
|
|
first.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancelling_inflight_launch_closes_result_before_releasing(provider):
|
|
"""
|
|
启动已经进入 SDK 后取消，必须取得并关闭最终结果，不能留下后台浏览器。
|
|
"""
|
|
provider.launch_started = asyncio.Event()
|
|
provider.finish_launch = asyncio.Event()
|
|
launching = asyncio.create_task(provider.module.launch_context_async())
|
|
await provider.launch_started.wait()
|
|
launching.cancel()
|
|
await asyncio.sleep(0)
|
|
assert not launching.done()
|
|
assert provider.capacity.active == 1
|
|
provider.finish_launch.set()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await launching
|
|
assert len(provider.created) == 1
|
|
assert provider.created[0].closed
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancelling_close_keeps_seat_until_actual_exit(provider):
|
|
"""
|
|
关闭协程取消不能提前释放仍在退出的实例，重复取消也等待真实终态。
|
|
"""
|
|
value = await provider.module.launch_async()
|
|
provider.close_started = asyncio.Event()
|
|
provider.finish_close = asyncio.Event()
|
|
closing = asyncio.create_task(value.close())
|
|
await provider.close_started.wait()
|
|
closing.cancel()
|
|
await asyncio.sleep(0)
|
|
closing.cancel()
|
|
assert provider.capacity.active == 1
|
|
assert not value.closed
|
|
provider.finish_close.set()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await closing
|
|
assert value.closed
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["fail_launch", "fail_context"])
|
|
def test_failed_launch_closes_partial_result_and_returns_seat(provider, failure):
|
|
"""
|
|
SDK 启动或上下文初始化失败时，已创建的底层浏览器也必须关闭。
|
|
"""
|
|
setattr(provider, failure, True)
|
|
with pytest.raises(RuntimeError, match="failed"):
|
|
provider.module.launch_context()
|
|
assert all(value.closed for value in provider.created)
|
|
assert provider.capacity.active == 0
|
|
|
|
|
|
def test_failed_close_retains_capacity_until_retry_succeeds(provider):
|
|
"""
|
|
浏览器关闭失败仍占用名额，只有后续原句柄重试成功才允许新实例。
|
|
"""
|
|
value = provider.module.launch()
|
|
value.close_failures = 1
|
|
with pytest.raises(RuntimeError, match="close failed"):
|
|
value.close()
|
|
assert provider.capacity.active == 1
|
|
value.close()
|
|
assert provider.capacity.active == 0
|
|
provider.module.launch().close()
|
|
|
|
|
|
def test_browser_disconnect_returns_capacity_without_context_close(provider):
|
|
"""
|
|
底层进程已经退出时，即使调用者尚未关闭上下文也能归还名额。
|
|
"""
|
|
context = provider.module.launch_context()
|
|
provider.created[0].disconnect()
|
|
assert provider.capacity.active == 0
|
|
context.close()
|
|
|
|
|
|
def test_closed_agent_owner_does_not_launch_after_wait(provider):
|
|
"""
|
|
Agent 等待过程中所属任务封口后，不得再创建页面或占用新的实例名额。
|
|
"""
|
|
first = provider.module.launch()
|
|
owner = TerminalScope("test", "queued", "conversation")
|
|
helper = browser.BrowserSessionHelper()
|
|
with ThreadPoolExecutor(max_workers=1) as executor:
|
|
pending = executor.submit(helper.with_session, "queued", lambda state: state, owner=owner)
|
|
_until(lambda: provider.capacity.waiting == 1)
|
|
owner.seal()
|
|
with pytest.raises(BrowserQueueCancelledError):
|
|
pending.result(timeout=3)
|
|
assert len(provider.created) == 1
|
|
first.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_waiting_sessions_cannot_starve_current_browser_actions(provider):
|
|
"""
|
|
多个会话等待唯一实例时，持有实例的会话仍可操作；等待取消后不遗留线程。
|
|
"""
|
|
helper = browser.BrowserSessionHelper()
|
|
await helper.run_action_async(
|
|
"holder", lambda: helper.with_session("holder", lambda _state: "opened"),
|
|
)
|
|
pending = [
|
|
asyncio.create_task(helper.run_action_async(
|
|
f"wait-{index}", lambda index=index: helper.with_session(f"wait-{index}", lambda state: state),
|
|
))
|
|
for index in range(3)
|
|
]
|
|
await _until_async(lambda: provider.capacity.waiting == 3)
|
|
try:
|
|
result = await asyncio.wait_for(helper.run_action_async(
|
|
"holder", lambda: helper.with_session("holder", lambda _state: "still usable"),
|
|
), 0.3)
|
|
assert result == "still usable"
|
|
finally:
|
|
for task in pending:
|
|
task.cancel()
|
|
await asyncio.gather(*pending, return_exceptions=True)
|
|
await _until_async(lambda: provider.capacity.waiting == 0)
|
|
await asyncio.to_thread(helper.close_session, "holder")
|
|
assert len(provider.created) == 1
|
|
await _until_async(lambda: not helper._session_executors)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancelled_same_session_waiter_does_not_reclaim_active_browser(provider):
|
|
"""
|
|
同一会话已有动作时，后续等待者取消只撤销自己的排队任务。
|
|
"""
|
|
helper = browser.BrowserSessionHelper()
|
|
started = threading.Event()
|
|
finish = threading.Event()
|
|
|
|
def hold(_state):
|
|
"""
|
|
保持会话线程占用，给第二个调用制造真实的会话内排队。
|
|
"""
|
|
started.set()
|
|
assert finish.wait(3)
|
|
return "held"
|
|
|
|
holder = asyncio.create_task(helper.run_action_async(
|
|
"same-session", lambda: helper.with_session("same-session", hold),
|
|
))
|
|
await _until_async(started.is_set)
|
|
waiter = asyncio.create_task(helper.run_action_async(
|
|
"same-session", lambda: helper.with_session("same-session", lambda _state: "waiter"),
|
|
))
|
|
await asyncio.sleep(0)
|
|
waiter.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await waiter
|
|
assert provider.created[0].closed is False
|
|
finish.set()
|
|
assert await holder == "held"
|
|
await asyncio.to_thread(helper.close_session, "same-session")
|
|
|
|
|
|
def test_failed_page_initialization_and_close_retains_cleanup_owner(provider, monkeypatch):
|
|
"""
|
|
页面初始化和首次关闭同时失败时，仍在原线程保留上下文供 owner 重试。
|
|
"""
|
|
owner = object()
|
|
context = provider.module.launch_context()
|
|
context.new_page.side_effect = RuntimeError("page failed")
|
|
provider.created[0].close_failures = 1
|
|
monkeypatch.setattr(browser.BrowserSessionHelper, "_launch_context", staticmethod(lambda **_kw: context))
|
|
helper = browser.BrowserSessionHelper()
|
|
with pytest.raises(RuntimeError, match="close failed"):
|
|
helper.with_session("broken", lambda state: state, owner=owner)
|
|
assert provider.capacity.active == 1
|
|
assert helper.close_owner(owner)
|
|
assert not helper._failed_closures
|
|
|
|
|
|
def test_missing_key_keeps_keyless_free_binary_path(monkeypatch):
|
|
"""
|
|
回退版本缺少 key 时继续使用旧版免费内核，不触发自动领取流程。
|
|
"""
|
|
from cloakbrowser import license as provider_license
|
|
|
|
monkeypatch.delenv("CLOAKBROWSER_BINARY_PATH", raising=False)
|
|
resolve = MagicMock(return_value=None)
|
|
monkeypatch.setattr(provider_license, "resolve_license_key", resolve)
|
|
browser._require_browser_license()
|
|
resolve.assert_called_once_with(None)
|
|
|
|
|
|
def test_explicit_key_is_preserved_for_sdk_resolution(monkeypatch):
|
|
"""
|
|
直接 SDK 调用传入 key 时继续优先使用其显式参数，不要求另配环境变量。
|
|
"""
|
|
from cloakbrowser import license as provider_license
|
|
|
|
resolve = MagicMock(return_value="test-key")
|
|
monkeypatch.setattr(provider_license, "resolve_license_key", resolve)
|
|
monkeypatch.delenv("CLOAKBROWSER_BINARY_PATH", raising=False)
|
|
browser._require_browser_license("test-key")
|
|
resolve.assert_called_once_with("test-key")
|
|
|
|
|
|
def test_browser_queue_defaults_to_three_instances(monkeypatch):
|
|
"""
|
|
未声明部署配置时，浏览器队列默认允许三个独立实例同时存活。
|
|
"""
|
|
configured = MagicMock()
|
|
monkeypatch.setattr(browser.browser_capacity, "configure", configured)
|
|
monkeypatch.setattr(browser, "get_runtime_setting", lambda _key, default=None: default)
|
|
|
|
assert browser._browser_queue_timeout() == 60
|
|
configured.assert_called_once_with(3)
|
|
|
|
|
|
def test_legacy_positional_key_and_arguments_are_preserved(provider, monkeypatch):
|
|
"""
|
|
旧插件用位置参数传 key 时，授权检查和 SDK 都收到原始值。
|
|
"""
|
|
checked = MagicMock()
|
|
monkeypatch.setattr(browser, "_require_browser_license", checked)
|
|
|
|
def launch(headless=True, license_key=None):
|
|
"""
|
|
模拟保留位置参数合同的 SDK 启动入口。
|
|
"""
|
|
assert headless is False
|
|
assert license_key == "test-key"
|
|
return _Browser()
|
|
|
|
wrapped = browser._queued_browser_launch(launch)
|
|
wrapped(False, "test-key").close()
|
|
checked.assert_called_once_with("test-key")
|
|
|
|
|
|
def test_sdk_copied_context_cannot_reuse_finished_launch_seat(provider):
|
|
"""
|
|
SDK 在启动时保存的上下文不能让后续回调误用早已结束的嵌套启动资格。
|
|
"""
|
|
captured = []
|
|
|
|
def launch():
|
|
"""
|
|
模拟 SDK 长期任务保存启动时上下文，返回正常浏览器句柄。
|
|
"""
|
|
captured.append(copy_context())
|
|
return _Browser()
|
|
|
|
wrapped = browser._queued_browser_launch(launch)
|
|
first = wrapped()
|
|
provider.settings["CLOAKBROWSER_QUEUE_TIMEOUT"] = 0.01
|
|
try:
|
|
with pytest.raises(TimeoutError):
|
|
captured[0].run(provider.module.launch)
|
|
assert provider.capacity.active == 1
|
|
finally:
|
|
first.close()
|
|
|
|
|
|
def test_failed_agent_page_creation_releases_browser(provider):
|
|
"""
|
|
新会话首次创建页面失败时关闭已经启动的上下文，不遗失唯一名额。
|
|
"""
|
|
provider.fail_page = True
|
|
with pytest.raises(RuntimeError, match="page failed"):
|
|
browser.BrowserSessionHelper().with_session("failed-page", lambda state: state)
|
|
assert provider.created[0].closed
|
|
assert provider.capacity.active == 0
|
|
|
|
|
|
def test_agent_failed_close_is_retried_on_original_thread(provider):
|
|
"""
|
|
显式关闭失败保留会话原执行器，任务回收可重试并释放唯一名额。
|
|
"""
|
|
helper = browser.BrowserSessionHelper()
|
|
owner = object()
|
|
thread_id = helper.with_session("retry", lambda _state: threading.get_ident(), owner=owner)
|
|
value = provider.created[0]
|
|
value.close_failures = 2
|
|
with pytest.raises(RuntimeError, match="close failed"):
|
|
helper.close_session("retry", owner)
|
|
assert provider.capacity.active == 1
|
|
assert not helper.close_owner(owner)
|
|
assert helper.close_owner(owner)
|
|
assert not helper._failed_closures
|
|
assert set(value.close_threads) == {thread_id}
|
|
|
|
|
|
def test_global_shutdown_retains_unclosed_browser_for_retry(provider):
|
|
"""
|
|
全局关闭不能把尚未退出的实例当成成功，保留原 owner 后可再次收敛。
|
|
"""
|
|
helper = browser.BrowserSessionHelper()
|
|
helper.with_session("shutdown", lambda state: state)
|
|
provider.created[0].close_failures = 2
|
|
with pytest.raises(RuntimeError, match="部分浏览器尚未关闭"):
|
|
helper.close_all_sessions()
|
|
assert provider.capacity.active == 1
|
|
helper.close_all_sessions()
|
|
assert provider.capacity.active == 0
|
|
|
|
|
|
def test_startup_shutdown_closes_untracked_browser_owner(provider):
|
|
"""
|
|
启动收尾会回收旧插件直接创建、未登记 Agent 会话的浏览器 owner。
|
|
"""
|
|
instance = provider.module.launch()
|
|
instance._moviepilot_browser_owner._tree = _WatchTree()
|
|
modules_initializer.close_browser_sessions()
|
|
assert instance._moviepilot_browser_owner._stopped.is_set()
|
|
assert provider.capacity.active == 0
|
|
|
|
|
|
def test_startup_shutdown_reports_browser_owner_that_did_not_stop(provider, monkeypatch):
|
|
"""
|
|
监控线程未收敛时启动收尾必须报告失败，避免静默释放未退出的资源。
|
|
"""
|
|
monkeypatch.setattr(browser.BrowserSessionHelper, "close_all_sessions", MagicMock())
|
|
monkeypatch.setattr(browser, "close_browser_instances", lambda: False)
|
|
with pytest.raises(RuntimeError, match="监控线程尚未退出"):
|
|
modules_initializer.close_browser_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_tool_cancellation_reaches_waiting_browser_thread(provider):
|
|
"""
|
|
浏览器工具的异步调用取消后，受保护的同步工作线程也会撤销队列请求。
|
|
"""
|
|
first = provider.module.launch()
|
|
tool = BrowseWebpageTool(session_id="cancelled-tool", user_id="test")
|
|
pending = asyncio.create_task(tool._run_browser_action(
|
|
browser_action="goto", url="https://example.com", selector=None, ref=None,
|
|
value=None, script=None, content_type="text", timeout=30, cookies=None,
|
|
user_agent=None, session_key="cancelled-tool", tab_index=None, allow_private_network=False,
|
|
))
|
|
await _until_async(lambda: provider.capacity.waiting == 1)
|
|
pending.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await pending
|
|
await _until_async(lambda: provider.capacity.waiting == 0)
|
|
first.close()
|
|
assert len(provider.created) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("method", ["action", "get_page_source"])
|
|
def test_site_page_close_failure_still_closes_instance(provider, method):
|
|
"""
|
|
站点路径关闭单页失败时也关闭整个实例，防止其它调用一直等不到名额。
|
|
"""
|
|
context = provider.module.launch_context()
|
|
context.new_page.return_value.close.side_effect = RuntimeError("page close failed")
|
|
with pytest.MonkeyPatch.context() as patch:
|
|
# 复用一个已经准备好的替身，保持本用例只验证站点 finally 的关闭次序。
|
|
patch.setattr(
|
|
browser.PlaywrightHelper, "_PlaywrightHelper__launch_cloakbrowser_context",
|
|
staticmethod(lambda **_kw: context),
|
|
)
|
|
helper = browser.PlaywrightHelper()
|
|
if method == "action":
|
|
helper.action("https://example.com", lambda _page: "ok", headless=True)
|
|
else:
|
|
helper.get_page_source("https://example.com", headless=True)
|
|
assert provider.created[0].closed
|
|
|
|
|
|
def test_configured_capacity_allows_more_independent_instances(provider):
|
|
"""
|
|
提高部署容量后可同时持有多个实例；归还仍按各自票据计数。
|
|
"""
|
|
provider.settings["CLOAKBROWSER_MAX_CONCURRENCY"] = 2
|
|
first = provider.module.launch()
|
|
second = provider.module.launch_context()
|
|
assert provider.capacity.active == 2
|
|
first.close()
|
|
assert provider.capacity.active == 1
|
|
second.close()
|
|
|
|
|
|
def test_legacy_plugin_preparation_wraps_before_import(provider, tmp_path, monkeypatch):
|
|
"""
|
|
旧插件导入 SDK 子模块前，加载准备入口已经安装同一套队列包装。
|
|
"""
|
|
root = tmp_path / "legacy"
|
|
root.mkdir()
|
|
(root / "__init__.py").write_text("from cloakbrowser.browser import launch\n", encoding="utf-8")
|
|
monkeypatch.setattr(plugins_initializer, "acquire_managed_resource", lambda *_args, **_kwargs: None)
|
|
original = provider.module.launch
|
|
plugins_initializer._prepare_legacy_plugin_import(plugin_id="Legacy", plugin_dir=root)
|
|
assert provider.module.launch is original
|
|
namespace = {}
|
|
exec((root / "__init__.py").read_text(), namespace)
|
|
namespace["launch"]().close()
|