feat(weibo): 增加工作台独占与快捷操作

tests/test_migrations.py

@@ -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_0072"
+HEAD_REVISION = "20261108_0073"
 
 # 迁移新增、但不声明在模型里的复合查询索引。
 EXTRA_INDEXES = (

tests/test_weibo_accounts.py

@@ -30,9 +30,11 @@ from web.backend.routers.weibo import (
     enable_account,
     get_account,
     import_cookies,
+    list_account_tags,
     list_accounts,
     list_workbench_accounts,
     list_workbench_pool,
+    list_workbench_tags,
     refresh_account,
     remove_workbench_accounts,
     update_account,
@@ -138,23 +140,42 @@ def stub_client(monkeypatch):
     monkeypatch.setattr(weibo_service, "WeiboClient", StubWeiboClient)
 
 
-def test_import_creates_encrypted_account_and_is_idempotent(db, current_user):
+def test_import_creates_once_and_updates_existing_only_when_requested(db, current_user):
     first = import_cookies(
         WeiboCookieImport(text="SCF=a; SUB=b", tag="运营"),
         db=db,
         current=current_user,
     )
-    second = import_cookies(
+    duplicate = import_cookies(
         WeiboCookieImport(text="SCF=c; SUB=d", tag="新标签"),
         db=db,
         current=current_user,
     )
 
     assert first.count == 1
-    assert second.count == 1
+    assert duplicate.count == 0
+    assert duplicate.skipped == 1
+    assert duplicate.accounts == []
     assert db.query(WeiboAccount).count() == 1
     account = db.query(WeiboAccount).one()
     assert account.uid == "4014000001"
+    assert account.cookie == "SCF=a; SUB=b"
+    assert account.tag == "运营"
+
+    updated = import_cookies(
+        WeiboCookieImport(
+            text="SCF=c; SUB=d",
+            tag="新标签",
+            update_existing=True,
+        ),
+        db=db,
+        current=current_user,
+    )
+
+    assert updated.count == 1
+    assert updated.skipped == 0
+    assert db.query(WeiboAccount).count() == 1
+    account = db.query(WeiboAccount).one()
     assert account.cookie == "SCF=c; SUB=d"
     assert account.tag == "新标签"
     assert account.ck_check_status == "valid"
@@ -162,7 +183,7 @@ def test_import_creates_encrypted_account_and_is_idempotent(db, current_user):
     assert raw_cookie.startswith("enc:v1:")
     assert "SUB=d" not in raw_cookie
 
-    dumped = second.accounts[0].model_dump()
+    dumped = updated.accounts[0].model_dump()
     assert dumped["uid"] == "4014000001"
     assert dumped["is_bound"] is False
     assert dumped["weidou_balance"] == 100
@@ -198,6 +219,50 @@ def test_import_reports_payment_access_failure(db, current_user):
     assert db.query(WeiboAccount).count() == 0
 
 
+def test_occupied_workbench_account_cannot_be_updated_by_another_user(db, current_user):
+    imported_out = import_cookies(
+        WeiboCookieImport(text="SCF=a; SUB=owner"),
+        db=db,
+        current=current_user,
+    ).accounts[0]
+    imported = db.query(WeiboAccount).filter(WeiboAccount.id == imported_out.id).one()
+    original_cookie = imported.cookie
+    owner = User(
+        username="weibo-owner",
+        password_hash="hash",
+        role="support",
+        is_active=True,
+    )
+    other = User(
+        username="weibo-other",
+        password_hash="hash",
+        role="support",
+        is_active=True,
+    )
+    db.add_all([owner, other])
+    db.commit()
+    add_workbench_accounts(
+        WeiboWorkbenchAccountsRequest(account_ids=[imported.id]),
+        db=db,
+        current=owner,
+    )
+
+    result = import_cookies(
+        WeiboCookieImport(
+            text="SCF=z; SUB=other",
+            update_existing=True,
+        ),
+        db=db,
+        current=other,
+    )
+
+    assert result.count == 0
+    assert len(result.errors) == 1
+    assert "其他人的工作台占用" in result.errors[0].message
+    db.refresh(imported)
+    assert imported.cookie == original_cookie
+
+
 def test_refresh_marks_payment_access_failure(db, current_user):
     account_out = import_cookies(
         WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
@@ -347,6 +412,15 @@ def test_weibo_workbench_is_isolated_per_user(db, current_user):
         db=db,
         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]
+    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
     add_workbench_accounts(
         WeiboWorkbenchAccountsRequest(account_ids=[second.id]),
         db=db,
@@ -357,10 +431,18 @@ def test_weibo_workbench_is_isolated_per_user(db, current_user):
     page_b = list_workbench_accounts(page=1, page_size=20, db=db, current=user_b)
     assert [item.uid for item in page_a["items"]] == [first.uid]
     assert [item.uid for item in page_b["items"]] == [second.uid]
+    assert (
+        list_workbench_accounts(tag="A", page=1, page_size=20, db=db, current=user_a)[
+            "total"
+        ]
+        == 1
+    )
+    assert list_workbench_tags(db=db, current=user_a)["tags"] == ["A"]
+    assert list_account_tags(db=db, current=current_user)["tags"] == ["A", "B"]
     assert db.query(WeiboWorkbenchAccount).count() == 2
 
     pool_a = list_workbench_pool(page=1, page_size=20, db=db, current=user_a)
-    assert [item.uid for item in pool_a["items"]] == [second.uid]
+    assert pool_a["items"] == []
 
     remove_workbench_accounts(
         WeiboWorkbenchAccountsRequest(account_ids=[first.id]),
@@ -375,6 +457,8 @@ def test_weibo_workbench_is_isolated_per_user(db, current_user):
         list_workbench_accounts(page=1, page_size=20, db=db, current=user_b)["total"]
         == 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]
 
 
 def test_batch_tag_updates_selected_accounts(db, current_user):