"""Cost 模块 API 测试""" from datetime import date from decimal import Decimal from django.contrib.auth import get_user_model from django.test import TestCase from rest_framework.test import APIClient from basic_info import models as basic_models from cost import models as cost_models class CostAPITest(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.user = get_user_model().objects.create_user(username='cost-api-user', password='pass12345') self.other_user = get_user_model().objects.create_user(username='cost-api-other', password='pass12345') self.employee = basic_models.Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='成本员工', ) self.other_employee = basic_models.Employee.objects.create( merchant=self.other_merchant, sys_user=self.other_user, name='其他员工', ) self.client.force_authenticate(user=self.user) # ==================== Category API Tests ==================== def test_list_categories(self): cost_models.CostCategory.objects.create( merchant=self.merchant, unique_key='electricity', name='电费', ) cost_models.CostCategory.objects.create( merchant=self.merchant, unique_key='water', name='水费', ) resp = self.client.get('/api/v2/cost-categories/') self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data), 2) self.assertEqual(resp.data[0]['name'], '水费') # name 排序 def test_list_categories_filter_by_merchant_id(self): cost_models.CostCategory.objects.create( merchant=self.merchant, unique_key='elec', name='电费', ) cost_models.CostCategory.objects.create( merchant=self.other_merchant, unique_key='water', name='水费', ) resp = self.client.get(f'/api/v2/cost-categories/?merchant_id={self.merchant.id}') self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data), 1) self.assertEqual(resp.data[0]['name'], '电费') def test_create_category(self): resp = self.client.post('/api/v2/cost-categories/', { 'unique_key': 'electricity', 'name': '电费', 'description': '每月电费', }, format='json') self.assertEqual(resp.status_code, 201) self.assertEqual(resp.data['unique_key'], 'electricity') self.assertEqual(resp.data['name'], '电费') self.assertEqual(resp.data['merchant_id'], self.merchant.id) def test_create_category_missing_required_field(self): resp = self.client.post('/api/v2/cost-categories/', { 'name': '电费', }, format='json') self.assertEqual(resp.status_code, 400) def test_get_category_detail(self): cat = cost_models.CostCategory.objects.create( merchant=self.merchant, unique_key='elec', name='电费', ) resp = self.client.get(f'/api/v2/cost-categories/{cat.id}/') self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['name'], '电费') def test_get_category_detail_404(self): resp = self.client.get('/api/v2/cost-categories/99999/') self.assertEqual(resp.status_code, 404) def test_get_category_detail_cross_merchant_404(self): """其他商户的类目返回 404""" cat = cost_models.CostCategory.objects.create( merchant=self.other_merchant, unique_key='elec', name='电费', ) resp = self.client.get(f'/api/v2/cost-categories/{cat.id}/') self.assertEqual(resp.status_code, 404) def test_update_category(self): cat = cost_models.CostCategory.objects.create( merchant=self.merchant, unique_key='elec', name='电费', ) resp = self.client.put(f'/api/v2/cost-categories/{cat.id}/', { 'name': '电力费', 'description': '已更新', }, format='json') self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['name'], '电力费') self.assertEqual(resp.data['description'], '已更新') def test_update_category_404(self): resp = self.client.put('/api/v2/cost-categories/99999/', { 'name': 'xxx', }, format='json') self.assertEqual(resp.status_code, 404) def test_delete_category(self): cat = cost_models.CostCategory.objects.create( merchant=self.merchant, unique_key='elec', name='电费', ) resp = self.client.delete(f'/api/v2/cost-categories/{cat.id}/') self.assertEqual(resp.status_code, 204) self.assertFalse(cost_models.CostCategory.objects.filter(id=cat.id).exists()) def test_delete_category_404(self): resp = self.client.delete('/api/v2/cost-categories/99999/') self.assertEqual(resp.status_code, 404) # ==================== Entry API Tests ==================== def _create_category(self, key='elec', name='电费'): return cost_models.CostCategory.objects.create( merchant=self.merchant, unique_key=key, name=name, ) def test_list_entries(self): cat = self._create_category() cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('100.00'), occurred_at=date(2026, 6, 1), operator=self.employee, ) cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('200.00'), occurred_at=date(2026, 6, 10), operator=self.employee, ) resp = self.client.get('/api/v2/cost-entries/') self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data), 2) self.assertEqual(resp.data[0]['amount'], '200.00') # 按 occurred_at 降序 def test_list_entries_filter_by_category(self): cat1 = self._create_category('elec', '电费') cat2 = self._create_category('water', '水费') cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat1, amount=Decimal('100.00'), occurred_at=date(2026, 6, 1), ) cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat2, amount=Decimal('200.00'), occurred_at=date(2026, 6, 10), ) resp = self.client.get(f'/api/v2/cost-entries/?category_id={cat1.id}') self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data), 1) self.assertEqual(resp.data[0]['amount'], '100.00') def test_list_entries_filter_by_date_range(self): cat = self._create_category() cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('100.00'), occurred_at=date(2026, 6, 1), ) cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('200.00'), occurred_at=date(2026, 7, 1), ) resp = self.client.get('/api/v2/cost-entries/?start=2026-06-15&end=2026-07-15') self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data), 1) self.assertEqual(resp.data[0]['amount'], '200.00') def test_list_entries_filter_by_merchant_id(self): cat = self._create_category() other_cat = cost_models.CostCategory.objects.create( merchant=self.other_merchant, unique_key='other', name='其他', ) cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('100.00'), occurred_at=date(2026, 6, 1), ) cost_models.CostEntry.objects.create( merchant=self.other_merchant, category=other_cat, amount=Decimal('999.00'), occurred_at=date(2026, 6, 1), ) resp = self.client.get(f'/api/v2/cost-entries/?merchant_id={self.merchant.id}') self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data), 1) def test_create_entry(self): cat = self._create_category() resp = self.client.post('/api/v2/cost-entries/', { 'category_id': cat.id, 'amount': '350.00', 'occurred_at': '2026-06-01', 'remarks': '六月份电费', }, format='json') self.assertEqual(resp.status_code, 201) self.assertEqual(resp.data['amount'], '350.00') self.assertEqual(resp.data['category_id'], cat.id) self.assertEqual(resp.data['category_name'], '电费') self.assertEqual(resp.data['operator_id'], self.employee.id) def test_create_formula_entry(self): cat = self._create_category('temp_worker', '临时工工资') resp = self.client.post('/api/v2/cost-entries/', { 'category_id': cat.id, 'unit_amount': '200.00', 'quantity': '3', 'unit_name': '人天', 'occurred_at': '2026-06-01', 'remarks': '临时工 3 人天', }, format='json') self.assertEqual(resp.status_code, 201) self.assertEqual(resp.data['amount'], '600.00') self.assertEqual(resp.data['unit_amount'], '200.0000') self.assertEqual(resp.data['quantity'], '3.0000') self.assertEqual(resp.data['unit_name'], '人天') def test_create_formula_entry_partial_fields_returns_400(self): cat = self._create_category('temp_worker', '临时工工资') resp = self.client.post('/api/v2/cost-entries/', { 'category_id': cat.id, 'unit_amount': '200.00', 'occurred_at': '2026-06-01', }, format='json') self.assertEqual(resp.status_code, 400) def test_create_entry_category_not_found(self): resp = self.client.post('/api/v2/cost-entries/', { 'category_id': 99999, 'amount': '100.00', 'occurred_at': '2026-06-01', }, format='json') self.assertEqual(resp.status_code, 404) def test_create_entry_cross_merchant_category_404(self): """不能用其他商户的类目创建""" other_cat = cost_models.CostCategory.objects.create( merchant=self.other_merchant, unique_key='other', name='其他', ) resp = self.client.post('/api/v2/cost-entries/', { 'category_id': other_cat.id, 'amount': '100.00', 'occurred_at': '2026-06-01', }, format='json') self.assertEqual(resp.status_code, 404) def test_create_entry_missing_required_field(self): cat = self._create_category() resp = self.client.post('/api/v2/cost-entries/', { 'category_id': cat.id, 'amount': '100.00', }, format='json') self.assertEqual(resp.status_code, 400) def test_get_entry_detail(self): cat = self._create_category() entry = cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('150.00'), occurred_at=date(2026, 6, 1), operator=self.employee, remarks='test', ) resp = self.client.get(f'/api/v2/cost-entries/{entry.id}/') self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['amount'], '150.00') self.assertEqual(resp.data['category_name'], '电费') def test_get_entry_detail_404(self): resp = self.client.get('/api/v2/cost-entries/99999/') self.assertEqual(resp.status_code, 404) def test_get_entry_detail_cross_merchant_404(self): other_cat = cost_models.CostCategory.objects.create( merchant=self.other_merchant, unique_key='other', name='其他', ) entry = cost_models.CostEntry.objects.create( merchant=self.other_merchant, category=other_cat, amount=Decimal('100.00'), occurred_at=date(2026, 6, 1), ) resp = self.client.get(f'/api/v2/cost-entries/{entry.id}/') self.assertEqual(resp.status_code, 404) def test_update_entry(self): cat = self._create_category() entry = cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('150.00'), occurred_at=date(2026, 6, 1), ) resp = self.client.put(f'/api/v2/cost-entries/{entry.id}/', { 'amount': '999.00', 'remarks': '已修改', }, format='json') self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['amount'], '999.00') self.assertEqual(resp.data['remarks'], '已修改') def test_update_formula_entry_recalculates_amount(self): cat = self._create_category('temp_worker', '临时工工资') entry = cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=None, unit_amount=Decimal('200.00'), quantity=Decimal('3'), unit_name='人天', occurred_at=date(2026, 6, 1), ) resp = self.client.put(f'/api/v2/cost-entries/{entry.id}/', { 'quantity': '4', }, format='json') self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['amount'], '800.00') self.assertEqual(resp.data['quantity'], '4.0000') entry.refresh_from_db() self.assertEqual(entry.amount, Decimal('800.00')) def test_update_formula_entry_to_manual_amount(self): cat = self._create_category('temp_worker', '临时工工资') entry = cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=None, unit_amount=Decimal('200.00'), quantity=Decimal('3'), unit_name='人天', occurred_at=date(2026, 6, 1), ) resp = self.client.put(f'/api/v2/cost-entries/{entry.id}/', { 'amount': '550.00', 'unit_amount': None, 'quantity': None, 'unit_name': '', }, format='json') self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['amount'], '550.00') self.assertIsNone(resp.data['unit_amount']) self.assertIsNone(resp.data['quantity']) def test_update_entry_change_category(self): cat1 = self._create_category('elec', '电费') cat2 = self._create_category('water', '水费') entry = cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat1, amount=Decimal('100.00'), occurred_at=date(2026, 6, 1), ) resp = self.client.put(f'/api/v2/cost-entries/{entry.id}/', { 'category_id': cat2.id, }, format='json') self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['category_id'], cat2.id) self.assertEqual(resp.data['category_name'], '水费') def test_update_entry_404(self): resp = self.client.put('/api/v2/cost-entries/99999/', { 'amount': '100.00', }, format='json') self.assertEqual(resp.status_code, 404) def test_delete_entry(self): cat = self._create_category() entry = cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('100.00'), occurred_at=date(2026, 6, 1), ) resp = self.client.delete(f'/api/v2/cost-entries/{entry.id}/') self.assertEqual(resp.status_code, 204) self.assertFalse(cost_models.CostEntry.objects.filter(id=entry.id).exists()) def test_delete_entry_404(self): resp = self.client.delete('/api/v2/cost-entries/99999/') self.assertEqual(resp.status_code, 404) # ==================== Summary API Tests ==================== def test_summary_by_category(self): cat1 = self._create_category('elec', '电费') cat2 = self._create_category('water', '水费') cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat1, amount=Decimal('300.00'), occurred_at=date(2026, 6, 1), ) cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat1, amount=Decimal('200.00'), occurred_at=date(2026, 6, 10), ) cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat2, amount=Decimal('80.00'), occurred_at=date(2026, 6, 5), ) resp = self.client.get('/api/v2/cost-summary/by-category/') self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data['results']), 2) self.assertEqual(resp.data['results'][0]['category_name'], '电费') self.assertEqual(resp.data['results'][0]['total_amount'], '500.00') self.assertEqual(resp.data['results'][0]['entry_count'], 2) self.assertEqual(resp.data['results'][1]['category_name'], '水费') self.assertEqual(resp.data['results'][1]['total_amount'], '80.00') def test_summary_by_category_with_date_range(self): cat = self._create_category() cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('100.00'), occurred_at=date(2026, 6, 1), ) cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('200.00'), occurred_at=date(2026, 7, 1), ) resp = self.client.get('/api/v2/cost-summary/by-category/?start=2026-06-15&end=2026-07-15') self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['results'][0]['total_amount'], '200.00') def test_summary_by_category_with_merchant_id(self): cat = self._create_category() other_cat = cost_models.CostCategory.objects.create( merchant=self.other_merchant, unique_key='other', name='其他', ) cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat, amount=Decimal('100.00'), occurred_at=date(2026, 6, 1), ) cost_models.CostEntry.objects.create( merchant=self.other_merchant, category=other_cat, amount=Decimal('9999.00'), occurred_at=date(2026, 6, 1), ) resp = self.client.get(f'/api/v2/cost-summary/by-category/?merchant_id={self.merchant.id}') self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data['results']), 1) self.assertEqual(resp.data['results'][0]['total_amount'], '100.00') def test_summary_by_category_no_data(self): resp = self.client.get('/api/v2/cost-summary/by-category/') self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['results'], []) def test_summary_includes_metadata(self): resp = self.client.get('/api/v2/cost-summary/by-category/') self.assertEqual(resp.status_code, 200) self.assertIn('merchant_id', resp.data) self.assertIn('start_date', resp.data) self.assertIn('end_date', resp.data) self.assertIn('results', resp.data) # ==================== Auth Tests ==================== def test_unauthenticated_returns_401(self): client = APIClient() resp = client.get('/api/v2/cost-categories/') self.assertEqual(resp.status_code, 401) def test_user_without_employee_raises_error(self): """用户未关联 Employee 时报错""" self.user = get_user_model().objects.create_user(username='noemp', password='pass12345') self.client.force_authenticate(user=self.user) resp = self.client.get('/api/v2/cost-categories/') self.assertEqual(resp.status_code, 400) self.assertIn('员工', str(resp.data)) def test_cross_merchant_isolation(self): """商户 A 用户看不到商户 B 的数据""" cat_a = self._create_category('elec', '电费') cost_models.CostEntry.objects.create( merchant=self.merchant, category=cat_a, amount=Decimal('100.00'), occurred_at=date(2026, 6, 1), ) cat_b = cost_models.CostCategory.objects.create( merchant=self.other_merchant, unique_key='water', name='水费', ) cost_models.CostEntry.objects.create( merchant=self.other_merchant, category=cat_b, amount=Decimal('999.00'), occurred_at=date(2026, 6, 1), ) # 当前用户属于 merchant，应该只看到自己的数据 resp = self.client.get('/api/v2/cost-entries/') self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data), 1) self.assertEqual(resp.data[0]['amount'], '100.00')