forked from erp-dev/erp
477 lines
20 KiB
Python
477 lines
20 KiB
Python
"""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') |