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 Invitation, Team, TeamMember, User from apps.billing.models import CreditAccount, CreditLedger def make_create_team_code(): """铸一个 create_team 邀请码(平台超管发的「开团队码」)。本期注册必须有码。""" return Invitation.objects.create(kind=Invitation.Kind.CREATE_TEAM).code 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", "invite_code": make_create_team_code(), }, 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", "invite_code": make_create_team_code(), }, 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) if trial > 0: # 若运营期把赠送额度调回 >0,genesis 凭证契约必须成立(I8) self.assertEqual(genesis.count(), 1) self.assertEqual(genesis.first().amount, trial) self.assertEqual(genesis.first().balance_after, trial) # 流水终点 == 账户余额,可对账 else: # 商业决策(2026-07-03):开户不赠送 → 余额 0 且不落赠送流水(I8 隐含余额 0 天然满足) self.assertEqual(trial, Decimal("0")) self.assertEqual(genesis.count(), 0) 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): owner_client = APIClient() r = self._register(owner_client, "inv-owner", team_name="Owner Team", invite_code=make_create_team_code()) 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", invite_code=make_create_team_code()) 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 MemberQuotaAndRoleTests(TestCase): """PMC#3:① 新建子账号不填限额时默认 100(不再 0=不限,避免子账号无上限花团队池); ② login/me 返回当前用户在团队的 role,前端据此只给主账号(owner)显示「团队」「消费」页。""" def _register(self, client, username, **extra): return client.post( "/api/auth/register/", {"username": username, "password": "strong-password", **extra}, format="json", ) def setUp(self): self.owner_client = APIClient() r = self._register(self.owner_client, "q-owner", team_name="Q Team", invite_code=make_create_team_code()) self.assertEqual(r.status_code, 201) self.token = r.data["token"] self.team = Team.objects.get(name="Q Team") self.owner_client.credentials(HTTP_AUTHORIZATION=f"Token {self.token}") def test_register_owner_payload_role_is_owner(self): # 注册即登录的返回体带 role=owner(主账号) r = self._register(APIClient(), "solo-owner", team_name="Solo", invite_code=make_create_team_code()) self.assertEqual(r.data.get("role"), "owner") def test_create_member_defaults_limit_1000(self): res = self.owner_client.post( "/api/auth/team/members/", {"username": "sub-a", "password": "strong-password", "role": "member"}, format="json", ) self.assertEqual(res.status_code, 201, res.content) member = TeamMember.objects.get(user__username="sub-a", team=self.team) # 积分制:默认月限额跟随 ×10(¥100 → 1000 积分),否则新成员限额实际缩水 10 倍 self.assertEqual(member.monthly_credit_limit, Decimal("1000")) def test_create_member_explicit_limit_respected(self): res = self.owner_client.post( "/api/auth/team/members/", {"username": "sub-b", "password": "strong-password", "role": "member", "monthly_credit_limit": 800}, format="json", ) self.assertEqual(res.status_code, 201, res.content) member = TeamMember.objects.get(user__username="sub-b", team=self.team) self.assertEqual(member.monthly_credit_limit, Decimal("800")) def test_member_login_returns_member_role(self): self.owner_client.post( "/api/auth/team/members/", {"username": "sub-c", "password": "strong-password", "role": "member"}, format="json", ) login = APIClient().post( "/api/auth/login/", {"username": "sub-c", "password": "strong-password"}, format="json", ) self.assertEqual(login.status_code, 200) self.assertEqual(login.data.get("role"), "member") 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", "invite_code": make_create_team_code(), }, format="json", ) self.assertEqual(r.status_code, 201) self.assertFalse(r.data["user"]["is_platform_admin"]) class RegisterInviteCodeTests(TestCase): """Phase 1:注册必须有码,kind 分流 create_team / join_team;取消「空码开团队」。""" def _register(self, username, **extra): return APIClient().post( "/api/auth/register/", {"username": username, "password": "strong-password", **extra}, format="json", ) def test_register_requires_invite_code(self): r = self._register("no-code-user", team_name="No Code Team") self.assertEqual(r.status_code, 400) self.assertIn("invite_code", r.data) self.assertFalse(User.objects.filter(username="no-code-user").exists()) self.assertFalse(Team.objects.filter(name="No Code Team").exists()) def test_register_blank_invite_code_rejected(self): r = self._register("blank-code-user", team_name="Blank", invite_code="") self.assertEqual(r.status_code, 400) self.assertFalse(User.objects.filter(username="blank-code-user").exists()) def test_create_team_code_opens_new_team_with_trial(self): code = make_create_team_code() r = self._register("ct-owner", team_name="Created Team", invite_code=code) self.assertEqual(r.status_code, 201) user = User.objects.get(username="ct-owner") team = Team.objects.get(name="Created Team") self.assertEqual(team.owner, user) self.assertTrue(TeamMember.objects.filter(team=team, user=user, role=TeamMember.Role.OWNER).exists()) # 试用额度发放 trial = Decimal(str(settings.DEFAULT_TRIAL_CREDITS)) self.assertEqual(CreditAccount.objects.get(team=team).balance, trial) # 邀请标 used + 回填团队 inv = Invitation.objects.get(code=code) self.assertEqual(inv.status, Invitation.Status.USED) self.assertEqual(inv.used_by, user) self.assertEqual(inv.team, team) def test_join_team_code_joins_existing_no_new_team(self): # 先用 create_team 码开团队 + 团管发 join 码 owner_client = APIClient() ro = owner_client.post( "/api/auth/register/", {"username": "jt-owner", "password": "strong-password", "team_name": "Host Team", "invite_code": make_create_team_code()}, format="json", ) owner_client.credentials(HTTP_AUTHORIZATION=f"Token {ro.data['token']}") gen = owner_client.post("/api/auth/team/invitations/", {"role": "member"}, format="json") self.assertEqual(gen.status_code, 201) self.assertEqual(gen.data["kind"], Invitation.Kind.JOIN_TEAM) code = gen.data["code"] host_team = Team.objects.get(name="Host Team") teams_before = Team.objects.count() r = self._register("jt-joiner", invite_code=code) self.assertEqual(r.status_code, 201) joiner = User.objects.get(username="jt-joiner") self.assertTrue(TeamMember.objects.filter(team=host_team, user=joiner).exists()) self.assertFalse(Team.objects.filter(owner=joiner).exists()) # 没开新团队 self.assertEqual(Team.objects.count(), teams_before) def test_register_with_used_code_rejected(self): code = make_create_team_code() self._register("first-use", team_name="T1", invite_code=code) r2 = self._register("second-use", team_name="T2", invite_code=code) self.assertEqual(r2.status_code, 400) self.assertFalse(User.objects.filter(username="second-use").exists()) def test_register_with_expired_code_rejected(self): from datetime import timedelta from django.utils import timezone inv = Invitation.objects.create(kind=Invitation.Kind.CREATE_TEAM) inv.expires_at = timezone.now() - timedelta(days=1) inv.save(update_fields=["expires_at"]) r = self._register("expired-user", team_name="Exp", invite_code=inv.code) self.assertEqual(r.status_code, 400) self.assertFalse(User.objects.filter(username="expired-user").exists()) class ValidateInviteTests(TestCase): """Phase 1:免登录验码端点 —— 注册页「填码 → 验过才展开」。""" def test_validate_create_team_code(self): code = make_create_team_code() r = APIClient().post("/api/auth/validate-invite/", {"code": code}, format="json") self.assertEqual(r.status_code, 200) self.assertTrue(r.data["valid"]) self.assertEqual(r.data["kind"], Invitation.Kind.CREATE_TEAM) self.assertIsNone(r.data["team_name"]) def test_validate_join_team_code_returns_team_name(self): owner_client = APIClient() ro = owner_client.post( "/api/auth/register/", {"username": "vi-owner", "password": "strong-password", "team_name": "Visible Team", "invite_code": make_create_team_code()}, format="json", ) 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"] r = APIClient().post("/api/auth/validate-invite/", {"code": code}, format="json") self.assertEqual(r.status_code, 200) self.assertTrue(r.data["valid"]) self.assertEqual(r.data["kind"], Invitation.Kind.JOIN_TEAM) self.assertEqual(r.data["team_name"], "Visible Team") def test_validate_unknown_code_invalid(self): r = APIClient().post("/api/auth/validate-invite/", {"code": "TEAM-ZZZZ-ZZZZ"}, format="json") self.assertEqual(r.status_code, 200) self.assertFalse(r.data["valid"]) def test_validate_blank_code_invalid(self): r = APIClient().post("/api/auth/validate-invite/", {"code": ""}, format="json") self.assertEqual(r.status_code, 200) self.assertFalse(r.data["valid"]) def test_validate_used_code_invalid(self): code = make_create_team_code() APIClient().post( "/api/auth/register/", {"username": "vi-consumer", "password": "strong-password", "team_name": "Consumed", "invite_code": code}, format="json", ) r = APIClient().post("/api/auth/validate-invite/", {"code": code}, format="json") self.assertEqual(r.status_code, 200) self.assertFalse(r.data["valid"]) class AvatarUploadTests(TestCase): """bug(9):上传新头像点「确认使用」弹 500。真因之一=对象存储上传异常裸抛 500; 现改为捕获后回 502 + 人话,成功路径回 200 + 新 avatar_url。""" def setUp(self): from rest_framework.authtoken.models import Token self.user = User.objects.create_user(username="av-user", password="strong-password") self.token = Token.objects.create(user=self.user) self.client = APIClient() self.client.credentials(HTTP_AUTHORIZATION=f"Token {self.token.key}") def _png(self): from django.core.files.uploadedfile import SimpleUploadedFile # 1x1 PNG 字节(够 multipart 上传,内容无所谓,storage 被 mock) png = ( b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00" b"\x01\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82" ) return SimpleUploadedFile("avatar.png", png, content_type="image/png") def test_upload_success_returns_200_and_sets_url(self): from unittest import mock with mock.patch("apps.assets.storage.TosStorage") as MockStorage: inst = MockStorage.return_value inst.upload_fileobj.return_value = None inst.public_url.return_value = "https://bucket.example.com/users/x/avatar/abc.png" r = self.client.post("/api/auth/me/avatar/", {"file": self._png()}, format="multipart") self.assertEqual(r.status_code, 200, r.data) self.user.refresh_from_db() self.assertEqual(self.user.avatar_url, "https://bucket.example.com/users/x/avatar/abc.png") def test_upload_storage_failure_returns_502_not_bare_500(self): from unittest import mock with mock.patch("apps.assets.storage.TosStorage") as MockStorage: MockStorage.return_value.upload_fileobj.side_effect = RuntimeError("TOS unreachable") r = self.client.post("/api/auth/me/avatar/", {"file": self._png()}, format="multipart") self.assertEqual(r.status_code, 502) self.assertIn("detail", r.data) def test_upload_without_file_returns_400(self): r = self.client.post("/api/auth/me/avatar/", {}, format="multipart") self.assertEqual(r.status_code, 400) class DeviceRevokeTests(TestCase): """bug(10):在用设备点「下线」提示成功但实际没下线。真因=只标记 LoginSession.revoked_at 没动 token(单 token 体系),被下线设备旧 token 仍有效。现改为旋转 token,旧 token 立即 401。""" def setUp(self): from rest_framework.authtoken.models import Token self.user = User.objects.create_user(username="dev-user", password="strong-password") # 模拟登录拿到 token(被下线的「目标设备」用的就是这个) self.token = Token.objects.create(user=self.user) def test_revoke_session_invalidates_old_token(self): from apps.accounts.models import LoginSession session = LoginSession.objects.create(user=self.user, user_agent="OtherDevice", ip_address="1.1.1.1") old_key = self.token.key client = APIClient() client.credentials(HTTP_AUTHORIZATION=f"Token {old_key}") r = client.post(f"/api/auth/me/sessions/{session.id}/revoke/") self.assertEqual(r.status_code, 200) # 必须真正吊销:旧 token 已失效,再用它访问受保护接口应 401 self.assertIn("token", r.data) self.assertNotEqual(r.data["token"], old_key) stale = APIClient() stale.credentials(HTTP_AUTHORIZATION=f"Token {old_key}") self.assertEqual(stale.get("/api/auth/me/").status_code, 401) # 会话被标记下线 session.refresh_from_db() self.assertIsNotNone(session.revoked_at) def test_revoke_unknown_session_no_op(self): import uuid as _uuid client = APIClient() client.credentials(HTTP_AUTHORIZATION=f"Token {self.token.key}") r = client.post(f"/api/auth/me/sessions/{_uuid.uuid4()}/revoke/") self.assertEqual(r.status_code, 200) self.assertEqual(r.data.get("revoked"), 0) class TeamSettingsTests(TestCase): """PMC#12:团队月限额持久化。设置后落库,刷新(重新 GET)仍在;成员不可改;清空回到「未设置」。""" def _register(self, client, username, **extra): return client.post( "/api/auth/register/", {"username": username, "password": "strong-password", **extra}, format="json", ) def setUp(self): self.owner_client = APIClient() r = self._register(self.owner_client, "ts-owner", team_name="TS Team", invite_code=make_create_team_code()) self.assertEqual(r.status_code, 201) self.team = Team.objects.get(name="TS Team") self.owner_client.credentials(HTTP_AUTHORIZATION=f"Token {r.data['token']}") # 注册返回体里团队月限额默认未设置(null) self.assertIsNone(r.data["team"].get("monthly_credit_limit")) def test_owner_sets_limit_persists(self): res = self.owner_client.patch( "/api/auth/team/settings/", {"monthly_credit_limit": 3000}, format="json" ) self.assertEqual(res.status_code, 200, res.content) self.assertEqual(Decimal(res.data["monthly_credit_limit"]), Decimal("3000")) # 落库 self.team.refresh_from_db() self.assertEqual(self.team.monthly_credit_limit, Decimal("3000")) # 「刷新」= 重新 GET,值仍在(PMC#12 核心) got = self.owner_client.get("/api/auth/team/settings/") self.assertEqual(Decimal(got.data["monthly_credit_limit"]), Decimal("3000")) def test_unlimited_minus_one_stored_as_is(self): res = self.owner_client.patch( "/api/auth/team/settings/", {"monthly_credit_limit": -1}, format="json" ) self.assertEqual(res.status_code, 200, res.content) self.team.refresh_from_db() self.assertEqual(self.team.monthly_credit_limit, Decimal("-1")) def test_empty_clears_to_unset(self): self.owner_client.patch("/api/auth/team/settings/", {"monthly_credit_limit": 3000}, format="json") res = self.owner_client.patch( "/api/auth/team/settings/", {"monthly_credit_limit": ""}, format="json" ) self.assertEqual(res.status_code, 200, res.content) self.team.refresh_from_db() self.assertIsNone(self.team.monthly_credit_limit) def test_invalid_number_rejected(self): res = self.owner_client.patch( "/api/auth/team/settings/", {"monthly_credit_limit": "abc"}, format="json" ) self.assertEqual(res.status_code, 400) def test_member_cannot_change_limit(self): self.owner_client.post( "/api/auth/team/members/", {"username": "ts-member", "password": "strong-password", "role": "member"}, format="json", ) member_client = APIClient() login = member_client.post( "/api/auth/login/", {"username": "ts-member", "password": "strong-password"}, format="json" ) member_client.credentials(HTTP_AUTHORIZATION=f"Token {login.data['token']}") # 成员可读 self.assertEqual(member_client.get("/api/auth/team/settings/").status_code, 200) # 但不可改 res = member_client.patch( "/api/auth/team/settings/", {"monthly_credit_limit": 9999}, format="json" ) self.assertEqual(res.status_code, 403) self.team.refresh_from_db() self.assertIsNone(self.team.monthly_credit_limit)