from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.test import TestCase from rest_framework import status from rest_framework.test import APIClient from basic_info import models as basic_models User = get_user_model() class CustomerAddressAPITestCase(TestCase): def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name="客户地址商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.other_merchant = basic_models.Merchant.objects.create( name="其它客户地址商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user(username="addr-user", password="testpass123") self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name="地址员工", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.other_user = User.objects.create_user( username="addr-other-user", password="testpass123" ) self.other_employee = basic_models.Employee.objects.create( sys_user=self.other_user, merchant=self.merchant, name="其它地址员工", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.foreign_user = User.objects.create_user( username="addr-foreign-user", password="testpass123" ) self.foreign_employee = basic_models.Employee.objects.create( sys_user=self.foreign_user, merchant=self.other_merchant, name="外部地址员工", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.visible_customer = basic_models.Customer.objects.create( merchant=self.merchant, name="可见客户", mobile="13800138000", area="杭州", created_by=self.employee, ) self.invisible_customer = basic_models.Customer.objects.create( merchant=self.merchant, name="不可见客户", mobile="13800138001", area="绍兴", created_by=self.other_employee, ) self.foreign_customer = basic_models.Customer.objects.create( merchant=self.other_merchant, name="外部客户", mobile="13800138002", area="苏州", created_by=self.foreign_employee, ) self.visible_address = basic_models.CustomerAddress.objects.create( merchant=self.merchant, customer=self.visible_customer, address="杭州市测试路1号", contact_name="张三", contact_phone="13800138000", area="杭州", coordinates="120.1,30.2", is_default=True, remark="一号仓", extra={"source": "test"}, created_by=self.employee, ) self.invisible_address = basic_models.CustomerAddress.objects.create( merchant=self.merchant, customer=self.invisible_customer, address="绍兴市测试路2号", contact_name="李四", contact_phone="13900139000", area="绍兴", created_by=self.other_employee, ) self.foreign_address = basic_models.CustomerAddress.objects.create( merchant=self.other_merchant, customer=self.foreign_customer, address="苏州市测试路3号", contact_name="王五", contact_phone="13700137000", area="苏州", created_by=self.foreign_employee, ) self.client.force_authenticate(user=self.user) def test_list_only_visible_customer_addresses(self): resp = self.client.get("/api/v1/customer-addresses/") self.assertEqual(resp.status_code, status.HTTP_200_OK) ids = {item["id"] for item in resp.json()["results"]} self.assertIn(self.visible_address.id, ids) self.assertNotIn(self.invisible_address.id, ids) self.assertNotIn(self.foreign_address.id, ids) def test_visible_employees_can_see_customer_address(self): self.invisible_customer.visible_employees.add(self.employee) resp = self.client.get("/api/v1/customer-addresses/") self.assertEqual(resp.status_code, status.HTTP_200_OK) ids = {item["id"] for item in resp.json()["results"]} self.assertIn(self.invisible_address.id, ids) def test_view_all_customers_permission_sees_same_merchant_addresses(self): permission = Permission.objects.get(codename="view_all_customers") self.user.user_permissions.add(permission) resp = self.client.get("/api/v1/customer-addresses/") self.assertEqual(resp.status_code, status.HTTP_200_OK) ids = {item["id"] for item in resp.json()["results"]} self.assertIn(self.visible_address.id, ids) self.assertIn(self.invisible_address.id, ids) self.assertNotIn(self.foreign_address.id, ids) def test_create_address_for_visible_customer_sets_merchant_and_creator(self): payload = { "customer": self.visible_customer.id, "address": "杭州市新增地址", "contact_name": "赵六", "contact_phone": "13600136000", "area": "杭州", "coordinates": "120.2,30.3", "is_default": False, "remark": "新增备注", "extra": {"tag": "new"}, } resp = self.client.post("/api/v1/customer-addresses/", payload, format="json") self.assertEqual(resp.status_code, status.HTTP_201_CREATED) data = resp.json() self.assertEqual(data["merchant"], self.merchant.id) self.assertEqual(data["created_by"], self.employee.id) self.assertEqual(data["coordinates"], "120.2,30.3") self.assertEqual(data["extra"], {"tag": "new"}) def test_create_address_for_invisible_customer_is_rejected(self): payload = { "customer": self.invisible_customer.id, "address": "不可见客户地址", } resp = self.client.post("/api/v1/customer-addresses/", payload, format="json") self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) def test_default_address_is_unique_per_customer(self): old_default = self.visible_address resp = self.client.post( "/api/v1/customer-addresses/", { "customer": self.visible_customer.id, "address": "新的默认地址", "is_default": True, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) old_default.refresh_from_db() self.assertFalse(old_default.is_default) self.assertTrue( basic_models.CustomerAddress.objects.get(id=resp.json()["id"]).is_default ) def test_set_default_address_endpoint_unsets_other_default_addresses(self): new_address = basic_models.CustomerAddress.objects.create( merchant=self.merchant, customer=self.visible_customer, address="待设默认地址", created_by=self.employee, ) resp = self.client.post( f"/api/v1/customer-address/{new_address.id}/set-default/" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.visible_address.refresh_from_db() new_address.refresh_from_db() self.assertFalse(self.visible_address.is_default) self.assertTrue(new_address.is_default) self.assertTrue(resp.json()["is_default"]) def test_delete_soft_deletes_address_and_query_deleted_records(self): resp = self.client.delete(f"/api/v1/customer-addresses/{self.visible_address.id}/") self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT) self.visible_address.refresh_from_db() self.assertIsNotNone(self.visible_address.deleted_at) self.assertEqual(self.visible_address.deleted_by, self.employee) self.assertFalse(self.visible_address.is_default) default_resp = self.client.get("/api/v1/customer-addresses/") default_ids = {item["id"] for item in default_resp.json()["results"]} self.assertNotIn(self.visible_address.id, default_ids) include_resp = self.client.get("/api/v1/customer-addresses/?include_deleted=true") include_ids = {item["id"] for item in include_resp.json()["results"]} self.assertIn(self.visible_address.id, include_ids) only_resp = self.client.get("/api/v1/customer-addresses/?only_deleted=true") only_ids = {item["id"] for item in only_resp.json()["results"]} self.assertEqual(only_ids, {self.visible_address.id}) def test_filters(self): query = ( f"/api/v1/customer-addresses/?customer={self.visible_customer.id}" "&customer_name=可见&address=测试路&contact_name=张" "&contact_phone=138&area=杭州&is_default=true&search=一号仓" ) resp = self.client.get(query) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["count"], 1) self.assertEqual(data["results"][0]["id"], self.visible_address.id) def test_retrieve_invisible_address_returns_404(self): resp = self.client.get(f"/api/v1/customer-addresses/{self.invisible_address.id}/") self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)