mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-05-08 14:23:27 +08:00
239 lines
8.3 KiB
Python
239 lines
8.3 KiB
Python
import importlib.util
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import call, patch
|
|
|
|
|
|
def _load_jellyfin_module():
|
|
module_name = "_test_jellyfin_module"
|
|
app_module = types.ModuleType("app")
|
|
core_module = types.ModuleType("app.core")
|
|
utils_module = types.ModuleType("app.utils")
|
|
log_module = types.ModuleType("app.log")
|
|
config_module = types.ModuleType("app.core.config")
|
|
schemas_module = types.ModuleType("app.schemas")
|
|
http_module = types.ModuleType("app.utils.http")
|
|
url_module = types.ModuleType("app.utils.url")
|
|
|
|
class _Logger:
|
|
def info(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
def warning(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
def error(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
def debug(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
class _RequestUtils:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def get_res(self, *args, **kwargs):
|
|
return None
|
|
|
|
class _UrlUtils:
|
|
@staticmethod
|
|
def standardize_base_url(host):
|
|
if not host:
|
|
return host
|
|
if not host.endswith("/"):
|
|
host += "/"
|
|
if not host.startswith("http://") and not host.startswith("https://"):
|
|
host = "http://" + host
|
|
return host
|
|
|
|
@staticmethod
|
|
def combine_url(host, path=None, query=None):
|
|
from urllib.parse import urljoin
|
|
|
|
if path is None:
|
|
path = "/"
|
|
host = _UrlUtils.standardize_base_url(host)
|
|
return urljoin(host, path)
|
|
|
|
log_module.logger = _Logger()
|
|
config_module.settings = types.SimpleNamespace(SUPERUSER="admin", USER_AGENT="MoviePilot")
|
|
schemas_module.MediaType = types.SimpleNamespace(MOVIE=types.SimpleNamespace(value="movie"))
|
|
schemas_module.MediaServerItem = object
|
|
schemas_module.MediaServerLibrary = object
|
|
schemas_module.Statistic = object
|
|
schemas_module.WebhookEventInfo = object
|
|
schemas_module.MediaServerItemUserState = object
|
|
schemas_module.MediaServerPlayItem = object
|
|
http_module.RequestUtils = _RequestUtils
|
|
url_module.UrlUtils = _UrlUtils
|
|
|
|
app_module.schemas = schemas_module
|
|
app_module.log = log_module
|
|
app_module.core = core_module
|
|
app_module.utils = utils_module
|
|
core_module.config = config_module
|
|
utils_module.http = http_module
|
|
utils_module.url = url_module
|
|
|
|
stub_modules = {
|
|
"app": app_module,
|
|
"app.log": log_module,
|
|
"app.core": core_module,
|
|
"app.core.config": config_module,
|
|
"app.schemas": schemas_module,
|
|
"app.utils": utils_module,
|
|
"app.utils.http": http_module,
|
|
"app.utils.url": url_module,
|
|
}
|
|
for stub_module in stub_modules.values():
|
|
stub_module._jellyfin_test_stub = True
|
|
|
|
jellyfin_path = Path(__file__).resolve().parents[1] / "app" / "modules" / "jellyfin" / "jellyfin.py"
|
|
spec = importlib.util.spec_from_file_location(module_name, jellyfin_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
with patch.dict(sys.modules, stub_modules):
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
jellyfin_module = _load_jellyfin_module()
|
|
Jellyfin = jellyfin_module.Jellyfin
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, payload: dict):
|
|
self._payload = payload
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
class JellyfinUserResolutionTest(unittest.TestCase):
|
|
def test_loader_does_not_leave_stub_modules_in_sys_modules(self):
|
|
self.assertNotIn("_test_jellyfin_module", sys.modules)
|
|
self.assertFalse(getattr(sys.modules.get("app.log"), "_jellyfin_test_stub", False))
|
|
self.assertFalse(getattr(sys.modules.get("app.core.config"), "_jellyfin_test_stub", False))
|
|
self.assertFalse(getattr(sys.modules.get("app.utils.http"), "_jellyfin_test_stub", False))
|
|
|
|
def _build_client(self) -> Jellyfin:
|
|
client = Jellyfin.__new__(Jellyfin)
|
|
client._host = "http://jellyfin.local:8096"
|
|
client._apikey = "api-key"
|
|
client._playhost = None
|
|
client._sync_libraries = []
|
|
client.user = "fallback-user"
|
|
return client
|
|
|
|
def test_get_user_prefers_exact_username_without_warning(self):
|
|
client = self._build_client()
|
|
payload = [
|
|
{"Id": "admin-id", "Name": "admin", "Policy": {"IsAdministrator": True}},
|
|
{"Id": "alice-id", "Name": "alice", "Policy": {"IsAdministrator": False}},
|
|
]
|
|
|
|
with patch.object(jellyfin_module, "RequestUtils") as request_utils_cls, patch.object(
|
|
jellyfin_module.logger, "warning"
|
|
) as warning_mock:
|
|
request_utils_cls.return_value.get_res.return_value = _FakeResponse(payload)
|
|
|
|
user_id = client.get_user("alice")
|
|
|
|
self.assertEqual(user_id, "alice-id")
|
|
warning_mock.assert_not_called()
|
|
|
|
def test_get_user_prefers_enable_all_folders_admin(self):
|
|
client = self._build_client()
|
|
payload = [
|
|
{
|
|
"Id": "visible-admin-id",
|
|
"Name": "visible",
|
|
"Policy": {"IsAdministrator": True, "EnabledFolders": ["lib-1", "lib-2", "lib-3"]},
|
|
},
|
|
{
|
|
"Id": "full-admin-id",
|
|
"Name": "full",
|
|
"Policy": {"IsAdministrator": True, "EnableAllFolders": True},
|
|
},
|
|
]
|
|
|
|
with patch.object(jellyfin_module, "RequestUtils") as request_utils_cls:
|
|
request_utils_cls.return_value.get_res.return_value = _FakeResponse(payload)
|
|
|
|
user_id = client.get_user()
|
|
|
|
self.assertEqual(user_id, "full-admin-id")
|
|
|
|
def test_get_user_warns_and_prefers_larger_visible_scope_admin(self):
|
|
client = self._build_client()
|
|
payload = [
|
|
{
|
|
"Id": "small-admin-id",
|
|
"Name": "small",
|
|
"Policy": {"IsAdministrator": True, "EnabledFolders": ["lib-1"]},
|
|
},
|
|
{
|
|
"Id": "large-admin-id",
|
|
"Name": "large",
|
|
"Policy": {"IsAdministrator": True, "EnabledFolders": ["lib-1", "lib-2", "lib-3"]},
|
|
},
|
|
{"Id": "user-id", "Name": "normal", "Policy": {"IsAdministrator": False}},
|
|
]
|
|
|
|
with patch.object(jellyfin_module, "RequestUtils") as request_utils_cls, patch.object(
|
|
jellyfin_module.logger, "warning"
|
|
) as warning_mock:
|
|
request_utils_cls.return_value.get_res.return_value = _FakeResponse(payload)
|
|
|
|
user_id = client.get_user("admin")
|
|
|
|
self.assertEqual(user_id, "large-admin-id")
|
|
self.assertGreaterEqual(warning_mock.call_count, 2)
|
|
|
|
warning_messages = [
|
|
call.args[0] for call in warning_mock.call_args_list if call.args and isinstance(call.args[0], str)
|
|
]
|
|
self.assertTrue(any("超级管理员" in message for message in warning_messages))
|
|
self.assertTrue(
|
|
any(
|
|
("部分" in message)
|
|
or ("可见" in message)
|
|
or ("访问范围" in message)
|
|
or ("EnabledFolders" in message)
|
|
for message in warning_messages
|
|
)
|
|
)
|
|
self.assertTrue(any(("回退" in message) or ("fallback" in message.lower()) for message in warning_messages))
|
|
|
|
def test_get_jellyfin_librarys_returns_empty_when_user_missing(self):
|
|
client = self._build_client()
|
|
client.user = None
|
|
|
|
with patch.object(jellyfin_module, "RequestUtils") as request_utils_cls:
|
|
libraries = client._Jellyfin__get_jellyfin_librarys()
|
|
|
|
self.assertEqual(libraries, [])
|
|
request_utils_cls.assert_not_called()
|
|
|
|
def test_get_jellyfin_librarys_uses_normalized_views_url(self):
|
|
client = self._build_client()
|
|
client._host = "http://jellyfin.local:8096"
|
|
client.user = "user-id"
|
|
|
|
with patch.object(jellyfin_module, "RequestUtils") as request_utils_cls:
|
|
request_utils_cls.return_value.get_res.return_value = _FakeResponse({"Items": []})
|
|
|
|
libraries = client._Jellyfin__get_jellyfin_librarys()
|
|
|
|
self.assertEqual(libraries, [])
|
|
request_utils_cls.return_value.get_res.assert_called_once_with(
|
|
"http://jellyfin.local:8096/Users/user-id/Views",
|
|
{"api_key": "api-key"},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|