DataOps-platform/tests/test_metric_check.py

"""
测试指标公式检查功能
"""

import unittest
from unittest.mock import Mock, patch, MagicMock
from app.core.data_metric.metric_interface import metric_check


class TestMetricCheck(unittest.TestCase):
    """测试 metric_check 函数"""

    @patch('app.core.data_metric.metric_interface.connect_graph')
    def test_simple_formula(self, mock_connect_graph):
        """测试简单的加法公式"""
        # Mock Neo4j session
        mock_driver = MagicMock()
        mock_session = MagicMock()
        mock_connect_graph.return_value = mock_driver
        mock_driver.session.return_value.__enter__ = Mock(return_value=mock_session)
        mock_driver.session.return_value.__exit__ = Mock(return_value=None)
        
        # Mock query results
        mock_record1 = {
            'n': {
                'name': '单价',
                'en_name': 'unit_price',
                'createTime': '2024-01-15 10:00:00'
            },
            'node_id': 101
        }
        mock_record2 = {
            'n': {
                'name': '数量',
                'en_name': 'quantity',
                'createTime': '2024-01-15 10:01:00'
            },
            'node_id': 102
        }
        
        mock_session.run.side_effect = [
            Mock(single=Mock(return_value=mock_record1)),
            Mock(single=Mock(return_value=mock_record2))
        ]
        
        # 测试
        formula = "销售额 = 单价 + 数量"
        result = metric_check(formula)
        
        # 验证
        self.assertEqual(len(result), 2)
        self.assertTrue(any(item['variable'] == '单价' for item in result))
        self.assertTrue(any(item['variable'] == '数量' for item in result))
        self.assertTrue(all(item['findit'] == 1 for item in result))

    @patch('app.core.data_metric.metric_interface.connect_graph')
    def test_complex_formula(self, mock_connect_graph):
        """测试复杂的公式（包含多种运算符）"""
        # Mock Neo4j session
        mock_driver = MagicMock()
        mock_session = MagicMock()
        mock_connect_graph.return_value = mock_driver
        mock_driver.session.return_value.__enter__ = Mock(return_value=mock_session)
        mock_driver.session.return_value.__exit__ = Mock(return_value=None)
        
        # Mock query results - 一个找到，一个未找到
        mock_record1 = {
            'n': {
                'name': '销售收入',
                'en_name': 'sales_revenue',
                'createTime': '2024-01-15 10:00:00'
            },
            'node_id': 201
        }
        
        mock_session.run.side_effect = [
            Mock(single=Mock(return_value=mock_record1)),
            Mock(single=Mock(return_value=None))  # 成本未找到
        ]
        
        # 测试
        formula = "利润率 = (销售收入 - 成本) / 销售收入 * 100"
        result = metric_check(formula)
        
        # 验证
        self.assertEqual(len(result), 2)
        found_items = [item for item in result if item['findit'] == 1]
        not_found_items = [item for item in result if item['findit'] == 0]
        self.assertEqual(len(found_items), 1)
        self.assertEqual(len(not_found_items), 1)

    def test_formula_without_equals(self):
        """测试没有等号的公式"""
        formula = "销售额"
        result = metric_check(formula)
        
        # 验证返回空列表
        self.assertEqual(result, [])

    @patch('app.core.data_metric.metric_interface.connect_graph')
    def test_formula_with_numbers_only(self, mock_connect_graph):
        """测试只包含数字的公式"""
        # Mock Neo4j session
        mock_driver = MagicMock()
        mock_session = MagicMock()
        mock_connect_graph.return_value = mock_driver
        mock_driver.session.return_value.__enter__ = Mock(return_value=mock_session)
        mock_driver.session.return_value.__exit__ = Mock(return_value=None)
        
        # 测试
        formula = "结果 = 100 + 200 * 3"
        result = metric_check(formula)
        
        # 验证返回空列表（因为没有中文变量）
        self.assertEqual(result, [])

    @patch('app.core.data_metric.metric_interface.connect_graph')
    def test_formula_with_chinese_brackets(self, mock_connect_graph):
        """测试包含中文括号的公式"""
        # Mock Neo4j session
        mock_driver = MagicMock()
        mock_session = MagicMock()
        mock_connect_graph.return_value = mock_driver
        mock_driver.session.return_value.__enter__ = Mock(return_value=mock_session)
        mock_driver.session.return_value.__exit__ = Mock(return_value=None)
        
        # Mock query result
        mock_record = {
            'n': {
                'name': '收入',
                'en_name': 'revenue',
                'createTime': '2024-01-15 10:00:00'
            },
            'node_id': 301
        }
        
        mock_session.run.side_effect = [
            Mock(single=Mock(return_value=mock_record)),
            Mock(single=Mock(return_value=mock_record))
        ]
        
        # 测试
        formula = "总额 = （收入 + 收入）"
        result = metric_check(formula)
        
        # 验证（收入会被去重）
        self.assertEqual(len(result), 1)
        self.assertEqual(result[0]['variable'], '收入')
        self.assertEqual(result[0]['findit'], 1)

    @patch('app.core.data_metric.metric_interface.connect_graph')
    def test_database_connection_failure(self, mock_connect_graph):
        """测试数据库连接失败"""
        # Mock connection failure
        mock_connect_graph.return_value = None
        
        # 测试
        formula = "销售额 = 单价 * 数量"
        result = metric_check(formula)
        
        # 验证返回空列表
        self.assertEqual(result, [])


class TestMetricCheckAPI(unittest.TestCase):
    """测试 metric_check API 接口"""

    def setUp(self):
        """设置测试环境"""
        from app import create_app
        self.app = create_app('testing')
        self.client = self.app.test_client()
        self.ctx = self.app.app_context()
        self.ctx.push()

    def tearDown(self):
        """清理测试环境"""
        self.ctx.pop()

    @patch('app.api.data_metric.routes.metric_check')
    def test_api_success(self, mock_metric_check):
        """测试API成功调用"""
        # Mock返回值
        mock_metric_check.return_value = [
            {
                "variable": "单价",
                "name_zh": "单价",
                "name_en": "unit_price",
                "id": 101,
                "create_time": "2024-01-15 10:00:00",
                "findit": 1
            }
        ]
        
        # 发送请求
        response = self.client.post(
            '/api/data/metric/check',
            json={'formula': '销售额 = 单价 * 数量'}
        )
        
        # 验证响应
        self.assertEqual(response.status_code, 200)
        data = response.get_json()
        self.assertEqual(data['code'], 200)
        self.assertEqual(data['message'], 'success')
        self.assertTrue(isinstance(data['data'], list))

    def test_api_empty_formula(self):
        """测试空公式"""
        response = self.client.post(
            '/api/data/metric/check',
            json={'formula': ''}
        )
        
        # 验证响应
        self.assertEqual(response.status_code, 200)
        data = response.get_json()
        self.assertIn('error', data)

    def test_api_missing_formula(self):
        """测试缺少formula参数"""
        response = self.client.post(
            '/api/data/metric/check',
            json={}
        )
        
        # 验证响应
        self.assertEqual(response.status_code, 200)
        data = response.get_json()
        self.assertIn('error', data)


if __name__ == '__main__':
    unittest.main()