forked from erp-dev/erp
445 lines
18 KiB
Python
445 lines
18 KiB
Python
"""Cost 模块服务测试"""
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
from django.contrib.auth import get_user_model
|
||
from django.test import TestCase
|
||
|
||
from basic_info import models as basic_models
|
||
from cost import models as cost_models
|
||
from cost import services as cost_services
|
||
from cost.models import CostEntryInput, CostProviderPort
|
||
|
||
|
||
class EnsureCategoryTests(TestCase):
|
||
def setUp(self):
|
||
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,
|
||
)
|
||
|
||
def test_ensure_category_creates_new(self):
|
||
"""category_key 不存在时创建新类目"""
|
||
cat = cost_services.ensure_category(
|
||
merchant=self.merchant,
|
||
category_key='electricity',
|
||
category_name='电费',
|
||
description='每月电费支出',
|
||
)
|
||
self.assertEqual(cat.unique_key, 'electricity')
|
||
self.assertEqual(cat.name, '电费')
|
||
self.assertEqual(cat.description, '每月电费支出')
|
||
self.assertEqual(cat.merchant_id, self.merchant.id)
|
||
|
||
def test_ensure_category_returns_existing(self):
|
||
"""category_key 已存在时返回已有类目"""
|
||
existing = cost_models.CostCategory.objects.create(
|
||
merchant=self.merchant, unique_key='electricity', name='电费',
|
||
)
|
||
cat = cost_services.ensure_category(
|
||
merchant=self.merchant, category_key='electricity', category_name='电力',
|
||
)
|
||
self.assertEqual(cat.id, existing.id)
|
||
self.assertEqual(cat.name, '电费') # 不覆盖已有 name
|
||
|
||
def test_ensure_category_empty_key_raises(self):
|
||
"""空 category_key 抛 ValueError"""
|
||
with self.assertRaises(ValueError) as ctx:
|
||
cost_services.ensure_category(
|
||
merchant=self.merchant, category_key='', category_name='电费',
|
||
)
|
||
self.assertIn('category_key', str(ctx.exception))
|
||
|
||
def test_ensure_category_whitespace_key_raises(self):
|
||
"""纯空格 category_key 抛 ValueError"""
|
||
with self.assertRaises(ValueError) as ctx:
|
||
cost_services.ensure_category(
|
||
merchant=self.merchant, category_key=' ', category_name='电费',
|
||
)
|
||
self.assertIn('category_key', str(ctx.exception))
|
||
|
||
def test_ensure_category_empty_name_raises(self):
|
||
"""空 category_name 抛 ValueError"""
|
||
with self.assertRaises(ValueError) as ctx:
|
||
cost_services.ensure_category(
|
||
merchant=self.merchant, category_key='key', category_name='',
|
||
)
|
||
self.assertIn('category_name', str(ctx.exception))
|
||
|
||
def test_ensure_category_different_merchants_same_key(self):
|
||
"""不同商户相同 key 不冲突"""
|
||
cat1 = cost_services.ensure_category(
|
||
merchant=self.merchant, category_key='same_key', category_name='类目A',
|
||
)
|
||
cat2 = cost_services.ensure_category(
|
||
merchant=self.other_merchant, category_key='same_key', category_name='类目B',
|
||
)
|
||
self.assertNotEqual(cat1.id, cat2.id)
|
||
self.assertEqual(cat1.name, '类目A')
|
||
self.assertEqual(cat2.name, '类目B')
|
||
|
||
|
||
class CreateCostEntryTests(TestCase):
|
||
def setUp(self):
|
||
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='testuser', password='pass12345')
|
||
self.employee = basic_models.Employee.objects.create(
|
||
merchant=self.merchant, sys_user=self.user, name='测试员工',
|
||
)
|
||
self.other_user = get_user_model().objects.create_user(username='otheruser', password='pass12345')
|
||
self.other_employee = basic_models.Employee.objects.create(
|
||
merchant=self.other_merchant, sys_user=self.other_user, name='其他员工',
|
||
)
|
||
self.category = cost_models.CostCategory.objects.create(
|
||
merchant=self.merchant, unique_key='transport', name='运输费',
|
||
)
|
||
self.other_category = cost_models.CostCategory.objects.create(
|
||
merchant=self.other_merchant, unique_key='transport', name='运输费',
|
||
)
|
||
|
||
def test_create_cost_entry_normal(self):
|
||
"""正常创建支出记录"""
|
||
entry = cost_services.create_cost_entry(
|
||
merchant=self.merchant,
|
||
category=self.category,
|
||
amount=Decimal('150.00'),
|
||
occurred_at=date(2026, 6, 1),
|
||
operator=self.employee,
|
||
source_module='manual',
|
||
source_id='',
|
||
remarks='测试备注',
|
||
)
|
||
self.assertEqual(entry.amount, Decimal('150.00'))
|
||
self.assertEqual(entry.occurred_at, date(2026, 6, 1))
|
||
self.assertEqual(entry.operator_id, self.employee.id)
|
||
self.assertEqual(entry.merchant_id, self.merchant.id)
|
||
self.assertEqual(entry.source_module, 'manual')
|
||
self.assertEqual(entry.remarks, '测试备注')
|
||
|
||
def test_create_cost_entry_operator_none(self):
|
||
"""operator=None 正常创建（provider 场景）"""
|
||
entry = cost_services.create_cost_entry(
|
||
merchant=self.merchant,
|
||
category=self.category,
|
||
amount=Decimal('200.00'),
|
||
occurred_at=date(2026, 6, 2),
|
||
operator=None,
|
||
)
|
||
self.assertIsNone(entry.operator_id)
|
||
self.assertEqual(entry.amount, Decimal('200.00'))
|
||
|
||
def test_create_cost_entry_category_mismatch_merchant_raises(self):
|
||
"""category 所属商户与 merchant 不一致抛 ValueError"""
|
||
with self.assertRaises(ValueError) as ctx:
|
||
cost_services.create_cost_entry(
|
||
merchant=self.merchant,
|
||
category=self.other_category,
|
||
amount=Decimal('100.00'),
|
||
occurred_at=date(2026, 6, 1),
|
||
operator=self.employee,
|
||
)
|
||
self.assertIn('category', str(ctx.exception))
|
||
|
||
def test_create_cost_entry_operator_mismatch_merchant_raises(self):
|
||
"""operator 所属商户与 merchant 不一致抛 ValueError"""
|
||
with self.assertRaises(ValueError) as ctx:
|
||
cost_services.create_cost_entry(
|
||
merchant=self.merchant,
|
||
category=self.category,
|
||
amount=Decimal('100.00'),
|
||
occurred_at=date(2026, 6, 1),
|
||
operator=self.other_employee,
|
||
)
|
||
self.assertIn('operator', str(ctx.exception))
|
||
|
||
def test_create_cost_entry_default_values(self):
|
||
"""不传 source_module/source_id/remarks 时默认为空字符串"""
|
||
entry = cost_services.create_cost_entry(
|
||
merchant=self.merchant,
|
||
category=self.category,
|
||
amount=Decimal('50.00'),
|
||
occurred_at=date(2026, 6, 3),
|
||
)
|
||
self.assertEqual(entry.source_module, '')
|
||
self.assertEqual(entry.source_id, '')
|
||
self.assertEqual(entry.remarks, '')
|
||
|
||
|
||
class MockCostProvider:
|
||
"""测试用 Provider"""
|
||
category_key = 'test_provider'
|
||
|
||
def __init__(self, entries):
|
||
self._entries = entries
|
||
|
||
def get_cost_entries(self, *, merchant, start_date, end_date):
|
||
return self._entries
|
||
|
||
|
||
class CollectFromProviderTests(TestCase):
|
||
def setUp(self):
|
||
self.merchant = basic_models.Merchant.objects.create(
|
||
name='测试商户', type=basic_models.MerchantTypeEnum.STORE,
|
||
)
|
||
|
||
def test_collect_creates_entries(self):
|
||
"""正常采集并创建 CostEntry"""
|
||
entries = [
|
||
CostEntryInput(
|
||
category_key='electricity', category_name='电费',
|
||
amount=Decimal('300.00'), occurred_at=date(2026, 6, 1),
|
||
source_module='test', source_id='E001',
|
||
),
|
||
CostEntryInput(
|
||
category_key='water', category_name='水费',
|
||
amount=Decimal('80.00'), occurred_at=date(2026, 6, 2),
|
||
source_module='test', source_id='W001',
|
||
),
|
||
]
|
||
provider = MockCostProvider(entries)
|
||
|
||
result = cost_services.collect_from_provider(
|
||
provider=provider,
|
||
merchant=self.merchant,
|
||
start_date=date(2026, 6, 1),
|
||
end_date=date(2026, 6, 30),
|
||
)
|
||
|
||
self.assertEqual(result['total_seen'], 2)
|
||
self.assertEqual(result['created_count'], 2)
|
||
self.assertEqual(result['error_count'], 0)
|
||
self.assertEqual(len(result['errors']), 0)
|
||
self.assertEqual(result['provider_category_key'], 'test_provider')
|
||
|
||
# 验证数据库
|
||
self.assertEqual(cost_models.CostEntry.objects.count(), 2)
|
||
self.assertEqual(cost_models.CostCategory.objects.count(), 2)
|
||
|
||
def test_collect_returns_not_list_raises(self):
|
||
"""Provider 返回非 list 抛 TypeError"""
|
||
class BadProvider:
|
||
category_key = 'bad'
|
||
def get_cost_entries(self, *, merchant, start_date, end_date):
|
||
return 'not_a_list'
|
||
|
||
with self.assertRaises(TypeError):
|
||
cost_services.collect_from_provider(
|
||
provider=BadProvider(),
|
||
merchant=self.merchant,
|
||
start_date=date(2026, 6, 1),
|
||
end_date=date(2026, 6, 30),
|
||
)
|
||
|
||
def test_collect_non_entry_input_records_error(self):
|
||
"""非 CostEntryInput 的条目记入 errors"""
|
||
provider = MockCostProvider(['not_a_dataclass', 123])
|
||
|
||
result = cost_services.collect_from_provider(
|
||
provider=provider,
|
||
merchant=self.merchant,
|
||
start_date=date(2026, 6, 1),
|
||
end_date=date(2026, 6, 30),
|
||
)
|
||
|
||
self.assertEqual(result['total_seen'], 2)
|
||
self.assertEqual(result['created_count'], 0)
|
||
self.assertEqual(result['error_count'], 2)
|
||
self.assertEqual(cost_models.CostEntry.objects.count(), 0)
|
||
|
||
def test_collect_empty_category_key_records_error(self):
|
||
"""category_key 为空导致 ValueError 记入 errors"""
|
||
entries = [
|
||
CostEntryInput(
|
||
category_key='', category_name='无key类目',
|
||
amount=Decimal('100.00'), occurred_at=date(2026, 6, 1),
|
||
source_module='test', source_id='X001',
|
||
),
|
||
]
|
||
provider = MockCostProvider(entries)
|
||
|
||
result = cost_services.collect_from_provider(
|
||
provider=provider,
|
||
merchant=self.merchant,
|
||
start_date=date(2026, 6, 1),
|
||
end_date=date(2026, 6, 30),
|
||
)
|
||
|
||
self.assertEqual(result['total_seen'], 1)
|
||
self.assertEqual(result['created_count'], 0)
|
||
self.assertEqual(result['error_count'], 1)
|
||
self.assertIn('category_key', result['errors'][0]['error'])
|
||
|
||
def test_collect_empty_list(self):
|
||
"""空列表返回 created_count=0"""
|
||
provider = MockCostProvider([])
|
||
|
||
result = cost_services.collect_from_provider(
|
||
provider=provider,
|
||
merchant=self.merchant,
|
||
start_date=date(2026, 6, 1),
|
||
end_date=date(2026, 6, 30),
|
||
)
|
||
|
||
self.assertEqual(result['total_seen'], 0)
|
||
self.assertEqual(result['created_count'], 0)
|
||
self.assertEqual(result['error_count'], 0)
|
||
|
||
def test_collect_mixed_valid_invalid_partial_success(self):
|
||
"""混合合法+非法条目，合法条目正常创建"""
|
||
entries = [
|
||
CostEntryInput(
|
||
category_key='electricity', category_name='电费',
|
||
amount=Decimal('300.00'), occurred_at=date(2026, 6, 1),
|
||
source_module='test', source_id='E001',
|
||
),
|
||
'invalid_item',
|
||
CostEntryInput(
|
||
category_key='', category_name='',
|
||
amount=Decimal('50.00'), occurred_at=date(2026, 6, 2),
|
||
source_module='test', source_id='X002',
|
||
),
|
||
]
|
||
provider = MockCostProvider(entries)
|
||
|
||
result = cost_services.collect_from_provider(
|
||
provider=provider,
|
||
merchant=self.merchant,
|
||
start_date=date(2026, 6, 1),
|
||
end_date=date(2026, 6, 30),
|
||
)
|
||
|
||
self.assertEqual(result['total_seen'], 3)
|
||
self.assertEqual(result['created_count'], 1)
|
||
self.assertEqual(result['error_count'], 2)
|
||
self.assertEqual(cost_models.CostEntry.objects.count(), 1)
|
||
|
||
def test_collect_reuses_existing_category(self):
|
||
"""已有类目时不重复创建"""
|
||
cost_models.CostCategory.objects.create(
|
||
merchant=self.merchant, unique_key='electricity', name='电费旧名',
|
||
)
|
||
entries = [
|
||
CostEntryInput(
|
||
category_key='electricity', category_name='电费新名',
|
||
amount=Decimal('100.00'), occurred_at=date(2026, 6, 1),
|
||
source_module='test', source_id='E001',
|
||
),
|
||
]
|
||
provider = MockCostProvider(entries)
|
||
|
||
cost_services.collect_from_provider(
|
||
provider=provider,
|
||
merchant=self.merchant,
|
||
start_date=date(2026, 6, 1),
|
||
end_date=date(2026, 6, 30),
|
||
)
|
||
|
||
self.assertEqual(cost_models.CostCategory.objects.count(), 1)
|
||
cat = cost_models.CostCategory.objects.first()
|
||
self.assertEqual(cat.name, '电费旧名') # 不覆盖已有 name
|
||
|
||
|
||
class AggregateByCategoryTests(TestCase):
|
||
def setUp(self):
|
||
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.cat_a = cost_models.CostCategory.objects.create(
|
||
merchant=self.merchant, unique_key='electricity', name='电费',
|
||
)
|
||
self.cat_b = cost_models.CostCategory.objects.create(
|
||
merchant=self.merchant, unique_key='water', name='水费',
|
||
)
|
||
|
||
def _create_entries(self):
|
||
"""创建测试数据：电费 300+200=500，水费 80，其他商户 1000"""
|
||
cost_models.CostEntry.objects.create(
|
||
merchant=self.merchant, category=self.cat_a,
|
||
amount=Decimal('300.00'), occurred_at=date(2026, 6, 1),
|
||
)
|
||
cost_models.CostEntry.objects.create(
|
||
merchant=self.merchant, category=self.cat_a,
|
||
amount=Decimal('200.00'), occurred_at=date(2026, 6, 10),
|
||
)
|
||
cost_models.CostEntry.objects.create(
|
||
merchant=self.merchant, category=self.cat_b,
|
||
amount=Decimal('80.00'), occurred_at=date(2026, 6, 5),
|
||
)
|
||
other_cat = cost_models.CostCategory.objects.create(
|
||
merchant=self.other_merchant, unique_key='other', name='其他',
|
||
)
|
||
cost_models.CostEntry.objects.create(
|
||
merchant=self.other_merchant, category=other_cat,
|
||
amount=Decimal('1000.00'), occurred_at=date(2026, 6, 15),
|
||
)
|
||
|
||
def test_aggregate_by_category_desc_order(self):
|
||
"""按 total_amount 降序，只统计本商户"""
|
||
self._create_entries()
|
||
result = cost_services.aggregate_by_category(merchant=self.merchant)
|
||
|
||
self.assertEqual(len(result), 2)
|
||
self.assertEqual(result[0]['category_name'], '电费')
|
||
self.assertEqual(result[0]['total_amount'], '500.00')
|
||
self.assertEqual(result[0]['entry_count'], 2)
|
||
self.assertEqual(result[1]['category_name'], '水费')
|
||
self.assertEqual(result[1]['total_amount'], '80.00')
|
||
self.assertEqual(result[1]['entry_count'], 1)
|
||
|
||
def test_aggregate_by_category_with_start_date(self):
|
||
"""start_date 过滤"""
|
||
self._create_entries()
|
||
result = cost_services.aggregate_by_category(
|
||
merchant=self.merchant,
|
||
start_date=date(2026, 6, 5),
|
||
)
|
||
self.assertEqual(len(result), 2)
|
||
totals = {r['category_name']: r['total_amount'] for r in result}
|
||
self.assertEqual(totals['电费'], '200.00') # 只有 6.10 那条
|
||
self.assertEqual(totals['水费'], '80.00')
|
||
|
||
def test_aggregate_by_category_with_end_date(self):
|
||
"""end_date 过滤"""
|
||
self._create_entries()
|
||
result = cost_services.aggregate_by_category(
|
||
merchant=self.merchant,
|
||
end_date=date(2026, 6, 5),
|
||
)
|
||
self.assertEqual(len(result), 2)
|
||
totals = {r['category_name']: r['total_amount'] for r in result}
|
||
self.assertEqual(totals['电费'], '300.00') # 只有 6.1 那条
|
||
self.assertEqual(totals['水费'], '80.00')
|
||
|
||
def test_aggregate_by_category_with_both_dates(self):
|
||
"""start_date + end_date 同时过滤"""
|
||
self._create_entries()
|
||
result = cost_services.aggregate_by_category(
|
||
merchant=self.merchant,
|
||
start_date=date(2026, 6, 5),
|
||
end_date=date(2026, 6, 8),
|
||
)
|
||
self.assertEqual(len(result), 1)
|
||
self.assertEqual(result[0]['category_name'], '水费')
|
||
self.assertEqual(result[0]['total_amount'], '80.00')
|
||
|
||
def test_aggregate_no_data_returns_empty(self):
|
||
"""无数据返回空列表"""
|
||
result = cost_services.aggregate_by_category(merchant=self.merchant)
|
||
self.assertEqual(result, [])
|
||
|
||
def test_aggregate_includes_category_key(self):
|
||
"""返回结果包含 category_key"""
|
||
self._create_entries()
|
||
result = cost_services.aggregate_by_category(merchant=self.merchant)
|
||
self.assertEqual(result[0]['category_key'], 'electricity')
|
||
self.assertEqual(result[1]['category_key'], 'water') |