from unittest.mock import patch from rest_framework.test import APIClient from django.test import TestCase from apps.accounts.models import AdminAuditLog, Invitation, Team, TeamMember, User class AdminInvitationApiTests(TestCase): """Phase 2:平台超管邀请码发放管理(发开团队码 / 列跨团队码 / 撤销)+ 权限 + 审计。""" 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="Normal Team", owner=self.normal) TeamMember.objects.create(team=self.team, user=self.normal, role=TeamMember.Role.OWNER) self.admin_client = APIClient() self.admin_client.force_authenticate(self.admin) self.normal_client = APIClient() self.normal_client.force_authenticate(self.normal) def test_list_requires_platform_admin(self): self.assertEqual(self.normal_client.get("/api/admin/invitations/").status_code, 403) self.assertEqual(self.admin_client.get("/api/admin/invitations/").status_code, 200) def test_create_create_team_code_and_audit(self): r = self.admin_client.post("/api/admin/invitations/", {"email": "x@y.z"}, format="json") self.assertEqual(r.status_code, 201) self.assertEqual(r.data["kind"], Invitation.Kind.CREATE_TEAM) self.assertTrue(r.data["code"].startswith("TEAM-")) self.assertIsNone(r.data["team"]) self.assertEqual(r.data["register_url"], f"/register?invite={r.data['code']}") self.assertTrue(AdminAuditLog.objects.filter(action="invite.issue_create_team").exists()) def test_create_requires_platform_admin(self): self.assertEqual(self.normal_client.post("/api/admin/invitations/", {}, format="json").status_code, 403) def test_list_is_cross_team(self): self.admin_client.post("/api/admin/invitations/", {}, format="json") Invitation.objects.create(kind=Invitation.Kind.JOIN_TEAM, team=self.team, role=TeamMember.Role.MEMBER) r = self.admin_client.get("/api/admin/invitations/") self.assertEqual(r.status_code, 200) self.assertGreaterEqual(r.data["count"], 2) kinds = {i["kind"] for i in r.data["results"]} self.assertIn(Invitation.Kind.CREATE_TEAM, kinds) self.assertIn(Invitation.Kind.JOIN_TEAM, kinds) def test_filter_by_kind(self): self.admin_client.post("/api/admin/invitations/", {}, format="json") Invitation.objects.create(kind=Invitation.Kind.JOIN_TEAM, team=self.team, role=TeamMember.Role.MEMBER) r = self.admin_client.get("/api/admin/invitations/?kind=create_team") self.assertTrue(all(i["kind"] == Invitation.Kind.CREATE_TEAM for i in r.data["results"])) def test_revoke_and_audit(self): created = self.admin_client.post("/api/admin/invitations/", {}, format="json") inv_id = created.data["id"] r = self.admin_client.post(f"/api/admin/invitations/{inv_id}/revoke/") self.assertEqual(r.status_code, 200) self.assertEqual(r.data["status"], Invitation.Status.REVOKED) self.assertTrue(AdminAuditLog.objects.filter(action="invite.revoke").exists()) def test_revoke_requires_platform_admin(self): created = self.admin_client.post("/api/admin/invitations/", {}, format="json") inv_id = created.data["id"] self.assertEqual(self.normal_client.post(f"/api/admin/invitations/{inv_id}/revoke/").status_code, 403) def test_issued_create_team_code_registers_new_team(self): created = self.admin_client.post("/api/admin/invitations/", {}, format="json") code = created.data["code"] reg = APIClient().post( "/api/auth/register/", {"username": "new-via-admin-code", "password": "strong-pass-1", "team_name": "Via Admin", "invite_code": code}, format="json", ) self.assertEqual(reg.status_code, 201) self.assertTrue(Team.objects.filter(name="Via Admin").exists()) # 码标 used self.assertEqual(Invitation.objects.get(code=code).status, Invitation.Status.USED) def test_revoked_code_cannot_register(self): created = self.admin_client.post("/api/admin/invitations/", {}, format="json") code = created.data["code"] self.admin_client.post(f"/api/admin/invitations/{created.data['id']}/revoke/") reg = APIClient().post( "/api/auth/register/", {"username": "blocked-user", "password": "strong-pass-1", "team_name": "Blocked", "invite_code": code}, format="json", ) self.assertEqual(reg.status_code, 400) class AdminTeamUserApiTests(TestCase): """Phase 3:平台团队 + 用户管理(列/详情/启停/强制改密)+ 权限 + 团队停用拦登录。""" def setUp(self): from apps.billing.models import CreditAccount self.admin = User.objects.create_user(username="padmin3", password="x", is_platform_admin=True) self.owner = User.objects.create_user(username="t-owner", password="ownerpass1") self.member = User.objects.create_user(username="t-member", password="memberpass1") self.team = Team.objects.create(name="Acme", owner=self.owner) TeamMember.objects.create(team=self.team, user=self.owner, role=TeamMember.Role.OWNER) TeamMember.objects.create(team=self.team, user=self.member, role=TeamMember.Role.MEMBER) CreditAccount.objects.create(team=self.team, balance="123.4500") self.ac = APIClient() self.ac.force_authenticate(self.admin) self.normal = APIClient() self.normal.force_authenticate(self.owner) def _login(self, username, password): return APIClient().post("/api/auth/login/", {"username": username, "password": password}, format="json") def test_teams_list_permission_and_fields(self): self.assertEqual(self.normal.get("/api/admin/teams/").status_code, 403) r = self.ac.get("/api/admin/teams/") self.assertEqual(r.status_code, 200) row = next(t for t in r.data["results"] if t["name"] == "Acme") self.assertEqual(row["member_count"], 2) self.assertEqual(row["owner_username"], "t-owner") self.assertEqual(row["balance"], "123.4500") def test_team_detail_has_members(self): r = self.ac.get(f"/api/admin/teams/{self.team.id}/") self.assertEqual(r.status_code, 200) self.assertEqual(len(r.data["members"]), 2) def test_team_toggle_blocks_then_restores_member_login(self): self.assertEqual(self._login("t-member", "memberpass1").status_code, 200) tr = self.ac.post(f"/api/admin/teams/{self.team.id}/toggle/") self.assertEqual(tr.status_code, 200) self.assertEqual(tr.data["status"], Team.Status.DISABLED) self.assertEqual(self._login("t-member", "memberpass1").status_code, 400) # 停用后被拒 self.ac.post(f"/api/admin/teams/{self.team.id}/toggle/") # 恢复 self.assertEqual(self._login("t-member", "memberpass1").status_code, 200) self.assertTrue(AdminAuditLog.objects.filter(action="team.toggle_status").exists()) def test_team_toggle_requires_admin(self): self.assertEqual(self.normal.post(f"/api/admin/teams/{self.team.id}/toggle/").status_code, 403) def test_users_list_with_teams(self): self.assertEqual(self.normal.get("/api/admin/users/").status_code, 403) r = self.ac.get("/api/admin/users/") self.assertEqual(r.status_code, 200) member_row = next(u for u in r.data["results"] if u["username"] == "t-member") self.assertEqual(member_row["teams"][0]["team_name"], "Acme") def test_user_toggle_blocks_login(self): r = self.ac.post(f"/api/admin/users/{self.member.id}/toggle/") self.assertEqual(r.status_code, 200) self.assertEqual(r.data["status"], User.Status.DISABLED) self.assertEqual(self._login("t-member", "memberpass1").status_code, 400) self.assertTrue(AdminAuditLog.objects.filter(action="user.toggle_status").exists()) def test_cannot_disable_platform_admin(self): self.assertEqual(self.ac.post(f"/api/admin/users/{self.admin.id}/toggle/").status_code, 400) def test_reset_password(self): r = self.ac.post(f"/api/admin/users/{self.member.id}/reset-password/", {"password": "newpass789"}, format="json") self.assertEqual(r.status_code, 204) self.assertEqual(self._login("t-member", "newpass789").status_code, 200) self.assertEqual(self._login("t-member", "memberpass1").status_code, 400) # 旧密码失效 self.assertTrue(AdminAuditLog.objects.filter(action="user.reset_password").exists()) def test_reset_password_too_short_rejected(self): self.assertEqual( self.ac.post(f"/api/admin/users/{self.member.id}/reset-password/", {"password": "short"}, format="json").status_code, 400, ) def test_user_actions_require_admin(self): self.assertEqual(self.normal.post(f"/api/admin/users/{self.member.id}/toggle/").status_code, 403) self.assertEqual( self.normal.post(f"/api/admin/users/{self.member.id}/reset-password/", {"password": "newpass789"}, format="json").status_code, 403, ) class AdminQualityWordTests(TestCase): """Phase 4:质量词 CRUD + 权限 + 生成侧读配置(无配置回落写死值,零回归)。""" def setUp(self): self.admin = User.objects.create_user(username="padmin4", password="x", is_platform_admin=True) self.normal = User.objects.create_user(username="normal4", password="x") self.ac = APIClient() self.ac.force_authenticate(self.admin) self.nc = APIClient() self.nc.force_authenticate(self.normal) def test_crud_and_permission_and_audit(self): self.assertEqual(self.nc.get("/api/admin/quality-words/").status_code, 403) self.assertEqual(self.nc.post("/api/admin/quality-words/", {"stage": "person", "text": "x"}, format="json").status_code, 403) r = self.ac.post("/api/admin/quality-words/", {"stage": "person", "text": "赛博朋克风"}, format="json") self.assertEqual(r.status_code, 201) wid = r.data["id"] self.assertTrue(AdminAuditLog.objects.filter(action="quality_word.create").exists()) lst = self.ac.get("/api/admin/quality-words/") self.assertEqual(lst.status_code, 200) self.assertTrue(any(w["text"] == "赛博朋克风" for w in lst.data)) up = self.ac.patch(f"/api/admin/quality-words/{wid}/", {"text": "国潮风"}, format="json") self.assertEqual(up.status_code, 200) self.assertEqual(up.data["text"], "国潮风") self.assertEqual(self.ac.delete(f"/api/admin/quality-words/{wid}/").status_code, 204) def test_invalid_stage_rejected(self): r = self.ac.post("/api/admin/quality-words/", {"stage": "bogus", "text": "x"}, format="json") self.assertEqual(r.status_code, 400) def test_generation_reads_config_with_fallback(self): from apps.ai.models import QualityWord from apps.ai.services import build_person_frontal_prompt # 无配置 → 回落写死值 self.assertIn("纯色背景", build_person_frontal_prompt("小姐姐")) # 配置后 → 生成用新词,描述仍保留 QualityWord.objects.create(stage="person", text="赛博朋克风", sort=0) QualityWord.objects.create(stage="person", text="霓虹光", sort=1) out = build_person_frontal_prompt("小姐姐") self.assertIn("赛博朋克风", out) self.assertIn("霓虹光", out) self.assertIn("小姐姐", out) self.assertNotIn("纯色背景", out) def test_disabled_word_not_used(self): from apps.ai.models import QualityWord from apps.ai.services import quality_words QualityWord.objects.create(stage="video", text="启用词", enabled=True) QualityWord.objects.create(stage="video", text="停用词", enabled=False) words = quality_words("video") self.assertIn("启用词", words) self.assertNotIn("停用词", words) class AdminAssetReviewTests(TestCase): """Phase 5:跨团队人像审核队列(列/筛/批量送审/轮询)+ 权限。火山调用全 mock。""" def setUp(self): from apps.assets.models import Asset self.admin = User.objects.create_user(username="padmin5", password="x", is_platform_admin=True) self.normal = User.objects.create_user(username="normal5", password="x") self.teamA = Team.objects.create(name="TeamA", owner=self.normal) TeamMember.objects.create(team=self.teamA, user=self.normal, role=TeamMember.Role.OWNER) self.user2 = User.objects.create_user(username="u2-5", password="x") self.teamB = Team.objects.create(name="TeamB", owner=self.user2) self.a_proc = Asset.objects.create(team=self.teamA, name="proc", asset_type="image", category=Asset.Category.PERSON, review_status="processing") self.a_active = Asset.objects.create(team=self.teamA, name="active", asset_type="image", category=Asset.Category.PERSON, review_status="active") self.a_failed = Asset.objects.create(team=self.teamB, name="failed", asset_type="image", category=Asset.Category.PERSON, review_status="failed") self.a_none = Asset.objects.create(team=self.teamA, name="none", asset_type="image", category=Asset.Category.PERSON, review_status="") # 非 person 资产不该进队列 Asset.objects.create(team=self.teamA, name="scene", asset_type="image", category=Asset.Category.SCENE, review_status="") self.ac = APIClient() self.ac.force_authenticate(self.admin) self.nc = APIClient() self.nc.force_authenticate(self.normal) def test_list_permission_and_only_person_cross_team(self): self.assertEqual(self.nc.get("/api/admin/asset-reviews/").status_code, 403) r = self.ac.get("/api/admin/asset-reviews/") self.assertEqual(r.status_code, 200) self.assertEqual(r.data["count"], 4) # 4 person 资产(跨 2 团队),scene 排除 self.assertEqual({row["team_name"] for row in r.data["results"]}, {"TeamA", "TeamB"}) def test_filter_by_status(self): self.assertEqual(self.ac.get("/api/admin/asset-reviews/?review_status=processing").data["count"], 1) self.assertEqual(self.ac.get("/api/admin/asset-reviews/?review_status=active").data["count"], 1) self.assertEqual(self.ac.get("/api/admin/asset-reviews/?review_status=failed").data["count"], 1) self.assertEqual(self.ac.get("/api/admin/asset-reviews/?review_status=none").data["count"], 1) def test_submit_calls_review_and_audit(self): with patch("apps.adminpanel.views.submit_asset_for_review") as mock_submit: r = self.ac.post( "/api/admin/asset-reviews/submit/", {"asset_ids": [str(self.a_none.id), str(self.a_failed.id)]}, format="json", ) self.assertEqual(r.status_code, 200) self.assertEqual(r.data["submitted"], 2) self.assertEqual(mock_submit.call_count, 2) self.assertTrue(AdminAuditLog.objects.filter(action="asset_review.submit").exists()) def test_submit_requires_admin(self): with patch("apps.adminpanel.views.submit_asset_for_review") as mock_submit: r = self.nc.post("/api/admin/asset-reviews/submit/", {"asset_ids": [str(self.a_none.id)]}, format="json") self.assertEqual(r.status_code, 403) mock_submit.assert_not_called() def test_poll_only_processing(self): with patch("apps.adminpanel.views.poll_asset_review", return_value="active") as mock_poll: r = self.ac.post("/api/admin/asset-reviews/poll/", {}, format="json") self.assertEqual(r.status_code, 200) self.assertEqual(r.data["polled"], 1) # 仅 a_proc 处于 processing mock_poll.assert_called_once()