""" 元数据新增接口优化测试用例 测试 meta_node_add 接口的冗余检测和处理逻辑 """ import json from unittest.mock import MagicMock, patch class TestMetaNodeAddOptimization: """测试元数据新增接口的优化逻辑""" def test_exact_match_should_not_create_node(self, client): """ 测试场景1：完全匹配 预期：返回失败，提示元数据已存在，不创建新节点 """ # 模拟冗余检测返回完全匹配 with patch("app.api.meta_data.routes.check_redundancy_for_add") as mock_check: mock_check.return_value = { "has_exact_match": True, "exact_match_id": 12345, "has_candidates": True, "candidates": [ { "id": 12345, "name_zh": "测试元数据", "name_en": "test_meta", "data_type": "varchar(255)", "tag_ids": [1, 2], } ], } response = client.post( "/api/meta/node/add", json={ "name_zh": "测试元数据", "name_en": "test_meta", "data_type": "varchar(255)", "tag": [1, 2], }, ) data = json.loads(response.data) assert data["code"] != 200 assert "已存在" in data["message"] assert "12345" in data["message"] def test_suspicious_duplicate_should_create_node_and_review(self, client): """ 测试场景2：疑似重复 预期：创建新节点，创建审核记录，返回成功并提示疑似重复 """ # 模拟冗余检测返回疑似重复 with patch( "app.api.meta_data.routes.check_redundancy_for_add" ) as mock_check, patch( "app.api.meta_data.routes.neo4j_driver.get_session" ) as mock_session, patch( "app.api.meta_data.routes.write_redundancy_review_record_with_new_id" ) as mock_write_review: # 模拟冗余检测结果 mock_check.return_value = { "has_exact_match": False, "exact_match_id": None, "has_candidates": True, "candidates": [ { "id": 12345, "name_zh": "测试元数据", "name_en": "test_meta_old", "data_type": "varchar(255)", "tag_ids": [1], } ], } # 模拟 Neo4j 创建节点 mock_node = MagicMock() mock_node.id = 99999 mock_node.__getitem__ = lambda self, key: { "name_zh": "测试元数据", "name_en": "test_meta_new", "data_type": "varchar(255)", }.get(key) mock_node.get = lambda key, default=None: { "name_zh": "测试元数据", "name_en": "test_meta_new", "data_type": "varchar(255)", }.get(key, default) mock_result = MagicMock() mock_result.single.return_value = {"n": mock_node} mock_session_instance = MagicMock() mock_session_instance.run.return_value = mock_result mock_session.return_value.__enter__.return_value = mock_session_instance response = client.post( "/api/meta/node/add", json={ "name_zh": "测试元数据", "name_en": "test_meta_new", "data_type": "varchar(255)", "tag": [1, 2], }, ) data = json.loads(response.data) assert data["code"] == 200 assert "疑似重复" in data["message"] assert "审核" in data["message"] assert data["data"]["id"] == 99999 # 验证审核记录已创建 mock_write_review.assert_called_once() call_args = mock_write_review.call_args assert call_args[1]["new_meta"]["id"] == 99999 assert len(call_args[1]["candidates"]) == 1 def test_no_duplicate_should_create_node_normally(self, client): """ 测试场景3：无重复 预期：创建新节点，不创建审核记录，正常返回 """ with patch( "app.api.meta_data.routes.check_redundancy_for_add" ) as mock_check, patch( "app.api.meta_data.routes.neo4j_driver.get_session" ) as mock_session: # 模拟冗余检测返回无重复 mock_check.return_value = { "has_exact_match": False, "exact_match_id": None, "has_candidates": False, "candidates": [], } # 模拟 Neo4j 创建节点 mock_node = MagicMock() mock_node.id = 88888 mock_node.__getitem__ = lambda self, key: { "name_zh": "全新元数据", "name_en": "brand_new_meta", "data_type": "varchar(255)", }.get(key) mock_node.get = lambda key, default=None: { "name_zh": "全新元数据", "name_en": "brand_new_meta", "data_type": "varchar(255)", }.get(key, default) mock_result = MagicMock() mock_result.single.return_value = {"n": mock_node} mock_session_instance = MagicMock() mock_session_instance.run.return_value = mock_result mock_session.return_value.__enter__.return_value = mock_session_instance response = client.post( "/api/meta/node/add", json={ "name_zh": "全新元数据", "name_en": "brand_new_meta", "data_type": "varchar(255)", }, ) data = json.loads(response.data) assert data["code"] == 200 assert "疑似重复" not in data.get("message", "") assert data["data"]["id"] == 88888 def test_force_create_should_skip_redundancy_check(self, client): """ 测试场景4：强制创建 预期：跳过冗余检测，直接创建节点 """ with patch( "app.api.meta_data.routes.check_redundancy_for_add" ) as mock_check, patch( "app.api.meta_data.routes.neo4j_driver.get_session" ) as mock_session: # 模拟 Neo4j 创建节点 mock_node = MagicMock() mock_node.id = 77777 mock_node.__getitem__ = lambda self, key: { "name_zh": "强制创建元数据", "name_en": "force_create_meta", "data_type": "varchar(255)", }.get(key) mock_node.get = lambda key, default=None: { "name_zh": "强制创建元数据", "name_en": "force_create_meta", "data_type": "varchar(255)", }.get(key, default) mock_result = MagicMock() mock_result.single.return_value = {"n": mock_node} mock_session_instance = MagicMock() mock_session_instance.run.return_value = mock_result mock_session.return_value.__enter__.return_value = mock_session_instance response = client.post( "/api/meta/node/add", json={ "name_zh": "强制创建元数据", "name_en": "force_create_meta", "data_type": "varchar(255)", "force_create": True, }, ) data = json.loads(response.data) assert data["code"] == 200 assert data["data"]["id"] == 77777 # 验证冗余检测未被调用 mock_check.assert_not_called() class TestRedundancyCheckFunctions: """测试冗余检测辅助函数""" def test_check_redundancy_for_add_should_not_create_review(self): """ 测试 check_redundancy_for_add 函数 预期：只进行检测，不创建审核记录 """ from app.core.meta_data.redundancy_check import check_redundancy_for_add with patch( "app.core.meta_data.redundancy_check.neo4j_driver.get_session" ) as mock_session, patch( "app.core.meta_data.redundancy_check.write_redundancy_review_record" ) as mock_write: # 模拟查询返回疑似重复 mock_result = MagicMock() mock_result.__iter__ = lambda self: iter( [ { "id": 12345, "m": MagicMock( get=lambda key, default=None: { "name_zh": "测试元数据", "name_en": "test_meta", "data_type": "varchar(255)", }.get(key, default) ), } ] ) mock_session_instance = MagicMock() mock_session_instance.run.return_value = mock_result mock_session.return_value.__enter__.return_value = mock_session_instance result = check_redundancy_for_add( name_zh="测试元数据", name_en="test_meta_new", data_type="varchar(255)", tag_ids=[1, 2], ) # 验证返回结果 assert result["has_exact_match"] is False assert result["has_candidates"] is True assert len(result["candidates"]) > 0 # 验证未创建审核记录 mock_write.assert_not_called() def test_write_redundancy_review_record_with_new_id(self): """ 测试 write_redundancy_review_record_with_new_id 函数 预期：创建包含新节点ID的审核记录 """ from app.core.meta_data.redundancy_check import ( write_redundancy_review_record_with_new_id, ) with patch("app.core.meta_data.redundancy_check.db.session") as mock_session: new_meta = { "id": 99999, # 新创建的节点ID "name_zh": "测试元数据", "name_en": "test_meta_new", "data_type": "varchar(255)", "tag_ids": [1, 2], } candidates = [ { "id": 12345, "name_zh": "测试元数据", "name_en": "test_meta_old", "data_type": "varchar(255)", "tag_ids": [1], } ] write_redundancy_review_record_with_new_id( new_meta=new_meta, candidates=candidates, source="api" ) # 验证数据库操作 mock_session.add.assert_called_once() mock_session.commit.assert_called_once() # 获取添加的审核记录 added_review = mock_session.add.call_args[0][0] assert added_review.record_type == "redundancy" assert added_review.source == "api" assert added_review.new_meta["id"] == 99999 assert len(added_review.candidates) == 1