248 lines
9.2 KiB
Python
248 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
|
|
BACKEND_DIR = Path(__file__).resolve().parents[1]
|
|
os.environ["KEFU_DB_TYPE"] = "sqlite"
|
|
os.environ["KEFU_DATABASE_URL"] = ""
|
|
os.environ["KEFU_DB_PATH"] = str(BACKEND_DIR / "kefu.db")
|
|
if str(BACKEND_DIR) not in sys.path:
|
|
sys.path.insert(0, str(BACKEND_DIR))
|
|
|
|
import main
|
|
|
|
|
|
class _ProfileResult:
|
|
def __init__(self, profile):
|
|
self.profile = profile
|
|
|
|
def scalar_one_or_none(self):
|
|
return self.profile
|
|
|
|
|
|
class DesktopLoginSecUserIdTests(unittest.IsolatedAsyncioTestCase):
|
|
@staticmethod
|
|
def _account(*, cookie_data: str | None = "cookie-json"):
|
|
return SimpleNamespace(
|
|
id=501,
|
|
cookie_data=cookie_data,
|
|
cookie_path=None,
|
|
cookie_updated_at=datetime(2026, 7, 22, 8, 0, 0),
|
|
im_session_data=None,
|
|
)
|
|
|
|
async def test_verified_sec_user_id_returns_credential(self):
|
|
account = self._account()
|
|
profile = SimpleNamespace(
|
|
sec_user_id=" MS4wLjAB-valid ",
|
|
synced_at=account.cookie_updated_at + timedelta(seconds=1),
|
|
)
|
|
db = SimpleNamespace(execute=AsyncMock(return_value=_ProfileResult(profile)))
|
|
response = main.AccountCookieResponse(account_id=501, cookie_data="cookie-json")
|
|
|
|
with (
|
|
patch.object(main, "get_owned_account", new=AsyncMock(return_value=account)),
|
|
patch.object(main, "_build_cookie_response", new=AsyncMock(return_value=response)) as build,
|
|
):
|
|
result = await main.get_desktop_login_credential(
|
|
account_id=501,
|
|
db=db,
|
|
user=SimpleNamespace(id=9, role="operator"),
|
|
)
|
|
|
|
self.assertIs(result, response)
|
|
db.execute.assert_awaited_once()
|
|
build.assert_awaited_once_with(account, "cookie-json")
|
|
|
|
async def test_missing_or_blank_sec_user_id_is_rejected_without_cookie(self):
|
|
for missing_value in (None, "", " "):
|
|
with self.subTest(sec_user_id=missing_value):
|
|
account = self._account()
|
|
profile = SimpleNamespace(
|
|
sec_user_id=missing_value,
|
|
synced_at=account.cookie_updated_at,
|
|
)
|
|
db = SimpleNamespace(
|
|
execute=AsyncMock(return_value=_ProfileResult(profile))
|
|
)
|
|
|
|
with (
|
|
patch.object(main, "get_owned_account", new=AsyncMock(return_value=account)),
|
|
patch.object(main, "_build_cookie_response", new=AsyncMock()) as build,
|
|
):
|
|
with self.assertRaises(main.HTTPException) as caught:
|
|
await main.get_desktop_login_credential(
|
|
account_id=501,
|
|
db=db,
|
|
user=SimpleNamespace(id=9, role="operator"),
|
|
)
|
|
|
|
self.assertEqual(caught.exception.status_code, 409)
|
|
self.assertIn("缺少 sec_user_id", str(caught.exception.detail))
|
|
build.assert_not_awaited()
|
|
|
|
async def test_missing_profile_or_stale_identity_is_rejected(self):
|
|
account = self._account()
|
|
stale_profile = SimpleNamespace(
|
|
sec_user_id="MS4wLjAB-old",
|
|
synced_at=account.cookie_updated_at - timedelta(seconds=1),
|
|
)
|
|
for profile in (None, stale_profile):
|
|
with self.subTest(profile=profile):
|
|
db = SimpleNamespace(
|
|
execute=AsyncMock(return_value=_ProfileResult(profile))
|
|
)
|
|
with (
|
|
patch.object(main, "get_owned_account", new=AsyncMock(return_value=account)),
|
|
patch.object(main, "_build_cookie_response", new=AsyncMock()) as build,
|
|
):
|
|
with self.assertRaises(main.HTTPException) as caught:
|
|
await main.get_desktop_login_credential(
|
|
account_id=501,
|
|
db=db,
|
|
user=SimpleNamespace(id=9, role="operator"),
|
|
)
|
|
self.assertEqual(caught.exception.status_code, 409)
|
|
build.assert_not_awaited()
|
|
|
|
async def test_profile_check_failure_is_unknown_and_fails_closed(self):
|
|
account = self._account()
|
|
db = SimpleNamespace(execute=AsyncMock(side_effect=RuntimeError("db locked")))
|
|
|
|
with (
|
|
patch.object(main, "get_owned_account", new=AsyncMock(return_value=account)),
|
|
patch.object(main, "_build_cookie_response", new=AsyncMock()) as build,
|
|
patch.object(main.logger, "exception") as log_exception,
|
|
):
|
|
with self.assertRaises(main.HTTPException) as caught:
|
|
await main.get_desktop_login_credential(
|
|
account_id=501,
|
|
db=db,
|
|
user=SimpleNamespace(id=9, role="operator"),
|
|
)
|
|
|
|
self.assertEqual(caught.exception.status_code, 503)
|
|
self.assertNotIn("缺少 sec_user_id", str(caught.exception.detail))
|
|
build.assert_not_awaited()
|
|
log_exception.assert_called_once()
|
|
|
|
async def test_no_cookie_keeps_existing_no_login_state_flow(self):
|
|
account = self._account(cookie_data=None)
|
|
db = SimpleNamespace(execute=AsyncMock())
|
|
response = main.AccountCookieResponse(account_id=501, cookie_data=None)
|
|
|
|
with (
|
|
patch.object(main, "get_owned_account", new=AsyncMock(return_value=account)),
|
|
patch.object(main, "_get_account_cookie_data", return_value=None),
|
|
patch.object(main, "_build_cookie_response", new=AsyncMock(return_value=response)) as build,
|
|
):
|
|
result = await main.get_desktop_login_credential(
|
|
account_id=501,
|
|
db=db,
|
|
user=SimpleNamespace(id=9, role="operator"),
|
|
)
|
|
|
|
self.assertIs(result, response)
|
|
db.execute.assert_not_awaited()
|
|
build.assert_awaited_once_with(account, None)
|
|
|
|
async def test_normal_cookie_management_endpoint_is_not_guarded(self):
|
|
account = self._account()
|
|
db = SimpleNamespace(execute=AsyncMock())
|
|
response = main.AccountCookieResponse(account_id=501, cookie_data="cookie-json")
|
|
|
|
with (
|
|
patch.object(main, "get_owned_account", new=AsyncMock(return_value=account)),
|
|
patch.object(main, "_build_cookie_response", new=AsyncMock(return_value=response)) as build,
|
|
):
|
|
result = await main.get_account_cookie(
|
|
account_id=501,
|
|
purpose="management",
|
|
db=db,
|
|
user=SimpleNamespace(id=9, role="operator"),
|
|
)
|
|
|
|
self.assertIs(result, response)
|
|
db.execute.assert_not_awaited()
|
|
build.assert_awaited_once_with(
|
|
account,
|
|
"cookie-json",
|
|
runtime_check=False,
|
|
)
|
|
|
|
async def test_cookie_response_forwards_static_management_check(self):
|
|
account = self._account()
|
|
summary = {
|
|
"cookie_count": 2,
|
|
"cookie_valid": True,
|
|
"cookie_expired": False,
|
|
"reason": "ok",
|
|
"expires_at": None,
|
|
"key_names": ["sessionid"],
|
|
}
|
|
detail = {
|
|
"has_sessionid": True,
|
|
"sessionid": "sid",
|
|
"sessionid_ss": "",
|
|
"im_ready": False,
|
|
"im_status": "static",
|
|
"can_skip_browser": False,
|
|
"should_reset": False,
|
|
}
|
|
|
|
with (
|
|
patch.object(main, "cookie_summary", return_value=summary),
|
|
patch.object(
|
|
main,
|
|
"build_cookie_credential_detail",
|
|
new=AsyncMock(return_value=detail),
|
|
) as build_detail,
|
|
):
|
|
result = await main._build_cookie_response(
|
|
account,
|
|
"cookie-json",
|
|
runtime_check=False,
|
|
)
|
|
|
|
self.assertTrue(result.has_sessionid)
|
|
build_detail.assert_awaited_once_with(
|
|
"cookie-json",
|
|
None,
|
|
runtime_check=False,
|
|
)
|
|
|
|
async def test_legacy_cookie_request_is_guarded_for_existing_desktop_clients(self):
|
|
account = self._account()
|
|
profile = SimpleNamespace(
|
|
sec_user_id=" ",
|
|
synced_at=account.cookie_updated_at,
|
|
)
|
|
db = SimpleNamespace(execute=AsyncMock(return_value=_ProfileResult(profile)))
|
|
|
|
with (
|
|
patch.object(main, "get_owned_account", new=AsyncMock(return_value=account)),
|
|
patch.object(main, "_build_cookie_response", new=AsyncMock()) as build,
|
|
):
|
|
with self.assertRaises(main.HTTPException) as caught:
|
|
await main.get_account_cookie(
|
|
account_id=501,
|
|
purpose=None,
|
|
db=db,
|
|
user=SimpleNamespace(id=9, role="operator"),
|
|
)
|
|
|
|
self.assertEqual(caught.exception.status_code, 409)
|
|
self.assertIn("缺少 sec_user_id", str(caught.exception.detail))
|
|
build.assert_not_awaited()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|