from decimal import Decimal from django.conf import settings from django.test import TestCase from rest_framework.test import APIClient from apps.accounts.models import Team, TeamMember, User from apps.billing.models import CreditAccount, CreditLedger class AuthApiTests(TestCase): def test_register_creates_user_team_member_credit_account_and_token(self): client = APIClient() response = client.post( "/api/auth/register/", { "username": "new-owner", "password": "strong-password", "email": "owner@example.com", "team_name": "Launch Team", }, format="json", ) self.assertEqual(response.status_code, 201) self.assertIn("token", response.data) user = User.objects.get(username="new-owner") team = Team.objects.get(name="Launch Team") self.assertEqual(team.owner, user) self.assertTrue(TeamMember.objects.filter(team=team, user=user, role=TeamMember.Role.OWNER).exists()) self.assertTrue(CreditAccount.objects.filter(team=team).exists()) def test_register_trial_grant_is_recorded_in_ledger(self): """开户赠送额度必须进流水:余额=赠送额,且有一条 balance_after=赠送额的 genesis 流水,账目从第一条起可对账。""" client = APIClient() client.post( "/api/auth/register/", {"username": "ledger-owner", "password": "strong-password", "team_name": "Ledger Team"}, format="json", ) team = Team.objects.get(name="Ledger Team") trial = Decimal(str(settings.DEFAULT_TRIAL_CREDITS)) account = CreditAccount.objects.get(team=team) self.assertEqual(account.balance, trial) genesis = CreditLedger.objects.filter(team=team, ledger_type=CreditLedger.Type.RECHARGE) self.assertEqual(genesis.count(), 1) self.assertEqual(genesis.first().amount, trial) self.assertEqual(genesis.first().balance_after, trial) # 流水终点 == 账户余额,可对账 class InvitationFlowTests(TestCase): def _register(self, client, username, **extra): return client.post( "/api/auth/register/", {"username": username, "password": "strong-password", **extra}, format="json", ) def test_invite_join_same_team_no_duplicate_credits(self): from apps.accounts.models import Invitation owner_client = APIClient() r = self._register(owner_client, "inv-owner", team_name="Owner Team") self.assertEqual(r.status_code, 201) team = Team.objects.get(name="Owner Team") accounts_before = CreditAccount.objects.filter(team=team).count() ledgers_before = CreditLedger.objects.filter(team=team).count() # 超管生成邀请码 owner_client.credentials(HTTP_AUTHORIZATION=f"Token {r.data['token']}") gen = owner_client.post( "/api/auth/team/invitations/", {"role": "member", "monthly_credit_limit": "500"}, format="json", ) self.assertEqual(gen.status_code, 201) code = gen.data["code"] self.assertTrue(code.startswith("TEAM-")) # 新用户凭码注册 r2 = self._register(APIClient(), "inv-joiner", invite_code=code) self.assertEqual(r2.status_code, 201) joiner = User.objects.get(username="inv-joiner") # 进的是同一团队、角色对、active、额度对 member = TeamMember.objects.get(team=team, user=joiner) self.assertEqual(member.role, TeamMember.Role.MEMBER) self.assertEqual(member.status, TeamMember.Status.ACTIVE) self.assertEqual(member.monthly_credit_limit, Decimal("500")) # 没开新团队 self.assertFalse(Team.objects.filter(owner=joiner).exists()) # 关键:不重复发额度 —— 团队 CreditAccount/ledger 数量不变 self.assertEqual(CreditAccount.objects.filter(team=team).count(), accounts_before) self.assertEqual(CreditLedger.objects.filter(team=team).count(), ledgers_before) # 邀请标 used inv = Invitation.objects.get(code=code) self.assertEqual(inv.status, Invitation.Status.USED) self.assertEqual(inv.used_by, joiner) def test_register_with_unknown_invite_rejected(self): r = self._register(APIClient(), "bad-inv", invite_code="TEAM-ABCD-EFGH") self.assertEqual(r.status_code, 400) self.assertFalse(User.objects.filter(username="bad-inv").exists()) def test_invite_generate_requires_manage_permission(self): owner_client = APIClient() ro = self._register(owner_client, "perm-owner", team_name="Perm Team") owner_client.credentials(HTTP_AUTHORIZATION=f"Token {ro.data['token']}") gen = owner_client.post("/api/auth/team/invitations/", {"role": "member"}, format="json") code = gen.data["code"] joiner_client = APIClient() rj = self._register(joiner_client, "perm-member", invite_code=code) # 普通成员尝试生成邀请码 → 403 joiner_client.credentials(HTTP_AUTHORIZATION=f"Token {rj.data['token']}") gen2 = joiner_client.post("/api/auth/team/invitations/", {"role": "member"}, format="json") self.assertEqual(gen2.status_code, 403) class PlatformAdminTests(TestCase): """Phase 0:平台超管基础 —— 管理命令 / 权限类 / 登录 me 返回标志且团队为空。""" def test_create_platform_admin_command(self): from django.core.management import call_command call_command("create_platform_admin") admin = User.objects.get(username="admin") self.assertTrue(admin.is_platform_admin) self.assertTrue(admin.is_staff) self.assertEqual(admin.status, User.Status.ACTIVE) self.assertTrue(admin.check_password("admin123")) def test_create_platform_admin_idempotent_no_password_reset(self): from django.core.management import call_command call_command("create_platform_admin", "--username", "root", "--password", "rootpass1") # 二次执行不带 --reset-password:不应覆盖已设密码 call_command("create_platform_admin", "--username", "root", "--password", "different2") admin = User.objects.get(username="root") self.assertTrue(admin.is_platform_admin) self.assertTrue(admin.check_password("rootpass1")) # 带 --reset-password 才重置 call_command("create_platform_admin", "--username", "root", "--password", "fresh3pass", "--reset-password") admin.refresh_from_db() self.assertTrue(admin.check_password("fresh3pass")) def test_login_returns_flag_and_null_team_for_teamless_admin(self): from django.core.management import call_command call_command("create_platform_admin") client = APIClient() r = client.post("/api/auth/login/", {"username": "admin", "password": "admin123"}, format="json") self.assertEqual(r.status_code, 200) self.assertTrue(r.data["user"]["is_platform_admin"]) self.assertIsNone(r.data["team"]) def test_me_returns_platform_admin_flag(self): from django.core.management import call_command call_command("create_platform_admin") client = APIClient() login = client.post("/api/auth/login/", {"username": "admin", "password": "admin123"}, format="json") client.credentials(HTTP_AUTHORIZATION=f"Token {login.data['token']}") me = client.get("/api/auth/me/") self.assertEqual(me.status_code, 200) self.assertTrue(me.data["user"]["is_platform_admin"]) self.assertIsNone(me.data["team"]) def test_is_platform_admin_permission(self): from rest_framework.test import APIRequestFactory from apps.accounts.permissions import IsPlatformAdmin admin = User.objects.create_user(username="padmin", password="strong-pass-1", is_platform_admin=True) normal = User.objects.create_user(username="normal-user", password="strong-pass-1") perm = IsPlatformAdmin() factory = APIRequestFactory() request = factory.get("/api/admin/ping/") request.user = admin self.assertTrue(perm.has_permission(request, None)) request.user = normal self.assertFalse(perm.has_permission(request, None)) # 被停用的超管也拒绝 admin.status = User.Status.DISABLED admin.save(update_fields=["status"]) request.user = admin self.assertFalse(perm.has_permission(request, None)) def test_normal_register_user_is_not_platform_admin(self): client = APIClient() r = client.post( "/api/auth/register/", {"username": "regular-owner", "password": "strong-password", "team_name": "Reg Team"}, format="json", ) self.assertEqual(r.status_code, 201) self.assertFalse(r.data["user"]["is_platform_admin"])