121 lines
8.2 KiB
Python
121 lines
8.2 KiB
Python
# -*- 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()
|