diff --git a/app/modules/emby/emby.py b/app/modules/emby/emby.py index 3c7f0564..e68424cd 100644 --- a/app/modules/emby/emby.py +++ b/app/modules/emby/emby.py @@ -191,7 +191,12 @@ class Emby: def get_user(self, user_name: Optional[str] = None) -> Optional[Union[str, int]]: """ - 获得管理员用户 + 获取用于查询用户范围数据的用户ID + + 优先匹配指定用户名,其次匹配媒体服务器配置用户名,最后回退管理员。 + + :param user_name: 优先匹配的用户名 + :return: 匹配到的用户ID,未找到可用用户时返回None """ if not self._host or not self._apikey: return None @@ -203,15 +208,18 @@ class Emby: res = RequestUtils().get_res(url, params) if res: users = res.json() - # 先查询是否有与当前用户名称匹配的 - if user_name: - for user in users: - if user.get("Name") == user_name: - return user.get("Id") + candidate_usernames = [] + for candidate_username in (user_name, self._username): + if candidate_username and candidate_username not in candidate_usernames: + candidate_usernames.append(candidate_username) + for candidate_username in candidate_usernames: + for emby_user in users: + if emby_user.get("Name") == candidate_username: + return emby_user.get("Id") # 查询管理员 - for user in users: - if user.get("Policy", {}).get("IsAdministrator"): - return user.get("Id") + for emby_user in users: + if emby_user.get("Policy", {}).get("IsAdministrator"): + return emby_user.get("Id") else: logger.error(f"Users 未获取到返回数据") except Exception as e: diff --git a/tests/test_emby_user_resolution.py b/tests/test_emby_user_resolution.py new file mode 100644 index 00000000..d2e0acd3 --- /dev/null +++ b/tests/test_emby_user_resolution.py @@ -0,0 +1,59 @@ +from unittest.mock import Mock, patch + +from app.modules.emby.emby import Emby + + +def _resolve_user(users: list[dict], requested_username: str, configured_username: str): + emby = Emby.__new__(Emby) + emby._host = "http://emby.local/" + emby._apikey = "test-api-key" + emby._username = configured_username + + response = Mock() + response.json.return_value = users + with patch("app.modules.emby.emby.RequestUtils") as request_utils: + request_utils.return_value.get_res.return_value = response + return emby.get_user(requested_username) + + +def test_get_user_prefers_requested_username(): + """指定用户名存在时应优先使用该用户。""" + result = _resolve_user( + users=[ + {"Id": "requested-id", "Name": "mp-user", "Policy": {}}, + {"Id": "configured-id", "Name": "configured-user", "Policy": {}}, + {"Id": "admin-id", "Name": "admin", "Policy": {"IsAdministrator": True}}, + ], + requested_username="mp-user", + configured_username="configured-user", + ) + + assert result == "requested-id" + + +def test_get_user_falls_back_to_configured_username(): + """指定用户名不存在时应回退媒体服务器配置用户。""" + result = _resolve_user( + users=[ + {"Id": "configured-id", "Name": "configured-user", "Policy": {}}, + {"Id": "admin-id", "Name": "admin", "Policy": {"IsAdministrator": True}}, + ], + requested_username="missing-mp-user", + configured_username="configured-user", + ) + + assert result == "configured-id" + + +def test_get_user_falls_back_to_administrator(): + """指定用户和配置用户均不存在时应回退管理员。""" + result = _resolve_user( + users=[ + {"Id": "regular-id", "Name": "regular", "Policy": {}}, + {"Id": "admin-id", "Name": "admin", "Policy": {"IsAdministrator": True}}, + ], + requested_username="missing-mp-user", + configured_username="missing-configured-user", + ) + + assert result == "admin-id"