DataOps-platform/tests/test_meta_node_add_optimization.py

"""
元数据新增接口优化测试用例

测试 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