# -*- coding: utf-8 -*- import unittest from unittest import mock from routes import ai from services import ai_datasources, audit_schema class AiOperationHistoryTest(unittest.TestCase): def test_history_query_returns_exact_and_legacy_matches_distinctly(self): fake_logs = [ {'id': 2, 'created_at': '2026-10-28 10:00:00', 'user_id': 3, 'page': '计划预测', 'action': '数据变更', 'field': None, 'object_type': 'product', 'object_id': 'SKU-1', 'old_values': '{"采购计划":"甲"}', 'new_values': '{"采购计划":"乙"}', 'detail': '更新计划预测', 'result_status': 'success', 'request_id': 'r2', 'real_name': '张三', 'username': 'zhang'}, {'id': 1, 'created_at': '2026-10-27 10:00:00', 'user_id': 4, 'page': '计划预测', 'action': '数据变更', 'field': None, 'object_type': None, 'object_id': None, 'old_values': None, 'new_values': None, 'detail': '货品编号：SKU-1，采购计划调整', 'result_status': 'success', 'request_id': 'r1', 'real_name': '李四', 'username': 'li'}, ] with mock.patch.object(ai_datasources, 'query', side_effect=[[{'total': 2}], fake_logs]) as query: result = ai_datasources._h_operation_history({'product_code': 'SKU-1', 'module': '计划预测'}) self.assertEqual(2, result['summary']['row_count']) self.assertEqual('精确对象匹配', result['rows'][0]['匹配方式']) self.assertIn('采购计划：甲 → 乙', result['rows'][0]['修改内容']) self.assertEqual('历史文本匹配', result['rows'][1]['匹配方式']) self.assertIn('object_id=%s', query.call_args_list[0].args[0]) self.assertIn('%计划预测%', query.call_args_list[0].args[1]) def test_chinese_and_nested_product_identifiers_are_inferred(self): self.assertEqual(('product', 'SKU-2'), audit_schema.infer_object({'货品编号': 'SKU-2'})) self.assertEqual(('product', 'SKU-3'), audit_schema.infer_object({'items': [{'产品编码': 'SKU-3'}]})) self.assertEqual(('product', 'SKU-4'), audit_schema.infer_object({'id': 22, 'product': {'product_code': 'SKU-4'}})) def test_history_requests_are_isolated_from_other_ai_sources(self): tasks = ai._normalize_tasks([ {'datasource': 'product_profile', 'params': {'product_code': 'SKU-1'}}, {'datasource': 'operation_history', 'params': {'product_code': 'SKU-1'}}, {'datasource': 'planning_metrics', 'params': {'product_code': 'SKU-1'}}, ]) self.assertEqual(['operation_history'], [task['datasource'] for task in tasks]) def test_history_can_search_legacy_text_by_product_name(self): fake_logs = [{'id': 3, 'created_at': None, 'user_id': 1, 'page': '计划预测', 'action': '编辑', 'object_type': None, 'object_id': None, 'old_values': None, 'new_values': None, 'detail': '修改 Aromat 精华液', 'result_status': 'success', 'real_name': '用户', 'username': 'u'}] with mock.patch.object(ai_datasources, 'query', side_effect=[[{'total': 1}], fake_logs]) as query: result = ai_datasources._h_operation_history({'product_name': 'Aromat 精华液'}) self.assertEqual('历史文本匹配', result['rows'][0]['匹配方式']) self.assertIn('%Aromat 精华液%', query.call_args_list[0].args[1]) def test_history_query_is_scoped_to_permitted_pages(self): with mock.patch.object(ai_datasources, 'query', side_effect=[[{'total': 0}], []]) as query: ai_datasources._h_operation_history({'_allowed_pages': ['planning', '计划预测']}) self.assertIn('COALESCE(NULLIF(l.page', query.call_args_list[0].args[0]) self.assertIn('planning', query.call_args_list[0].args[1]) def test_history_field_values_are_hidden_without_field_permission(self): rows = [{'板块': '计划预测', '修改内容': '采购计划: （未记录） → 王某', '字段': '请求字段（旧日志未结构化）'}] with mock.patch('routes.auth._has_effective_page_permission', return_value=True), \ mock.patch('services.field_perms.field_perms_for_page', return_value={'采购计划': {'view': False}}): result = ai._filter_operation_history_permissions(1, '采购跟单', rows) self.assertEqual('修改字段内容因字段权限不可见', result[0]['修改内容']) def test_history_content_localizes_common_fields_and_keeps_full_values(self): fake_logs = [{'id': 4, 'created_at': None, 'user_id': 1, 'page': '产品信息管理', 'action': '编辑产品', 'object_type': 'product', 'object_id': 'SKU-5', 'old_values': '{"purchase_plan":"甲"}', 'new_values': '{"purchase_plan":"乙","product_name":"Aromat 精华液"}', 'detail': '', 'result_status': 'success', 'real_name': '用户', 'username': 'u'}] with mock.patch.object(ai_datasources, 'query', side_effect=[[{'total': 1}], fake_logs]): result = ai_datasources._h_operation_history({'product_code': 'SKU-5'}) row = result['rows'][0] self.assertEqual('采购计划', row['字段']) self.assertIn('采购计划：甲 → 乙', row['修改内容']) self.assertNotIn('货品名称', row['修改内容']) def test_nested_json_field_keys_are_localized(self): fake_logs = [{'id': 5, 'created_at': None, 'user_id': 1, 'page': 'planning', 'action': '数据变更', 'object_type': None, 'object_id': None, 'old_values': '{"columns":[{"field":"product_code","label":"货品编号"}],"sort_by":"product_name"}', 'new_values': '{"columns":[{"field":"product_code","label":"货品编号"}],"sort_by":"product_code"}', 'detail': '', 'result_status': 'success', 'real_name': '用户', 'username': 'u'}] with mock.patch.object(ai_datasources, 'query', side_effect=[[{'total': 1}], fake_logs]): row = ai_datasources._h_operation_history({'module': '计划预测'})['rows'][0] self.assertEqual('排序字段', row['字段']) self.assertIn('排序字段：货品名称 → 货品编号', row['修改内容']) self.assertNotIn('字段列', row['修改内容']) self.assertEqual('计划预测', row['板块']) def test_history_omits_unverifiable_bulk_payload_and_shows_only_nested_changes(self): fake_logs = [ {'id': 6, 'created_at': None, 'user_id': 1, 'page': '物料清单管理', 'action': '数据变更', 'object_type': 'product', 'object_id': 'SKU-6', 'old_values': None, 'new_values': '{"product_codes":["SKU-6"],"bom_data_map":{"SKU-6":[{"material_code":"MAT-1","quantity":2}]}}', 'detail': 'POST /api/bom/save；提交内容', 'result_status': 'success', 'real_name': '用户', 'username': 'u'}, {'id': 7, 'created_at': None, 'user_id': 1, 'page': '物料清单管理', 'action': '数据变更', 'object_type': 'product', 'object_id': 'SKU-7', 'old_values': '{"bom_data_map":{"SKU-7":[{"material_code":"MAT-1","quantity":2,"remark":"保留"}]}}', 'new_values': '{"bom_data_map":{"SKU-7":[{"material_code":"MAT-1","quantity":3,"remark":"保留"}]}}', 'detail': '', 'result_status': 'success', 'real_name': '用户', 'username': 'u'}, ] with mock.patch.object(ai_datasources, 'query', side_effect=[[{'total': 2}], fake_logs]): rows = ai_datasources._h_operation_history({'module': '物料清单管理'})['rows'] self.assertEqual('该记录未保存可比较的字段前后值', rows[0]['修改内容']) self.assertNotIn('SKU-6', rows[0]['修改内容']) self.assertIn('数量：2 → 3', rows[1]['修改内容']) self.assertNotIn('保留', rows[1]['修改内容']) def test_history_source_requires_operation_log_page_permission(self): with mock.patch('routes.auth._has_effective_page_permission', return_value=False): self.assertFalse(ai._can_view_datasource(8, '采购跟单', 'operation_history')) if __name__ == '__main__': unittest.main()