seaislee1209andClaude Opus 4.8
01d4f235ed
后端:apps/ai QualityWord(stage/slot/text/sort/enabled)+ migration;services.py quality_words/quality_suffix, 4 个 builder 接入(person/model_tryon/storyboard/video)无配置回落原写死值零回归;adminpanel CRUD + 审计。 前端:adminApi quality 系列;Admin 质量词页(按 stage 分组卡 + 编辑模式开关 + 行内增改删 chip)。 测试:adminpanel 23 + accounts 22 = 45 单测过(CRUD/权限/生成读配置+回落/停用词不取); 无头 e2e _admin-p4.mjs 4 断言过 + 0 console error + live 库零残留;tsc+build 绿。 注:apps.ai 3 个 image_edit 路由测试失败为进场前既有(已 stash 比对 HEAD 同样失败),非本 Phase 引入。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
235 lines
12 KiB
Python
235 lines
12 KiB
Python
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)
|