mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-06-10 11:57:07 +08:00
243 lines
8.8 KiB
Python
243 lines
8.8 KiB
Python
import asyncio
|
||
import io
|
||
import inspect
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
from fastapi import HTTPException
|
||
from starlette.requests import Request
|
||
from starlette.responses import Response
|
||
|
||
from app.api.endpoints import login as login_endpoint
|
||
from app.api.endpoints import plugin as plugin_endpoint
|
||
from app.api.endpoints import system as system_endpoint
|
||
from app.api.endpoints import user as user_endpoint
|
||
from app.api.endpoints import dashboard as dashboard_endpoint
|
||
from app.core.security import verify_resource_token
|
||
from app.db.user_oper import (
|
||
get_current_active_superuser,
|
||
get_current_active_superuser_async,
|
||
get_current_active_user_async,
|
||
)
|
||
from app.schemas.types import SystemConfigKey
|
||
|
||
|
||
def _dependency_of(func, parameter_name: str):
|
||
"""读取 FastAPI 函数参数上声明的依赖函数。"""
|
||
return inspect.signature(func).parameters[parameter_name].default.dependency
|
||
|
||
|
||
def _build_request() -> Request:
|
||
"""构造最小测试请求。"""
|
||
return Request(
|
||
{
|
||
"type": "http",
|
||
"method": "POST",
|
||
"path": "/api/v1/login/access-token",
|
||
"headers": [(b"host", b"testserver")],
|
||
"scheme": "http",
|
||
"server": ("testserver", 80),
|
||
"client": ("testclient", 123),
|
||
}
|
||
)
|
||
|
||
|
||
def test_system_sensitive_read_endpoints_require_superuser():
|
||
"""系统敏感读取接口必须只允许管理员访问。"""
|
||
assert _dependency_of(system_endpoint.get_env_setting, "_") is get_current_active_superuser_async
|
||
assert _dependency_of(system_endpoint.get_setting, "_") is get_current_active_superuser_async
|
||
|
||
|
||
def test_system_public_read_endpoints_require_active_user():
|
||
"""公开读取接口只要求登录且启用的用户。"""
|
||
assert _dependency_of(system_endpoint.ping, "_") is get_current_active_user_async
|
||
assert _dependency_of(system_endpoint.get_public_setting, "_") is get_current_active_user_async
|
||
|
||
|
||
def test_dashboard_endpoints_require_superuser():
|
||
"""仪表板页面相关接口必须只允许管理员访问。"""
|
||
assert _dependency_of(dashboard_endpoint.statistic, "_") is get_current_active_superuser
|
||
assert _dependency_of(dashboard_endpoint.storage, "_") is get_current_active_superuser
|
||
assert _dependency_of(dashboard_endpoint.processes, "_") is get_current_active_superuser
|
||
assert _dependency_of(dashboard_endpoint.downloader, "_") is get_current_active_superuser
|
||
assert _dependency_of(dashboard_endpoint.schedule, "_") is get_current_active_superuser
|
||
assert _dependency_of(dashboard_endpoint.transfer, "_") is get_current_active_superuser
|
||
assert _dependency_of(dashboard_endpoint.cpu, "_") is get_current_active_superuser
|
||
assert _dependency_of(dashboard_endpoint.memory, "_") is get_current_active_superuser
|
||
assert _dependency_of(dashboard_endpoint.network, "_") is get_current_active_superuser
|
||
|
||
|
||
def test_plugin_dashboard_endpoints_require_superuser():
|
||
"""插件仪表板接口必须只允许管理员访问。"""
|
||
assert _dependency_of(plugin_endpoint.plugin_dashboard_meta, "_") is get_current_active_superuser
|
||
assert _dependency_of(plugin_endpoint.plugin_dashboard_by_key, "_") is get_current_active_superuser
|
||
assert _dependency_of(plugin_endpoint.plugin_dashboard, "_") is get_current_active_superuser
|
||
|
||
|
||
def test_system_public_setting_allows_only_non_sensitive_keys(monkeypatch):
|
||
"""公开系统设置接口只能读取明确列入白名单的非敏感配置。"""
|
||
calls = []
|
||
|
||
class FakeSystemConfigOper:
|
||
"""返回测试配置值的系统配置桩。"""
|
||
|
||
def get(self, key):
|
||
"""返回测试配置值。"""
|
||
calls.append(key)
|
||
return [{"path": "/downloads"}]
|
||
|
||
monkeypatch.setattr(system_endpoint, "SystemConfigOper", FakeSystemConfigOper)
|
||
|
||
response = asyncio.run(
|
||
system_endpoint.get_public_setting(SystemConfigKey.Directories.value)
|
||
)
|
||
|
||
assert response.success is True
|
||
assert response.data == {"value": [{"path": "/downloads"}]}
|
||
assert calls == [SystemConfigKey.Directories]
|
||
|
||
response = asyncio.run(system_endpoint.get_public_setting("PLUGIN_MARKET"))
|
||
|
||
assert response.success is True
|
||
assert response.data == {"value": system_endpoint.settings.PLUGIN_MARKET}
|
||
assert calls == [SystemConfigKey.Directories]
|
||
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
asyncio.run(system_endpoint.get_public_setting("API_TOKEN"))
|
||
|
||
assert exc_info.value.status_code == 404
|
||
assert exc_info.value.detail == "配置项不存在"
|
||
|
||
|
||
def test_system_ping_returns_success():
|
||
"""服务存活检测接口返回标准成功响应。"""
|
||
response = asyncio.run(system_endpoint.ping())
|
||
|
||
assert response.success is True
|
||
|
||
|
||
def test_login_sets_resource_token_cookie(monkeypatch):
|
||
"""登录成功时应立即写入资源 Cookie，避免插件静态文件抢先加载失败。"""
|
||
|
||
class FakeUserChain:
|
||
"""返回登录成功用户的用户链桩。"""
|
||
|
||
def user_authenticate(self, username, password, mfa_code=None):
|
||
"""返回认证成功结果。"""
|
||
return True, SimpleNamespace(
|
||
id=1,
|
||
name=username,
|
||
is_superuser=False,
|
||
avatar="",
|
||
permissions={"discovery": True},
|
||
)
|
||
|
||
class FakeSystemConfigOper:
|
||
"""返回已完成向导状态的系统配置桩。"""
|
||
|
||
def get(self, key):
|
||
"""返回测试配置值。"""
|
||
return "1"
|
||
|
||
form_data = SimpleNamespace(username="user", password="password")
|
||
request = _build_request()
|
||
response = Response()
|
||
|
||
monkeypatch.setattr(login_endpoint, "UserChain", FakeUserChain)
|
||
monkeypatch.setattr(login_endpoint, "SystemConfigOper", FakeSystemConfigOper)
|
||
|
||
token = login_endpoint.login_access_token(
|
||
request=request,
|
||
response=response,
|
||
form_data=form_data,
|
||
)
|
||
|
||
assert token.user_id == 1
|
||
assert token.permissions == {"discovery": True}
|
||
assert "set-cookie" in response.headers
|
||
|
||
resource_cookie = response.headers["set-cookie"].split("=", 1)[1].split(";", 1)[0]
|
||
payload = verify_resource_token(resource_cookie)
|
||
assert payload.sub == 1
|
||
assert payload.username == "user"
|
||
assert payload.purpose == "resource"
|
||
|
||
|
||
def test_plugin_static_file_requires_resource_token_by_default(monkeypatch):
|
||
"""普通插件静态资源必须校验资源令牌。"""
|
||
calls = []
|
||
|
||
class FakePluginManager:
|
||
"""返回空认证提供方的插件管理器桩。"""
|
||
|
||
def get_plugin_auth_providers(self):
|
||
"""返回插件认证入口列表。"""
|
||
return []
|
||
|
||
monkeypatch.setattr(plugin_endpoint, "PluginManager", FakePluginManager)
|
||
monkeypatch.setattr(plugin_endpoint, "verify_resource_token", lambda token: calls.append(token))
|
||
|
||
plugin_endpoint._verify_plugin_static_file_access(
|
||
plugin_id="DemoPlugin",
|
||
filepath="dist/remoteEntry.js",
|
||
resource_token="resource-token",
|
||
)
|
||
|
||
assert calls == ["resource-token"]
|
||
|
||
|
||
def test_plugin_auth_remote_files_allow_anonymous_bootstrap(monkeypatch):
|
||
"""插件登录认证远程组件需要允许登录前匿名加载。"""
|
||
calls = []
|
||
|
||
class FakePluginManager:
|
||
"""返回认证插件 remote 信息的插件管理器桩。"""
|
||
|
||
def get_plugin_auth_providers(self):
|
||
"""返回插件认证入口列表。"""
|
||
return [
|
||
{
|
||
"remote": {
|
||
"id": "AuthPlugin",
|
||
"url": "/plugin/file/AuthPlugin/dist/remoteEntry.js",
|
||
}
|
||
}
|
||
]
|
||
|
||
monkeypatch.setattr(plugin_endpoint, "PluginManager", FakePluginManager)
|
||
monkeypatch.setattr(plugin_endpoint, "verify_resource_token", lambda token: calls.append(token))
|
||
|
||
plugin_endpoint._verify_plugin_static_file_access(
|
||
plugin_id="AuthPlugin",
|
||
filepath="dist/remoteEntry.js",
|
||
)
|
||
plugin_endpoint._verify_plugin_static_file_access(
|
||
plugin_id="AuthPlugin",
|
||
filepath="dist/assets/chunk.js",
|
||
)
|
||
plugin_endpoint._verify_plugin_static_file_access(
|
||
plugin_id="authplugin",
|
||
filepath="dist/assets/chunk.js",
|
||
)
|
||
|
||
assert calls == []
|
||
|
||
|
||
def test_upload_avatar_rejects_other_user_for_non_superuser():
|
||
"""普通用户不能通过 user_id 参数修改其他用户头像。"""
|
||
current_user = SimpleNamespace(id=1, is_superuser=False)
|
||
upload_file = SimpleNamespace(file=io.BytesIO(b"avatar"), filename="avatar.png")
|
||
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
asyncio.run(
|
||
user_endpoint.upload_avatar(
|
||
user_id=2,
|
||
db=object(),
|
||
file=upload_file,
|
||
current_user=current_user,
|
||
)
|
||
)
|
||
|
||
assert exc_info.value.status_code == 400
|
||
assert exc_info.value.detail == "用户权限不足"
|