test_analysisFactory.py 文件源码

def test_get_analysis_lambda_and_analysis_lambda_reduce(self, mock_data_delivery, mock_alert_factory,
                                                            mock_historical_data):
        input_data_structure = {'rule': [{'key': False, 'func_name': 'Max', 'input_field': 'traffic'},
                                         {'key': False, 'func_name': 'Max', 'input_field': 'ip_size'},
                                         {'key': False, 'func_name': 'Sum', 'input_field': 'ip_size_sum'}],
                                'operation_type': 'reduceByKey'}
        enumerate_output_aggregation_field = {"traffic": 0, "ip_size": 1, "ip_size_sum": 2}

        mock_class = MagicMock()
        mock_analysis = MagicMock()
        mock_class.MockAnalysis.return_value = mock_analysis
        mock_analysis.analysis = MagicMock()
        sys.modules['analysis.MockAnalysis'] = mock_class
        mock_history_data_singleton = MagicMock()
        mock_data_delivery.return_value = mock_history_data_singleton
        obj_mock_alert_factory = MagicMock()
        mock_alert_factory.return_value = obj_mock_alert_factory
        obj_mock_historical_data = MagicMock()
        mock_historical_data.return_value = obj_mock_historical_data

        analysis_factory = AnalysisFactory(self._config, input_data_structure, enumerate_output_aggregation_field)

        test_lambda = analysis_factory.get_analysis_lambda()

        self.assertIsInstance(test_lambda, types.LambdaType, "get_analysis_lambda should return lambda function")

        test_lambda((6666, 7777, 8888))

        mock_analysis.analysis.assert_called_with(obj_mock_historical_data, obj_mock_alert_factory)