102 lines
4.2 KiB
Python
102 lines
4.2 KiB
Python
"""工作台「导入外部 CK」：入库 + 自动加入工作台 + 归属导入者。"""
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from tests.support.db import get_test_engine, make_test_session
|
|
from web.backend.models import HuyaAccount, HuyaWorkbench, HuyaWorkbenchAccount, User
|
|
from web.backend.permissions import ROLE_PERMISSIONS
|
|
from web.backend.routers.huya import import_workbench_cookies
|
|
from web.backend.routers.huya._common import _require_huya_workbench_import_ck
|
|
from web.backend.schemas import HuyaWorkbenchCookieImport
|
|
|
|
COOKIE_1 = (
|
|
"yyuid=1001; udb_uid=1001; udb_passport=hy_test_1; "
|
|
"udb_cred=cred1; udb_biztoken=biz1"
|
|
)
|
|
COOKIE_2 = (
|
|
"yyuid=1002; udb_uid=1002; udb_passport=hy_test_2; "
|
|
"udb_cred=cred2; udb_biztoken=biz2"
|
|
)
|
|
|
|
|
|
class TestImportExternalCkToWorkbench:
|
|
def setup_method(self):
|
|
self.engine = get_test_engine()
|
|
self.session = make_test_session(self.engine)
|
|
self.user = User(username="operator", password_hash="hash", role="super_admin")
|
|
self.session.add(self.user)
|
|
self.session.commit()
|
|
|
|
def teardown_method(self):
|
|
self.session.close()
|
|
|
|
def test_import_creates_accounts_and_joins_workbench(self):
|
|
result = import_workbench_cookies(
|
|
HuyaWorkbenchCookieImport(
|
|
text=f"小号A----{COOKIE_1}\n这不是一行 CK\n{COOKIE_2}",
|
|
tag="外部CK",
|
|
handbook_scope="elite",
|
|
),
|
|
db=self.session,
|
|
current=self.user,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["count"] == 2
|
|
assert result["skipped"] == 1
|
|
assert len(result["account_ids"]) == 2
|
|
|
|
accounts = {
|
|
account.yyuid: account for account in self.session.query(HuyaAccount).all()
|
|
}
|
|
assert set(accounts) == {"1001", "1002"}
|
|
# 无归属的新账号自动归属导入者，否则客服在工作台看不到自己导入的账号
|
|
assert all(account.assigned_to == self.user.id for account in accounts.values())
|
|
assert accounts["1001"].tag == "外部CK"
|
|
|
|
joined = {
|
|
row.account_id
|
|
for row in self.session.query(HuyaWorkbenchAccount)
|
|
.filter(HuyaWorkbenchAccount.user_id == self.user.id)
|
|
.all()
|
|
}
|
|
assert joined == set(result["account_ids"])
|
|
assert self.session.query(HuyaWorkbench).count() == 1
|
|
|
|
def test_reimport_is_idempotent(self):
|
|
payload = HuyaWorkbenchCookieImport(text=COOKIE_1, handbook_scope="elite")
|
|
first = import_workbench_cookies(payload, db=self.session, current=self.user)
|
|
second = import_workbench_cookies(payload, db=self.session, current=self.user)
|
|
assert first["account_ids"] == second["account_ids"]
|
|
assert len(first["account_ids"]) == 1
|
|
assert self.session.query(HuyaAccount).count() == 1
|
|
assert self.session.query(HuyaWorkbenchAccount).count() == 1
|
|
|
|
def test_all_invalid_lines_reports_zero(self):
|
|
result = import_workbench_cookies(
|
|
HuyaWorkbenchCookieImport(text="垃圾行\n更多垃圾", handbook_scope="elite"),
|
|
db=self.session,
|
|
current=self.user,
|
|
)
|
|
assert result["count"] == 0
|
|
assert result["skipped"] == 2
|
|
assert result["account_ids"] == []
|
|
assert self.session.query(HuyaAccount).count() == 0
|
|
assert self.session.query(HuyaWorkbench).count() == 0
|
|
|
|
def test_support_can_import_workbench_ck_by_default(self):
|
|
"""客服默认允许工作台导入外部 CK，但旧 huya:import 仍只属于运营。"""
|
|
assert "huya:elite:import_ck" in ROLE_PERMISSIONS["support"]
|
|
assert "huya:elite:import_ck" not in ROLE_PERMISSIONS["operation"]
|
|
|
|
support = User(username="cs", password_hash="hash", role="support")
|
|
operation = User(username="op", password_hash="hash", role="operation")
|
|
guest = User(username="guest", password_hash="hash", role="guest")
|
|
assert _require_huya_workbench_import_ck(support) is None
|
|
assert _require_huya_workbench_import_ck(operation) is None
|
|
|
|
with pytest.raises(HTTPException) as excinfo:
|
|
_require_huya_workbench_import_ck(guest)
|
|
assert excinfo.value.status_code == 403
|