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:
yml2213
2026-11-08 18:30:51 +08:00
parent c8b62e5267
commit 6b2285913f
16 changed files with 1320 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_migrations.py
+1 -1
View File
@@ -26,7 +26,7 @@ TESTS_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = TESTS_DIR.parent
VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"
HEAD_REVISION = "20261108_0073"
HEAD_REVISION = "20261108_0074"
# 迁移新增、但不声明在模型里的复合查询索引。
EXTRA_INDEXES = (
tests/test_weibo_accounts.py
+249 -7
View File
@@ -23,6 +23,8 @@ from web.backend.permissions import ROLE_PERMISSIONS
from web.backend.routers import weibo as weibo_router
from web.backend.routers.weibo import (
add_workbench_accounts,
assignments_summary,
batch_assign_accounts,
batch_tag_accounts,
copy_account_ck,
create_bind_qrcodes,
@@ -30,6 +32,7 @@ from web.backend.routers.weibo import (
enable_account,
get_account,
import_cookies,
list_account_ids,
list_account_tags,
list_accounts,
list_workbench_accounts,
@@ -43,6 +46,7 @@ from web.backend.schemas import (
WeiboAccountBatchTagRequest,
WeiboAccountDisableRequest,
WeiboAccountUpdate,
WeiboBatchAssignRequest,
WeiboCookieImport,
WeiboExchangeRequest,
WeiboPaymentStatusRequest,
@@ -241,11 +245,13 @@ def test_occupied_workbench_account_cannot_be_updated_by_another_user(db, curren
)
db.add_all([owner, other])
db.commit()
add_workbench_accounts(
WeiboWorkbenchAccountsRequest(account_ids=[imported.id]),
db=db,
current=owner,
db.add(
WeiboWorkbenchAccount(
user_id=owner.id,
account_id=imported.id,
)
)
db.commit()
result = import_cookies(
WeiboCookieImport(
@@ -328,6 +334,9 @@ def test_copy_account_ck_uses_export_permission_defaults(db, current_user):
)
db.add(support)
db.commit()
account = db.query(WeiboAccount).filter(WeiboAccount.id == account_out.id).one()
account.assigned_to = support.id
db.commit()
support_result = copy_account_ck(account_out.id, db=db, current=support)
assert support_result.ck_text == "SCF=a; SUB=b"
@@ -406,6 +415,9 @@ def test_weibo_workbench_is_isolated_per_user(db, current_user):
)
db.add_all([user_a, user_b])
db.commit()
first.assigned_to = user_a.id
second.assigned_to = user_b.id
db.commit()
add_workbench_accounts(
WeiboWorkbenchAccountsRequest(account_ids=[first.id]),
@@ -413,14 +425,14 @@ def test_weibo_workbench_is_isolated_per_user(db, current_user):
current=user_a,
)
pool_before = list_workbench_pool(page=1, page_size=20, db=db, current=user_a)
assert [item.uid for item in pool_before["items"]] == [second.uid]
assert pool_before["items"] == []
with pytest.raises(HTTPException) as exc_info:
add_workbench_accounts(
WeiboWorkbenchAccountsRequest(account_ids=[first.id]),
db=db,
current=user_b,
)
assert exc_info.value.status_code == 409
assert exc_info.value.status_code == 403
add_workbench_accounts(
WeiboWorkbenchAccountsRequest(account_ids=[second.id]),
db=db,
@@ -458,7 +470,10 @@ def test_weibo_workbench_is_isolated_per_user(db, current_user):
== 1
)
pool_b = list_workbench_pool(page=1, page_size=20, db=db, current=user_b)
assert [item.uid for item in pool_b["items"]] == [first.uid]
assert pool_b["items"] == []
pool_a = list_workbench_pool(page=1, page_size=20, db=db, current=user_a)
assert [item.uid for item in pool_a["items"]] == [first.uid]
def test_batch_tag_updates_selected_accounts(db, current_user):
@@ -522,6 +537,233 @@ def test_list_accounts_returns_parallel_page_shape(db, current_user):
}
def test_weibo_account_scope_for_support_and_operator(db):
support_a = User(
username="weibo-scope-a",
password_hash="hash",
role="support",
is_active=True,
)
support_b = User(
username="weibo-scope-b",
password_hash="hash",
role="support",
is_active=True,
)
operator = User(
username="weibo-scope-operator",
password_hash="hash",
role="operation",
is_active=True,
)
db.add_all([support_a, support_b, operator])
db.commit()
own = WeiboAccount(
uid="4014000301",
nickname="自己账号",
cookie="SCF=a; SUB=own",
tag="自己标签",
assigned_to=support_a.id,
)
other = WeiboAccount(
uid="4014000302",
nickname="他人账号",
cookie="SCF=b; SUB=other",
tag="他人标签",
assigned_to=support_b.id,
)
db.add_all([own, other])
db.commit()
own_page = list_accounts(page=1, page_size=20, db=db, current=support_a)
assert own_page["total"] == 1
own_dump = own_page["items"][0].model_dump()
assert own_dump["uid"] == own.uid
assert own_dump["assigned_to"] == support_a.id
assert own_dump["assigned_username"] == support_a.username
assert list_account_ids(db=db, current=support_a)["account_ids"] == [own.id]
assert list_account_tags(db=db, current=support_a)["tags"] == ["自己标签"]
with pytest.raises(HTTPException) as exc_info:
get_account(other.id, db=db, current=support_a)
assert exc_info.value.status_code == 404
own.tag = "原标签"
db.commit()
with pytest.raises(HTTPException) as exc_info:
batch_tag_accounts(
WeiboAccountBatchTagRequest(ids=[own.id, other.id], tag="越权"),
db=db,
current=support_a,
)
assert exc_info.value.status_code == 404
db.refresh(own)
db.refresh(other)
assert own.tag == "原标签"
assert other.tag == "他人标签"
operator_page = list_accounts(page=1, page_size=20, db=db, current=operator)
assert operator_page["total"] == 2
assert set(list_account_tags(db=db, current=operator)["tags"]) == {
"他人标签",
"原标签",
}
preview = assignments_summary(db=db, current=operator)
assert preview["total_count"] == 2
assert preview["assigned_count"] == 2
assert preview["unassigned_count"] == 0
counts = {item["id"]: item["assigned_count"] for item in preview["support_users"]}
assert counts == {support_a.id: 1, support_b.id: 1}
with pytest.raises(HTTPException) as exc_info:
batch_assign_accounts(
WeiboBatchAssignRequest(account_ids=[own.id], assigned_to=support_b.id),
db=db,
current=support_a,
)
assert exc_info.value.status_code == 403
def test_weibo_batch_assign_cleans_previous_workbench(db):
operator = User(
username="weibo-assign-operator",
password_hash="hash",
role="operation",
is_active=True,
)
old_support = User(
username="weibo-assign-old",
password_hash="hash",
role="support",
is_active=True,
)
new_support = User(
username="weibo-assign-new",
password_hash="hash",
role="support",
is_active=True,
)
db.add_all([operator, old_support, new_support])
db.commit()
account = WeiboAccount(
uid="4014000311",
nickname="待改分配",
cookie="SCF=a; SUB=b",
assigned_to=old_support.id,
)
db.add(account)
db.commit()
add_workbench_accounts(
WeiboWorkbenchAccountsRequest(account_ids=[account.id]),
db=db,
current=old_support,
)
result = batch_assign_accounts(
WeiboBatchAssignRequest(
account_ids=[account.id],
assigned_to=new_support.id,
),
db=db,
current=operator,
)
assert result["updated"] == 1
db.refresh(account)
assert account.assigned_to == new_support.id
assert (
db.query(WeiboWorkbenchAccount)
.filter(WeiboWorkbenchAccount.account_id == account.id)
.count()
== 0
)
add_workbench_accounts(
WeiboWorkbenchAccountsRequest(account_ids=[account.id]),
db=db,
current=new_support,
)
audit = (
db.query(AuditLog)
.filter(AuditLog.action == "weibo:account:assign")
.order_by(AuditLog.id.desc())
.first()
)
assert audit is not None
assert str(account.id) in audit.detail
assert "SCF=a" not in audit.detail
def test_support_cannot_update_another_users_account_by_reimport(db):
support_a = User(
username="weibo-reimport-a",
password_hash="hash",
role="support",
is_active=True,
)
support_b = User(
username="weibo-reimport-b",
password_hash="hash",
role="support",
is_active=True,
)
db.add_all([support_a, support_b])
db.commit()
account = WeiboAccount(
uid="4014000321",
nickname="他人 CK",
cookie="SCF=old; SUB=owner",
assigned_to=support_a.id,
)
db.add(account)
db.commit()
StubWeiboClient.snapshot = WeiboHomeSnapshot(
uid=account.uid,
nickname="他人 CK",
is_bound=False,
role_name="",
can_rebind=True,
rebind_remain_days=0,
weidou_balance=100,
bind_url=BIND_URL,
)
denied = import_cookies(
WeiboCookieImport(
text="SCF=new; SUB=attacker",
update_existing=True,
),
db=db,
current=support_b,
)
assert denied.count == 0
assert len(denied.errors) == 1
assert "已分配给其他客服" in denied.errors[0].message
db.refresh(account)
assert account.cookie == "SCF=old; SUB=owner"
assert account.assigned_to == support_a.id
StubWeiboClient.snapshot = WeiboHomeSnapshot(
uid="4014000322",
nickname="新账号",
is_bound=False,
role_name="",
can_rebind=True,
rebind_remain_days=0,
weidou_balance=100,
bind_url=BIND_URL,
)
imported = import_cookies(
WeiboCookieImport(text="SCF=new; SUB=own"),
db=db,
current=support_b,
)
assert imported.count == 1
new_account = db.query(WeiboAccount).filter(WeiboAccount.uid == "4014000322").one()
assert new_account.assigned_to == support_b.id
def test_account_crud_and_lifecycle(db, current_user):
account_out = import_cookies(
WeiboCookieImport(text="SCF=a; SUB=b", tag="初始"),