优化后台用户管理和发积分

2026-09-20 16:30:45 +08:00
parent fb3b2904ae
commit 690eb3d843
12 changed files with 470 additions and 22 deletions
@@ -5,6 +5,7 @@ from rest_framework.test import APIClient
 from django.test import TestCase

 from apps.accounts.models import AdminAuditLog, Invitation, Team, TeamMember, User
+from apps.billing.models import CreditAccount, CreditLedger


 class AdminInvitationApiTests(TestCase):
@@ -850,3 +851,102 @@ class AdminGovernanceTests(TestCase):
        self.assertGreaterEqual(checks["teams_without_account"], 1)  # GovTeam 无 credit_account
        self.assertIn("totals", r.data)
        self.assertGreaterEqual(r.data["totals"]["projects"], 2)
+
+
+class AdminUserProvisioningTests(TestCase):
+    """后台直建用户 + 按用户发积分。积分账户仍挂 Team(计费链路不变),
+    新用户配一个 is_personal 的一人团队当专属钱包;老的多人团队用户发积分则进共享池。"""
+
+    def setUp(self):
+        self.admin = User.objects.create_user(username="padmin", password="x", is_platform_admin=True)
+        self.normal = User.objects.create_user(username="normal", password="x")
+        self.team = Team.objects.create(name="Legacy Team", owner=self.normal)
+        TeamMember.objects.create(team=self.team, user=self.normal, role=TeamMember.Role.OWNER)
+        self.mate = User.objects.create_user(username="mate", password="x")
+        TeamMember.objects.create(team=self.team, user=self.mate, role=TeamMember.Role.MEMBER)
+        CreditAccount.objects.create(team=self.team, balance=Decimal("100"))
+        self.ac = APIClient()
+        self.ac.force_authenticate(self.admin)
+        self.nc = APIClient()
+        self.nc.force_authenticate(self.normal)
+
+    def _create(self, **overrides):
+        payload = {"username": "created-user", "password": "strong-pass-1", "initial_credits": "500"}
+        payload.update(overrides)
+        return self.ac.post("/api/admin/users/", payload, format="json")
+
+    def test_create_user_provisions_personal_wallet(self):
+        r = self._create()
+        self.assertEqual(r.status_code, 201)
+        self.assertEqual(r.data["balance"], "500.0000")
+        self.assertFalse(r.data["wallet_shared"])
+        self.assertEqual(r.data["teams"], [])  # 个人团队是钱包实现细节,不当「所属团队」露出
+
+        user = User.objects.get(username="created-user")
+        team = Team.objects.get(owner=user)
+        self.assertTrue(team.is_personal)
+        self.assertEqual(team.credit_account.balance, Decimal("500"))
+        ledger = CreditLedger.objects.get(team=team, ledger_type=CreditLedger.Type.RECHARGE)
+        self.assertEqual(ledger.metadata["target_user_id"], str(user.id))
+        self.assertTrue(AdminAuditLog.objects.filter(action="user.create").exists())
+
+    def test_created_user_can_login(self):
+        self._create()
+        r = APIClient().post("/api/auth/login/", {"username": "created-user", "password": "strong-pass-1"}, format="json")
+        self.assertEqual(r.status_code, 200)
+
+    def test_create_user_validation_and_permission(self):
+        self.assertEqual(self.nc.post("/api/admin/users/", {}, format="json").status_code, 403)
+        self.assertEqual(self._create(username="").status_code, 400)
+        self.assertEqual(self._create(password="short").status_code, 400)
+        self.assertEqual(self._create(initial_credits="-5").status_code, 400)
+        self.assertEqual(self._create(username="normal").status_code, 400)  # 重名
+        self.assertEqual(User.objects.filter(username="created-user").count(), 0)
+
+    def test_create_user_without_initial_credits(self):
+        r = self._create(initial_credits="")
+        self.assertEqual(r.status_code, 201)
+        self.assertEqual(r.data["balance"], "0.0000")
+        team = Team.objects.get(owner__username="created-user")
+        self.assertFalse(CreditLedger.objects.filter(team=team).exists())  # 0 积分不落空流水
+
+    def test_adjust_user_credit(self):
+        user_id = self._create().data["id"]
+        r = self.ac.post(f"/api/admin/users/{user_id}/credits/", {"amount": "250", "reason": "活动赠送"}, format="json")
+        self.assertEqual(r.status_code, 201)
+        self.assertEqual(r.data["balance"], "750.0000")
+        ledger = CreditLedger.objects.filter(ledger_type=CreditLedger.Type.ADJUSTMENT).latest("created_at")
+        self.assertEqual(ledger.reason, "活动赠送")
+        self.assertEqual(ledger.metadata["target_user_id"], user_id)
+        self.assertTrue(AdminAuditLog.objects.filter(action="user.credit_adjust").exists())
+
+        # 扣减合法,但不能扣成负数
+        self.assertEqual(self.ac.post(f"/api/admin/users/{user_id}/credits/", {"amount": "-50"}, format="json").data["balance"], "700.0000")
+        self.assertEqual(self.ac.post(f"/api/admin/users/{user_id}/credits/", {"amount": "-9999"}, format="json").status_code, 400)
+        self.assertEqual(self.ac.post(f"/api/admin/users/{user_id}/credits/", {"amount": "0"}, format="json").status_code, 400)
+        self.assertEqual(self.nc.post(f"/api/admin/users/{user_id}/credits/", {"amount": "1"}, format="json").status_code, 403)
+
+    def test_legacy_multi_member_user_uses_shared_pool(self):
+        r = self.ac.get("/api/admin/users/?search=mate")
+        row = next(u for u in r.data["results"] if u["username"] == "mate")
+        self.assertEqual(row["balance"], "100.0000")
+        self.assertTrue(row["wallet_shared"])
+        self.assertEqual(row["wallet_team_name"], "Legacy Team")
+
+        self.ac.post(f"/api/admin/users/{row['id']}/credits/", {"amount": "40"}, format="json")
+        self.team.credit_account.refresh_from_db()
+        self.assertEqual(self.team.credit_account.balance, Decimal("140"))  # 进的是团队共享池
+
+    def test_platform_admin_without_team_has_no_wallet(self):
+        r = self.ac.get("/api/admin/users/?search=padmin")
+        row = next(u for u in r.data["results"] if u["username"] == "padmin")
+        self.assertIsNone(row["wallet_team"])
+        self.assertEqual(row["balance"], "0")
+
+    def test_personal_teams_hidden_from_team_list(self):
+        self._create()
+        names = {t["name"] for t in self.ac.get("/api/admin/teams/").data["results"]}
+        self.assertIn("Legacy Team", names)
+        self.assertNotIn("created-user", names)
+        all_names = {t["name"] for t in self.ac.get("/api/admin/teams/?include_personal=1").data["results"]}
+        self.assertIn("created-user", all_names)