feat(weibo): 增加外部 CK 账号 CRUD 与生命周期

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

tests/test_weibo_accounts.py

@@ -21,10 +21,17 @@ from web.backend.models import User, WeiboAccount
 from web.backend.routers import weibo as weibo_router
 from web.backend.routers.weibo import (
     create_bind_qrcodes,
+    disable_account,
+    enable_account,
+    get_account,
     import_cookies,
     list_accounts,
+    refresh_account,
+    update_account,
 )
 from web.backend.schemas import (
+    WeiboAccountDisableRequest,
+    WeiboAccountUpdate,
     WeiboCookieImport,
     WeiboExchangeRequest,
     WeiboPaymentStatusRequest,
@@ -204,6 +211,9 @@ def test_list_accounts_returns_parallel_page_shape(db, current_user):
         "id",
         "uid",
         "nickname",
+        "tag",
+        "remark",
+        "is_enabled",
         "is_bound",
         "can_rebind",
         "rebind_remain_days",
@@ -212,6 +222,96 @@ def test_list_accounts_returns_parallel_page_shape(db, current_user):
     }
 
 
+def test_account_crud_and_lifecycle(db, current_user):
+    account_out = import_cookies(
+        WeiboCookieImport(text="SCF=a; SUB=b", tag="初始"),
+        db=db,
+        current=current_user,
+    ).accounts[0]
+
+    detail = get_account(
+        account_out.id,
+        db=db,
+        current=current_user,
+    ).model_dump()
+    assert detail["is_enabled"] is True
+    assert detail["remark"] == ""
+
+    updated = update_account(
+        account_out.id,
+        WeiboAccountUpdate(tag="重点", remark="客户 A"),
+        db=db,
+        current=current_user,
+    ).model_dump()
+    assert updated["tag"] == "重点"
+    assert updated["remark"] == "客户 A"
+
+    disabled = disable_account(
+        account_out.id,
+        WeiboAccountDisableRequest(reason="暂停使用"),
+        db=db,
+        current=current_user,
+    ).model_dump()
+    assert disabled["is_enabled"] is False
+    assert disabled["disabled_reason"] == "暂停使用"
+
+    disabled_page = list_accounts(
+        enabled=False,
+        page=1,
+        page_size=20,
+        db=db,
+        current=current_user,
+    )
+    assert disabled_page["total"] == 1
+    assert disabled_page["items"][0].model_dump()["is_enabled"] is False
+
+    with pytest.raises(HTTPException) as exc_info:
+        refresh_account(account_out.id, db=db, current=current_user)
+    assert exc_info.value.status_code == 409
+
+    disabled_actions = [
+        lambda: create_bind_qrcodes(account_out.id, db=db, current=current_user),
+        lambda: weibo_router.list_mall_goods(
+            account_out.id,
+            db=db,
+            current=current_user,
+        ),
+        lambda: weibo_router.exchange_goods(
+            account_out.id,
+            WeiboExchangeRequest(goods_id=37, expected_price=100, confirm=True),
+            db=db,
+            current=current_user,
+        ),
+        lambda: weibo_router.create_recharge_order(
+            account_out.id,
+            WeiboRechargeRequest(amount_yuan=1),
+            db=db,
+            current=current_user,
+        ),
+    ]
+    for action in disabled_actions:
+        with pytest.raises(HTTPException) as disabled_exc:
+            action()
+        assert disabled_exc.value.status_code == 409
+
+    import_cookies(
+        WeiboCookieImport(text="SCF=c; SUB=d", tag=""),
+        db=db,
+        current=current_user,
+    )
+    account = db.query(WeiboAccount).filter(WeiboAccount.id == account_out.id).one()
+    assert account.is_enabled is False
+    assert account.tag == "重点"
+
+    enabled = enable_account(
+        account_out.id,
+        db=db,
+        current=current_user,
+    ).model_dump()
+    assert enabled["is_enabled"] is True
+    assert enabled["disabled_reason"] == ""
+
+
 def test_bind_qrcodes_returns_both_channels_and_partial_error(db, current_user):
     account = import_cookies(
         WeiboCookieImport(text="SCF=a; SUB=b", tag=""),