mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 19:37:54 +08:00
test: stabilize coverage ratchet paths
This commit is contained in:
@@ -287,6 +287,7 @@ def test_only_declared_tables_are_created(plugin_data_root, sqlite_backend):
|
||||
def test_sqlite_database_lands_in_the_plugin_data_directory(plugin_data_root, sqlite_backend):
|
||||
"""SQLite 库文件落在插件数据目录下,句柄独占引擎,不带 schema。"""
|
||||
handle = registry_module.get_database("demo")
|
||||
assert registry_module.database_handles() == (handle,)
|
||||
assert handle.db_path == plugin_data_root / "demo" / "plugin.db"
|
||||
assert handle.owns_engine is True
|
||||
assert handle.schema is None
|
||||
|
||||
@@ -7,6 +7,7 @@ from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
import app.startup.initializers.modules as modules_initializer
|
||||
from app.foundation.singleton import Singleton
|
||||
from app.runtime.event.binding import (
|
||||
EventBindingResolver,
|
||||
@@ -188,6 +189,7 @@ def test_unloaded_module_decorator_wrapped_method_is_skipped() -> None:
|
||||
|
||||
def test_free_function_handler_still_invoked_directly() -> None:
|
||||
"""自由函数处理器不属于类声明,保持直调路径不被新跳过逻辑影响。"""
|
||||
assert EventRegistry.handler_class_identifier(_free_function_handler) is None
|
||||
binding = EventBindingResolver(
|
||||
lock=threading.Lock(),
|
||||
resolvers=lambda: {},
|
||||
@@ -324,6 +326,17 @@ def test_all_decorated_host_handler_classes_have_explicit_factories() -> None:
|
||||
}
|
||||
|
||||
|
||||
def test_plugin_classification_fields_skip_unmaterialized_manager(monkeypatch) -> None:
|
||||
"""插件管理器尚未创建时,分类字段 provider 应返回空快照。"""
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"get_existing_plugin_manager",
|
||||
lambda: None,
|
||||
)
|
||||
|
||||
assert modules_initializer._plugin_classification_fields() == ()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("backend_type", "redis_owners"),
|
||||
[
|
||||
|
||||
@@ -217,6 +217,20 @@ def test_fatal_error_stops_reconnect(monkeypatch):
|
||||
assert attempts == [1]
|
||||
|
||||
|
||||
def test_stop_requested_during_connection_error_stops_without_reconnect(monkeypatch):
|
||||
"""连接异常发生后若已收到停止请求,run() 应立即退出而不再重连。"""
|
||||
conn = _connection()
|
||||
|
||||
def fail_after_stop():
|
||||
"""模拟底层连接先收到停止请求、随后抛出断连异常。"""
|
||||
conn._stop_event.set()
|
||||
raise ConnectionError("peer closed")
|
||||
|
||||
monkeypatch.setattr(conn, "_run_once", fail_after_stop)
|
||||
|
||||
conn.run()
|
||||
|
||||
|
||||
def test_reconnect_count_limit(monkeypatch):
|
||||
"""服务端限定重连次数时,连续失败超过上限后停止。"""
|
||||
attempts = []
|
||||
|
||||
@@ -643,6 +643,20 @@ def test_openai_compatible_patch_skips_when_reasoning_content_missing() -> None:
|
||||
assert "reasoning_content" not in payload["messages"][1]
|
||||
|
||||
|
||||
def test_interleaved_reasoning_patch_skips_model_without_payload_method() -> None:
|
||||
"""模型没有请求负载方法时应记录并跳过补丁,而不是构造临时兼容实现。"""
|
||||
|
||||
class _ModelWithoutPayload:
|
||||
"""没有请求负载方法的最小模型替身。"""
|
||||
|
||||
llm_module._patch_interleaved_reasoning_request_support(
|
||||
_ModelWithoutPayload,
|
||||
patch_marker="_interleaved_reasoning_test_patch",
|
||||
)
|
||||
|
||||
assert not hasattr(_ModelWithoutPayload, "_interleaved_reasoning_test_patch")
|
||||
|
||||
|
||||
def test_get_llm_uses_deepseek_thinking_level_controls() -> None:
|
||||
calls = []
|
||||
patch_calls = []
|
||||
|
||||
@@ -877,7 +877,7 @@ assert configured_modules.isdisjoint(sys.modules)
|
||||
|
||||
|
||||
def test_event_resolver_uses_exact_class_and_blocks_stopped_owner_fallback(
|
||||
module_manager_harness,
|
||||
module_manager_harness, monkeypatch,
|
||||
) -> None:
|
||||
"""同名 class 不能冒充 owner;已停止 owner 必须返回 Binding(None)。"""
|
||||
manager = module_manager_harness.manager
|
||||
@@ -892,6 +892,13 @@ def test_event_resolver_uses_exact_class_and_blocks_stopped_owner_fallback(
|
||||
owner_name="Sample",
|
||||
)
|
||||
|
||||
manager._modules.pop("SampleModule")
|
||||
snapshot = Mock()
|
||||
monkeypatch.setattr(manager._runtime, "snapshot", snapshot)
|
||||
materialized_binding = manager.resolve_event_handler_instance(module_class)
|
||||
assert materialized_binding == active_binding
|
||||
snapshot.assert_called_once_with("SampleModule")
|
||||
|
||||
impostor = type("SampleModule", (), {})
|
||||
impostor.__module__ = module_class.__module__
|
||||
assert manager.resolve_event_handler_instance(impostor) is None
|
||||
|
||||
@@ -99,6 +99,7 @@ def test_loader_attributes_module_import_to_installed_plugin_id(tmp_path, monkey
|
||||
for name in ("pureplugin", "nativeplugin"):
|
||||
(tmp_path / name).mkdir()
|
||||
(tmp_path / name / "__init__.py").write_text("", encoding="utf-8")
|
||||
(tmp_path / "_ignored").mkdir()
|
||||
|
||||
def import_module(module_name: str) -> ModuleType:
|
||||
if module_name == "app.plugins.nativeplugin":
|
||||
|
||||
@@ -3,6 +3,8 @@
|
||||
import importlib
|
||||
import inspect
|
||||
|
||||
import pytest
|
||||
|
||||
from app.scheduler import Scheduler, SchedulerChain
|
||||
from app.scheduler.facade import Scheduler as ConcreteScheduler
|
||||
|
||||
@@ -17,6 +19,12 @@ def test_scheduler_package_root_keeps_legacy_type_identity() -> None:
|
||||
assert SchedulerChain.__module__ == "app.scheduler"
|
||||
|
||||
|
||||
def test_scheduler_package_root_rejects_unknown_attribute() -> None:
|
||||
"""旧包根只暴露声明的兼容入口,未知属性应保持标准 AttributeError。"""
|
||||
with pytest.raises(AttributeError, match="missing_scheduler_entrypoint"):
|
||||
getattr(importlib.import_module("app.scheduler"), "missing_scheduler_entrypoint")
|
||||
|
||||
|
||||
def test_scheduler_legacy_methods_keep_plugin_call_shapes() -> None:
|
||||
"""旧插件使用的四个方法继续保留原调用参数形态。"""
|
||||
start = inspect.signature(Scheduler.start)
|
||||
|
||||
Reference in New Issue
Block a user