yingqing/core/backend/apps/accounts/tests.py

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)
        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):
        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 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"])