84 lines
3.8 KiB
Python
84 lines
3.8 KiB
Python
"""现货统计查询的范围、日期口径和实时刷新回归测试（不连接数据库）。"""
|
|
import ast
|
|
from datetime import date, datetime, timedelta
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import Mock
|
|
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1] / 'routes' / 'spot.py'
|
|
|
|
|
|
def load_function(name, query):
|
|
# 只加载纯查询函数，不初始化应用、账号权限或真实数据库。
|
|
tree = ast.parse(SOURCE.read_text(encoding='utf-8'))
|
|
node = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == name)
|
|
if node.decorator_list:
|
|
raise AssertionError('现货范围查询必须实时执行，避免分类变化后复用旧缓存')
|
|
namespace = dict(query=query, date=date, datetime=datetime, timedelta=timedelta)
|
|
exec(compile(ast.Module(body=[node], type_ignores=[]), str(SOURCE), 'exec'), namespace)
|
|
return namespace[name]
|
|
|
|
|
|
class SpotDashboardQueriesTest(unittest.TestCase):
|
|
def test_sales_use_product_date_index_and_keep_all_windows(self):
|
|
query = Mock(side_effect=[
|
|
[{'d': datetime(2026, 10, 14)}],
|
|
[{'货品编号': 'SKU-1', 'yesterday_sales': 2, 'ante_sales': None,
|
|
'sales_3': 3, 'sales_7': 7, 'sales_14': 14,
|
|
'sales_30': 30, 'total_sales': Decimal('50')}],
|
|
])
|
|
result, latest = load_function('_spot_sales_map', query)()
|
|
self.assertEqual(latest, date(2026, 10, 14))
|
|
self.assertEqual(result['SKU-1'], {
|
|
'昨日销量': 2, '前日销量': 0, '3天销量': 3, '7天销量': 7,
|
|
'14天销量': 14, '30天销量': 30, '总销量': 50,
|
|
})
|
|
sql, params = query.call_args.args
|
|
self.assertIn('FORCE INDEX (idx_product_date)', sql)
|
|
self.assertIn("p.product_type = '现货' AND p.deleted_at IS NULL", sql)
|
|
self.assertIn('o.货品编号 = p.product_code', sql)
|
|
self.assertEqual(params, (
|
|
latest, date(2026, 10, 13), date(2026, 10, 12), latest,
|
|
date(2026, 10, 8), latest, date(2026, 10, 1), latest,
|
|
date(2026, 9, 15), latest, latest, '%-NL', '%内料%',
|
|
))
|
|
|
|
def test_empty_outbound_does_not_run_aggregate(self):
|
|
query = Mock(return_value=[{'d': None}])
|
|
self.assertEqual(load_function('_spot_sales_map', query)(), ({}, None))
|
|
query.assert_called_once()
|
|
|
|
def test_sales_requery_after_data_or_product_changes(self):
|
|
query = Mock(side_effect=[
|
|
[{'d': '2026-10-14'}], [], [{'d': '2026-10-15'}], [],
|
|
])
|
|
function = load_function('_spot_sales_map', query)
|
|
self.assertEqual(function()[1], date(2026, 10, 14))
|
|
self.assertEqual(function()[1], date(2026, 10, 15))
|
|
self.assertEqual(query.call_count, 4)
|
|
|
|
def test_first_cost_still_comes_from_first_inventory_day(self):
|
|
row = {'product_code': 'SKU-1', 'first_inventory_date': date(2026, 9, 1),
|
|
'first_cost_price': Decimal('12.50')}
|
|
query = Mock(return_value=[row])
|
|
self.assertEqual(load_function('_spot_first_inventory_map', query)(), {'SKU-1': row})
|
|
sql = query.call_args.args[0]
|
|
self.assertIn("p2.product_type = '现货' AND p2.deleted_at IS NULL", sql)
|
|
self.assertIn('MIN(i2.data_date)', sql)
|
|
self.assertIn('first_batch.first_inventory_date = i.data_date', sql)
|
|
self.assertIn('ANY_VALUE(i.cost_price)', sql)
|
|
|
|
def test_inventory_requeries_instead_of_reusing_previous_product_set(self):
|
|
query = Mock(side_effect=[[], [{'product_code': 'NEW-SPOT',
|
|
'first_inventory_date': date(2026, 9, 1)}]])
|
|
function = load_function('_spot_first_inventory_map', query)
|
|
self.assertEqual(function(), {})
|
|
self.assertIn('NEW-SPOT', function())
|
|
self.assertEqual(query.call_count, 2)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|