erpnew/api_core/tests.py

from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group, Permission
from django.contrib.contenttypes.models import ContentType
from django.test import TestCase
from rest_framework.test import APIClient

from basic_info import models as basic_models


User = get_user_model()


class CoreAPITestCase(TestCase):
    def setUp(self):
        self.client = APIClient()
        self.merchant = basic_models.Merchant.objects.create(
            name="核心商户",
            type=basic_models.MerchantTypeEnum.STORE,
        )
        self.other_merchant = basic_models.Merchant.objects.create(
            name="其他商户",
            type=basic_models.MerchantTypeEnum.STORE,
        )
        self.super_user = User.objects.create_superuser(
            username="core_admin",
            password="pass12345",
            email="core@example.com",
        )
        self.employee = basic_models.Employee.objects.create(
            merchant=self.merchant,
            sys_user=self.super_user,
            name="核心管理员",
        )
        self.normal_user = User.objects.create_user(username="normal", password="pass12345")
        self.no_employee_super_user = User.objects.create_superuser(
            username="no_employee_admin",
            password="pass12345",
        )

    def authenticate_super_admin(self):
        self.client.force_authenticate(user=self.super_user)

    def test_core_api_requires_superuser_and_employee_merchant(self):
        response = self.client.get("/api/core/users/")
        self.assertEqual(response.status_code, 401)

        self.client.force_authenticate(user=self.normal_user)
        response = self.client.get("/api/core/users/")
        self.assertEqual(response.status_code, 403)

        self.client.force_authenticate(user=self.no_employee_super_user)
        response = self.client.get("/api/core/users/")
        self.assertEqual(response.status_code, 403)

    def test_user_list_is_merchant_scoped(self):
        same_user = User.objects.create_user(username="same_merchant", password="pass12345")
        basic_models.UserProfile.objects.create(user=same_user, merchant=self.merchant)

        other_user = User.objects.create_user(username="other_merchant", password="pass12345")
        basic_models.UserProfile.objects.create(user=other_user, merchant=self.other_merchant)

        self.authenticate_super_admin()
        response = self.client.get("/api/core/users/", {"limit": 100})
        self.assertEqual(response.status_code, 200)
        usernames = {item["username"] for item in response.data["results"]}
        self.assertIn("core_admin", usernames)
        self.assertIn("same_merchant", usernames)
        self.assertNotIn("other_merchant", usernames)

        detail_response = self.client.get(f"/api/core/users/{other_user.id}/")
        self.assertEqual(detail_response.status_code, 404)

    def test_create_user_creates_profile_but_not_employee(self):
        group = Group.objects.create(name="核心角色")
        self.authenticate_super_admin()
        response = self.client.post(
            "/api/core/users/",
            {
                "username": "core_created",
                "password": "pass12345",
                "email": "created@example.com",
                "first_name": "Core",
                "is_active": True,
                "groups": [group.id],
            },
            format="json",
        )
        self.assertEqual(response.status_code, 201)

        user = User.objects.get(username="core_created")
        self.assertEqual(user.profile.merchant, self.merchant)
        self.assertFalse(hasattr(user, "employee"))
        self.assertIn(group, user.groups.all())

    def test_user_cannot_be_deleted_and_self_cannot_be_deactivated_or_demoted(self):
        self.authenticate_super_admin()
        response = self.client.delete(f"/api/core/users/{self.super_user.id}/")
        self.assertEqual(response.status_code, 405)

        response = self.client.patch(
            f"/api/core/users/{self.super_user.id}/",
            {"is_active": False},
            format="json",
        )
        self.assertEqual(response.status_code, 400)
        self.assertIn("is_active", response.data)

        response = self.client.patch(
            f"/api/core/users/{self.super_user.id}/",
            {"is_superuser": False},
            format="json",
        )
        self.assertEqual(response.status_code, 400)
        self.assertIn("is_superuser", response.data)

    def test_groups_are_global_full_crud(self):
        content_type = ContentType.objects.get_for_model(User)
        permission = Permission.objects.get(content_type=content_type, codename="view_user")
        self.authenticate_super_admin()

        create_response = self.client.post(
            "/api/core/groups/",
            {"name": "全局核心组", "permissions": [permission.id]},
            format="json",
        )
        self.assertEqual(create_response.status_code, 201)
        group_id = create_response.data["id"]

        detail_response = self.client.get(f"/api/core/groups/{group_id}/")
        self.assertEqual(detail_response.status_code, 200)
        self.assertEqual(detail_response.data["name"], "全局核心组")
        self.assertEqual(detail_response.data["permissions"], [permission.id])

        patch_response = self.client.patch(
            f"/api/core/groups/{group_id}/",
            {"name": "已更新核心组"},
            format="json",
        )
        self.assertEqual(patch_response.status_code, 200)
        self.assertEqual(patch_response.data["name"], "已更新核心组")

        delete_response = self.client.delete(f"/api/core/groups/{group_id}/")
        self.assertEqual(delete_response.status_code, 204)
        self.assertFalse(Group.objects.filter(id=group_id).exists())

    def test_permissions_are_read_only(self):
        content_type = ContentType.objects.get_for_model(User)
        permission = Permission.objects.get(content_type=content_type, codename="view_user")
        self.authenticate_super_admin()

        list_response = self.client.get("/api/core/permissions/", {"limit": 100})
        self.assertEqual(list_response.status_code, 200)
        self.assertTrue(any(item["id"] == permission.id for item in list_response.data["results"]))

        detail_response = self.client.get(f"/api/core/permissions/{permission.id}/")
        self.assertEqual(detail_response.status_code, 200)
        self.assertEqual(detail_response.data["full_code"], "auth.view_user")

        create_response = self.client.post(
            "/api/core/permissions/",
            {"name": "fake", "codename": "fake_permission", "content_type": content_type.id},
            format="json",
        )
        self.assertEqual(create_response.status_code, 405)

        patch_response = self.client.patch(
            f"/api/core/permissions/{permission.id}/",
            {"name": "renamed"},
            format="json",
        )
        self.assertEqual(patch_response.status_code, 405)

        delete_response = self.client.delete(f"/api/core/permissions/{permission.id}/")
        self.assertEqual(delete_response.status_code, 405)