dachui/live-hub-py
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(weibo): 支持管理员强制转入工作台且不影响原使用人

Browse Source
This commit is contained in:
yml
2026-11-12 12:05:26 +08:00
parent 11b5a4d938
commit 31045e0f2e
11 changed files with 604 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_migrations.py
+115 -1
View File
@@ -17,16 +17,18 @@ from alembic import command
from alembic.config import Config
from alembic.script import ScriptDirectory
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.orm import Session
import web.backend.database as database_module
from tests.support.db import get_mysql_credentials, get_test_engine
from web.backend.database import Base
from web.backend.models import User, WeiboAccount
TESTS_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = TESTS_DIR.parent
VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"
HEAD_REVISION = "20261111_0076"
HEAD_REVISION = "20261112_0077"
# 迁移新增、但不声明在模型里的复合查询索引。
EXTRA_INDEXES = (
@@ -154,6 +156,118 @@ class TestMigrationSmoke:
with admin_engine.begin() as conn:
conn.execute(text(f"DROP DATABASE IF EXISTS `{mig_db}`;"))
def test_0077_converts_super_admin_membership_to_overlay(self):
creds = get_mysql_credentials()
mig_db = f"douyu_login_mig_{secrets.token_hex(4)}"
mig_url = (
f"mysql+pymysql://{quote_plus(creds['user'])}:"
f"{quote_plus(creds['password'])}@{creds['host']}:{creds['port']}/"
f"{mig_db}?charset=utf8mb4"
)
admin_engine = get_test_engine()
with admin_engine.begin() as conn:
conn.execute(
text(
f"CREATE DATABASE `{mig_db}` "
"CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;"
)
)
original_url = database_module.DATABASE_URL
mig_engine = create_engine(mig_url)
try:
database_module.DATABASE_URL = mig_url
cfg = _migration_config()
cfg.set_main_option("sqlalchemy.url", mig_url)
command.upgrade(cfg, "20261111_0076")
with Session(mig_engine) as session:
admin = User(
username="migration-admin",
password_hash="hash",
role="super_admin",
is_active=True,
)
support = User(
username="migration-support",
password_hash="hash",
role="support",
is_active=True,
)
admin_account = WeiboAccount(
uid="4014000771",
nickname="管理员旧映射",
cookie="SCF=a; SUB=admin",
)
support_account = WeiboAccount(
uid="4014000772",
nickname="客服旧映射",
cookie="SCF=b; SUB=support",
)
session.add_all([admin, support, admin_account, support_account])
session.commit()
admin_id = admin.id
support_id = support.id
admin_account_id = admin_account.id
support_account_id = support_account.id
with mig_engine.begin() as conn:
for user_id, account_id in (
(admin_id, admin_account_id),
(support_id, support_account_id),
):
conn.execute(
text(
"INSERT INTO weibo_workbench_accounts "
"(user_id, account_id, created_at) "
"VALUES (:user_id, :account_id, UTC_TIMESTAMP())"
),
{"user_id": user_id, "account_id": account_id},
)
command.upgrade(cfg, "20261112_0077")
with mig_engine.begin() as conn:
rows = conn.execute(
text(
"SELECT user_id, account_id, exclusive_slot "
"FROM weibo_workbench_accounts ORDER BY user_id"
)
).all()
assert {
(user_id, account_id, exclusive_slot)
for user_id, account_id, exclusive_slot in rows
} == {
(admin_id, admin_account_id, None),
(support_id, support_account_id, 1),
}
# 管理员可以给同一个账号再挂一条并行映射，普通唯一槽位保持不变。
conn.execute(
text(
"INSERT INTO weibo_workbench_accounts "
"(user_id, account_id, exclusive_slot, created_at) "
"VALUES (:user_id, :account_id, NULL, UTC_TIMESTAMP())"
),
{"user_id": admin_id, "account_id": support_account_id},
)
assert (
conn.execute(
text(
"SELECT COUNT(*) FROM weibo_workbench_accounts "
"WHERE account_id = :account_id"
),
{"account_id": support_account_id},
).scalar_one()
== 2
)
finally:
database_module.DATABASE_URL = original_url
mig_engine.dispose()
with admin_engine.begin() as conn:
conn.execute(text(f"DROP DATABASE IF EXISTS `{mig_db}`;"))
class TestRollbackBackfillScript:
"""回滚前置脚本：把账号侧 CK 搬回 `login_tasks`。
tests/test_weibo_accounts.py
+189 -6
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
from dataclasses import replace
from typing import ClassVar
import pytest
@@ -277,6 +278,7 @@ def test_occupied_workbench_account_cannot_be_updated_by_another_user(db, curren
WeiboWorkbenchAccount(
user_id=owner.id,
account_id=imported.id,
exclusive_slot=1,
)
)
db.commit()
@@ -549,8 +551,16 @@ def test_batch_import_workbench_accounts_reports_each_identifier(db, current_use
db.commit()
db.add_all(
[
WeiboWorkbenchAccount(user_id=current_user.id, account_id=existing.id),
WeiboWorkbenchAccount(user_id=other_user.id, account_id=occupied.id),
WeiboWorkbenchAccount(
user_id=current_user.id,
account_id=existing.id,
exclusive_slot=1,
),
WeiboWorkbenchAccount(
user_id=other_user.id,
account_id=occupied.id,
exclusive_slot=1,
),
]
)
db.commit()
@@ -581,6 +591,11 @@ def test_batch_import_workbench_accounts_reports_each_identifier(db, current_use
"批量导入重名": "ambiguous",
"批量导入停用": "disabled",
}
occupied_item = next(
item for item in result["items"] if item["identifier"] == occupied.uid
)
assert occupied_item["occupied_by_user_id"] == other_user.id
assert occupied_item["occupied_by_username"] == other_user.username
workbench_ids = {
account_id
for (account_id,) in db.query(WeiboWorkbenchAccount.account_id)
@@ -590,6 +605,151 @@ def test_batch_import_workbench_accounts_reports_each_identifier(db, current_use
assert workbench_ids == {existing.id, added.id}
def test_super_admin_force_adds_account_without_removing_original_owner(
db, current_user
):
other_user = User(
username="weibo-batch-import-owner",
password_hash="hash",
role="support",
is_active=True,
)
db.add(other_user)
db.commit()
account = WeiboAccount(
uid="4014000911",
nickname="用户8477791859",
cookie="SCF=a; SUB=force-add",
assigned_to=other_user.id,
)
db.add(account)
db.commit()
db.add(
WeiboWorkbenchAccount(
user_id=other_user.id,
account_id=account.id,
exclusive_slot=1,
)
)
db.commit()
admin_pool = list_workbench_pool(page=1, page_size=20, db=db, current=current_user)
assert [item.id for item in admin_pool["items"]] == [account.id]
result = import_workbench_accounts_by_text(
WeiboWorkbenchBatchImportRequest(
text=account.nickname,
force_add_occupied=True,
),
db=db,
current=current_user,
).model_dump()
assert result["requested"] == 1
assert result["added"] == 0
assert result["forced_added"] == 1
assert result["already_added"] == 0
assert result["failed"] == 0
assert result["account_ids"] == [account.id]
assert result["items"] == [
{
"identifier": account.nickname,
"status": "forced_added",
"message": (f"已强制转入，{other_user.username}的工作台保持不变"),
"account_id": account.id,
"uid": account.uid,
"nickname": account.nickname,
"occupied_by_user_id": other_user.id,
"occupied_by_username": other_user.username,
}
]
db.expire_all()
memberships = (
db.query(WeiboWorkbenchAccount)
.filter(WeiboWorkbenchAccount.account_id == account.id)
.order_by(WeiboWorkbenchAccount.user_id)
.all()
)
assert {
(membership.user_id, membership.exclusive_slot) for membership in memberships
} == {
(other_user.id, 1),
(current_user.id, None),
}
assert (
list_workbench_accounts(page=1, page_size=20, db=db, current=other_user)[
"total"
]
== 1
)
assert (
list_workbench_accounts(page=1, page_size=20, db=db, current=current_user)[
"total"
]
== 1
)
StubWeiboClient.snapshot = replace(StubWeiboClient.snapshot, uid=account.uid)
updated = import_cookies(
WeiboCookieImport(
text="SCF=updated; SUB=owner-still-works",
update_existing=True,
),
db=db,
current=other_user,
)
assert updated.count == 1
assert updated.errors == []
db.refresh(account)
assert "SUB=owner-still-works" in account.cookie
remove_workbench_accounts(
WeiboWorkbenchAccountsRequest(account_ids=[account.id]),
db=db,
current=current_user,
)
assert (
list_workbench_accounts(page=1, page_size=20, db=db, current=other_user)[
"total"
]
== 1
)
audit = (
db.query(AuditLog)
.filter(AuditLog.action == "weibo:workbench:batch_import")
.one()
)
assert f'"forced_added_account_ids":[{account.id}]' in audit.detail
assert f'"existing_owner_user_id":{other_user.id}' in audit.detail
assert f'"existing_owner_username":"{other_user.username}"' in audit.detail
@pytest.mark.parametrize("role", ["operation", "support"])
def test_non_super_admin_cannot_force_add_occupied_workbench_account(db, role):
user = User(
username=f"weibo-force-add-{role}",
password_hash="hash",
role=role,
is_active=True,
)
db.add(user)
db.commit()
with pytest.raises(HTTPException) as excinfo:
import_workbench_accounts_by_text(
WeiboWorkbenchBatchImportRequest(
text="用户8477791859",
force_add_occupied=True,
),
db=db,
current=user,
)
assert excinfo.value.status_code == 403
assert "仅超级管理员" in str(excinfo.value.detail)
def test_operation_can_transfer_workbench_account_to_collab(db):
operator = User(
username="weibo-transfer-operator",
@@ -612,6 +772,7 @@ def test_operation_can_transfer_workbench_account_to_collab(db):
WeiboWorkbenchAccount(
user_id=operator.id,
account_id=account.id,
exclusive_slot=1,
)
)
db.commit()
@@ -830,7 +991,13 @@ def test_weibo_batch_assign_cleans_previous_workbench(db):
role="support",
is_active=True,
)
db.add_all([operator, old_support, new_support])
admin = User(
username="weibo-assign-admin",
password_hash="hash",
role="super_admin",
is_active=True,
)
db.add_all([operator, old_support, new_support, admin])
db.commit()
account = WeiboAccount(
uid="4014000311",
@@ -845,6 +1012,14 @@ def test_weibo_batch_assign_cleans_previous_workbench(db):
db=db,
current=old_support,
)
db.add(
WeiboWorkbenchAccount(
user_id=admin.id,
account_id=account.id,
exclusive_slot=None,
)
)
db.commit()
result = batch_assign_accounts(
WeiboBatchAssignRequest(
@@ -858,17 +1033,25 @@ def test_weibo_batch_assign_cleans_previous_workbench(db):
assert result["updated"] == 1
db.refresh(account)
assert account.assigned_to == new_support.id
assert (
memberships = (
db.query(WeiboWorkbenchAccount)
.filter(WeiboWorkbenchAccount.account_id == account.id)
.count()
== 0
.all()
)
assert {
(membership.user_id, membership.exclusive_slot) for membership in memberships
} == {(admin.id, None)}
add_workbench_accounts(
WeiboWorkbenchAccountsRequest(account_ids=[account.id]),
db=db,
current=new_support,
)
assert (
db.query(WeiboWorkbenchAccount)
.filter(WeiboWorkbenchAccount.account_id == account.id)
.count()
== 2
)
audit = (
db.query(AuditLog)
.filter(AuditLog.action == "weibo:account:assign")