# -*- coding: utf-8 -*- import unittest from unittest import mock from flask import Flask, session from datetime import timedelta from routes import ai from services import ai_insights class AiAnalysisP1Test(unittest.TestCase): def setUp(self): self.app = Flask(__name__) self.app.secret_key = 'test' def test_position_profiles_are_business_specific(self): planner = ai_insights.position_profile('procurement_planner') follower = ai_insights.position_profile('procurement_follower') operator = ai_insights.position_profile('operator') self.assertIn('stockout', planner['focus_types']) self.assertIn('overdue_order', follower['focus_types']) self.assertIn('sales_decline', operator['focus_types']) self.assertNotEqual(planner['questions'], follower['questions']) def test_filter_requires_all_source_permissions(self): raw = {'generated_at': '2026-10-01 10:00:00', 'items': [ {'id': 'a', 'type': 'overstock', 'category': '库存健康', 'severity': '高', 'title': 'A', 'detail': 'A', 'source_keys': ['planning_metrics', 'inventory_value'], 'metrics': {'库存金额': 1000, '库存合计': 20}}, {'id': 'b', 'type': 'supplier_risk', 'category': '供应商风险', 'severity': '高', 'title': 'B', 'detail': 'B', 'source_keys': ['supplier_delivery'], 'metrics': {}}, ]} result = ai_insights.filter_insights( raw, ['planning_metrics'], ai_insights.position_profile('procurement_planner'), limit=20) self.assertEqual([], result['items']) permitted = ai_insights.filter_insights( raw, ['planning_metrics', 'inventory_value'], ai_insights.position_profile('procurement_planner'), limit=20) self.assertEqual(['a'], [x['id'] for x in permitted['items']]) self.assertEqual(1000, permitted['items'][0]['metrics']['库存金额']) def test_deterministic_builder_detects_core_anomalies(self): today = ai_insights.date.today() expected = (today - timedelta(days=1)).isoformat() datasets = { 'planning_metrics': [{'product_code': 'SKU1', '日销量预测': 10, '总支撑天数': 3, '库存可支撑天数': 2, '建议下单数量': 500, '库存': 20, '加工厂库存': 0}], 'product_profile': [{'product_code': 'SKU1', '货品名称': '测试成品', '产品类型': '成品', '生产货期': 10, '物流时间': 2}], 'product_sales': [{'product_code': 'SKU1', '7天平均销量': 3, '30天平均销量': 10}], 'order_detail': [{'订单类型': '国内订单', '订单ID': 1, 'product_code': 'SKU1', '状态': '未完结', '未到货数量': 6000, '预计到货日期': expected, '供应商': '测试供应商'}], 'supplier_delivery': [{'供应商': '测试供应商', '未完结订单数': 4, '未到货数量': 6000, '到货完成率': 40, '不良品率': 0}], 'bom_supply_chain': [{'product_code': 'SKU1', '物料名称': '测试彩盒', '单位用量': 1, '可支撑成品数量': 100, '物料供应商': '测试供应商'}], 'inventory_value': [{'product_code': 'SKU1', '库存金额': 1000}], } with mock.patch.object(ai_insights, '_fetch', side_effect=lambda key: datasets[key]), \ mock.patch.object(ai_insights, 'datasource_meta', return_value={'data_date': '2026-10-01'}): result = ai_insights._build_raw_insights() types = {x['type'] for x in result['items']} self.assertIn('stockout', types) self.assertIn('sales_decline', types) self.assertIn('supplier_risk', types) self.assertIn('bom_shortage', types) def test_insight_api_returns_role_profile(self): raw = {'generated_at': '2026-10-01 10:00:00', 'data_dates': {}, 'items': []} with self.app.test_request_context('/api/ai/insights'): session['user_id'] = 1 session['role'] = 'super_admin' with mock.patch.object(ai, 'operation_required', return_value=True), \ mock.patch.object(ai, 'query', return_value=[{'position': 'procurement_planner'}]), \ mock.patch('services.ai_insights.get_operational_insights', return_value=raw), \ mock.patch.object(ai, '_can_view_datasource', return_value=True): response = ai.ai_insights.__wrapped__() payload = response.get_json() self.assertEqual(0, payload['code']) self.assertEqual('采购计划', payload['data']['profile']['name']) def test_report_api_saves_report_history(self): raw = {'generated_at': '2026-10-01 10:00:00', 'data_dates': {}, 'items': [{ 'id': 'x', 'type': 'stockout', 'category': '缺货与补货', 'severity': '高', 'title': 'SKU缺货', 'detail': '支撑天数不足', 'question': '分析SKU缺货', 'source_keys': ['planning_metrics'], 'metrics': {'建议下单数量': 500}, }]} report = {'title': '今日经营风险报告', 'summary': '存在1项高风险。', 'sections': [], 'priorities': [], 'follow_ups': []} with self.app.test_request_context('/api/ai/report', method='POST', json={'report_type': 'daily'}): session['user_id'] = 1 session['role'] = 'super_admin' with mock.patch.object(ai, 'operation_required', return_value=True), \ mock.patch.object(ai, 'query', return_value=[{'position': 'bp'}]), \ mock.patch('services.ai_insights.get_operational_insights', return_value=raw), \ mock.patch.object(ai, '_can_view_datasource', side_effect=lambda uid, role, key: key == 'planning_metrics'), \ mock.patch.object(ai, '_call_deepseek_json', return_value=report), \ mock.patch.object(ai, '_save_history') as save_history, \ mock.patch.object(ai, 'add_operation_log'): response = ai.ai_generate_report.__wrapped__() payload = response.get_json() self.assertEqual(200, payload['code']) self.assertEqual('report', payload['data']['type']) self.assertEqual('存在1项高风险。', payload['data']['summary']) self.assertEqual('report', save_history.call_args.args[4]) def test_frontend_has_insight_chat_and_report_modes(self): with open('static/js/app.js', encoding='utf-8') as fh: text = fh.read() self.assertIn('function aiSwitchMode(mode)', text) self.assertIn('aiModeInsights', text) self.assertIn('aiModeChat', text) self.assertIn('aiModeReports', text) self.assertIn('/api/ai/insights', text) self.assertIn('/api/ai/report', text) if __name__ == '__main__': unittest.main()