157 lines
9.7 KiB
Python
157 lines
9.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""防止预览截断、排序剔零和模型生成数字改变品牌库存合计。"""
|
||
import json
|
||
import unittest
|
||
from unittest import mock
|
||
|
||
from flask import Flask, session
|
||
|
||
from routes import ai
|
||
from services import ai_datasources as sources
|
||
|
||
|
||
QUESTIONS = [
|
||
'查询品牌为Aromat Herbs的产品，库存*单价的总成本是多少',
|
||
'Aromat Herbs品牌库存总成本的具体数值是多少？',
|
||
'Aromat Herbs查一下这个品牌的商品数据目前的库存和库存金额分别是多少？',
|
||
]
|
||
|
||
|
||
class InventoryAccuracyTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.rows = [{'product_code': f'SKU-{i:03}', '产品名称': f'商品{i}',
|
||
'品牌': 'Aromat Herbs', '库存': 0 if i < 25 else 2,
|
||
'成本价': 0.1, '库存金额': 0 if i < 25 else 0.2}
|
||
for i in range(251)]
|
||
self.rows.append({'product_code': 'OTHER', '产品名称': '其他品牌',
|
||
'品牌': 'Aromat Herbs Plus', '库存': 999, '成本价': 10, '库存金额': 9990})
|
||
|
||
def _inventory(self, params):
|
||
with mock.patch.object(sources, '_get_latest_inv_date', return_value='2026-10-14'), \
|
||
mock.patch.object(sources, '_inventory_value_rows', return_value=self.rows):
|
||
return sources._h_inventory_value(params)
|
||
|
||
def test_totals_cover_more_than_200_and_do_not_change_with_sort_or_limit(self):
|
||
for options in ({'limit': 1}, {'limit': 30}, {'limit': 200},
|
||
{'limit': 200, 'sort_by': '库存金额'},
|
||
{'limit': 7, 'sort_by': '库存', 'sort_order': 'asc'}):
|
||
with self.subTest(options=options):
|
||
result = self._inventory({'brand': ' Aromat Herbs ', **options})
|
||
summary = result['summary']
|
||
self.assertEqual(251, summary['sku_count'])
|
||
self.assertEqual(452, summary['stock_qty'])
|
||
self.assertEqual(45.2, summary['inventory_value'])
|
||
self.assertEqual(25, summary['zero_stock_sku_count'])
|
||
self.assertLessEqual(len(result['rows']), options['limit'])
|
||
self.assertNotIn('OTHER', [r['product_code'] for r in result['rows']])
|
||
|
||
def test_brand_and_detail_sources_produce_same_totals(self):
|
||
detail = self._inventory({'brand': 'Aromat Herbs', 'limit': 200, 'sort_by': '库存金额'})
|
||
with mock.patch.object(sources, '_get_latest_inv_date', return_value='2026-10-14'), \
|
||
mock.patch.object(sources, '_inventory_value_rows', return_value=self.rows), \
|
||
mock.patch.object(sources, 'compute_product_sales_cached', return_value={}), \
|
||
mock.patch.object(sources, 'query', return_value=[]):
|
||
brand = sources._h_brand_overview({'brand': 'Aromat Herbs', 'limit': 1, 'sort_by': '库存金额'})
|
||
for field in ('sku_count', 'stock_qty', 'inventory_value', 'zero_stock_sku_count', 'date'):
|
||
self.assertEqual(detail['summary'][field], brand['summary'][field])
|
||
self.assertEqual(1, brand['summary']['row_count'])
|
||
self.assertEqual(251, brand['rows'][0]['SKU数'])
|
||
self.assertEqual(45.2, brand['rows'][0]['库存金额'])
|
||
|
||
def test_source_uses_single_date_active_products_and_decimal_rounding(self):
|
||
with mock.patch.object(sources, '_get_latest_inv_date', side_effect=['2026-10-14', '2026-10-15']) as latest, \
|
||
mock.patch.object(sources, 'query', return_value=[{'product_code': 'A', '库存': 1, '成本价': 0.005}]) as query, \
|
||
mock.patch.object(sources, '_products_map', return_value={'A': {'brand': ' Test ', 'product_type': '成品'}}) as products:
|
||
result = sources._h_inventory_value({'brand': 'Test'})
|
||
latest.assert_called_once()
|
||
products.assert_called_once_with(['A'], active_only=True)
|
||
self.assertEqual(('2026-10-14',), query.call_args.args[1])
|
||
self.assertEqual(0.01, result['summary']['inventory_value'])
|
||
self.assertEqual('2026-10-14', result['summary']['date'])
|
||
|
||
def test_wrong_model_values_are_removed_from_all_narrative_sections(self):
|
||
dataset = {'key': 'inventory_value', 'name': '商品库存金额',
|
||
**self._inventory({'brand': 'Aromat Herbs'})}
|
||
wrong = {k: ['错误总额999999.99元'] for k in ('findings', 'evidence', 'risks', 'actions', 'limitations', 'follow_ups')}
|
||
wrong.update(summary='错误总額999999.99元', title='999999.99')
|
||
result = ai._apply_authoritative_inventory_summary(wrong, [dataset], QUESTIONS[0])
|
||
self.assertNotIn('999999.99', json.dumps(result, ensure_ascii=False))
|
||
self.assertIn('251个SKU', result['summary'])
|
||
self.assertIn('45.20元', result['summary'])
|
||
|
||
def test_inventory_guard_does_not_replace_cross_business_or_sales_analysis(self):
|
||
dataset = {'key': 'inventory_value', 'name': '商品库存金额',
|
||
**self._inventory({'brand': 'Aromat Herbs'})}
|
||
original = {'summary': '建议核对BOM', 'actions': ['核对包材']}
|
||
self.assertIs(original, ai._apply_authoritative_inventory_summary(
|
||
original, [dataset, {'key': 'bom_supply_chain'}], '库存不足，需要下单吗？'))
|
||
self.assertIs(original, ai._apply_authoritative_inventory_summary(
|
||
original, [dataset], '该品牌库存和销量变化的原因是什么？'))
|
||
|
||
def test_empty_and_unpriced_inventory_are_explicit(self):
|
||
empty = self._inventory({'brand': '不存在'})
|
||
self.assertEqual(0, empty['summary']['sku_count'])
|
||
dataset = {'key': 'inventory_value', 'name': '商品库存金额', **empty}
|
||
self.assertIn('0个SKU', ai._apply_authoritative_inventory_summary(None, [dataset], '库存多少钱')['summary'])
|
||
self.rows[30]['成本价'] = 0
|
||
self.rows[30]['库存金额'] = 0
|
||
dataset = {'key': 'inventory_value', 'name': '商品库存金额', **self._inventory({'brand': 'Aromat Herbs'})}
|
||
result = ai._apply_authoritative_inventory_summary(None, [dataset], '库存多少钱')
|
||
self.assertEqual('中', result['confidence'])
|
||
self.assertIn('1个有库存', result['limitations'][0])
|
||
|
||
def test_planner_extracts_same_brand_and_does_not_hijack_other_scopes(self):
|
||
with mock.patch.object(ai, 'query', return_value=[{'brand': 'Aromat Herbs'}, {'brand': 'PUNOMO'}]):
|
||
for question in QUESTIONS:
|
||
self.assertEqual('Aromat Herbs', ai._plan_brand_inventory_lookup(question)['tasks'][0]['params']['brand'])
|
||
for question in ('Aromat Herbs和PUNOMO库存分别多少？', 'Aromat Herbs的包材库存多少？',
|
||
'Aromat Herbs中AH-CP-4842-TMGJ库存多少？', 'Aromat Herbs昨日库存多少？'):
|
||
self.assertIsNone(ai._plan_brand_inventory_lookup(question))
|
||
top = ai._plan_brand_inventory_lookup('Aromat Herbs库存金额排名前10的产品？')
|
||
self.assertEqual(10, top['tasks'][0]['params']['limit'])
|
||
|
||
def test_api_different_questions_return_same_facts_without_calling_llm(self):
|
||
app = Flask(__name__)
|
||
app.secret_key = 'test'
|
||
old_handler = sources.AI_DATASOURCES['inventory_value']['handler']
|
||
sources.AI_DATASOURCES['inventory_value']['handler'] = lambda params: self._inventory(params)
|
||
try:
|
||
for question in QUESTIONS:
|
||
with app.test_request_context('/api/ai/ask', method='POST', json={'question': question}):
|
||
session.update(user_id=1, role='super_admin')
|
||
with mock.patch.object(ai, 'query', return_value=[{'brand': 'Aromat Herbs'}]), \
|
||
mock.patch.object(ai, 'operation_required', return_value=True), \
|
||
mock.patch.object(ai, '_can_view_datasource', return_value=True), \
|
||
mock.patch.object(ai, '_can_export_ai', return_value=True), \
|
||
mock.patch.object(ai, 'datasource_meta', return_value={'name': '商品库存金额', 'data_date': '旧日期'}), \
|
||
mock.patch.object(ai, '_plan_analysis', side_effect=AssertionError('must not call LLM')), \
|
||
mock.patch.object(ai, '_synthesize_analysis', side_effect=AssertionError('must not call LLM')), \
|
||
mock.patch.object(ai, '_save_history'), mock.patch.object(ai, 'add_operation_log'):
|
||
payload = ai.ai_ask.__wrapped__().get_json()['data']
|
||
self.assertIn('251个SKU', payload['summary'])
|
||
self.assertIn('45.20元', payload['summary'])
|
||
self.assertEqual(251, payload['total_rows'])
|
||
self.assertEqual('2026-10-14', payload['data_date'])
|
||
self.assertEqual(200, len(payload['rows']))
|
||
finally:
|
||
sources.AI_DATASOURCES['inventory_value']['handler'] = old_handler
|
||
|
||
def test_permission_denial_never_returns_authoritative_money(self):
|
||
app = Flask(__name__)
|
||
app.secret_key = 'test'
|
||
with app.test_request_context('/api/ai/ask', method='POST', json={'question': QUESTIONS[0]}):
|
||
session.update(user_id=1, role='viewer')
|
||
with mock.patch.object(ai, 'query', return_value=[{'brand': 'Aromat Herbs'}]), \
|
||
mock.patch.object(ai, 'operation_required', return_value=True), \
|
||
mock.patch.object(ai, '_can_view_datasource', return_value=False), \
|
||
mock.patch.object(ai, '_can_export_ai', return_value=False), \
|
||
mock.patch.object(ai, '_save_history'):
|
||
payload = ai.ai_ask.__wrapped__().get_json()['data']
|
||
self.assertEqual([], payload['rows'])
|
||
self.assertIn('没有查看', payload['summary'])
|
||
self.assertNotIn('45.20', json.dumps(payload))
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|