# -*- coding:utf-8 -*- from unittest.mock import MagicMock, patch import pytest from dw_base.datax import cli @patch('dw_base.datax.cli.JobConfigGenerator') def test_gen_json_invokes_generator(MockGen): mock_inst = MagicMock() MockGen.return_value = mock_inst cli.main(['gen-json', 'some.ini', '-start-date', '20260422', '-stop-date', '20260423']) MockGen.assert_called_once() call_args = MockGen.call_args[0] # 位置参数:(base_dir, ini, start, stop, output) assert call_args[1] == 'some.ini' assert call_args[2] == '20260422' assert call_args[3] == '20260423' assert call_args[4].endswith('some.json') mock_inst.run.assert_called_once() def test_gen_json_missing_args_exits(): with pytest.raises(SystemExit): cli.main(['gen-json', 'some.ini']) # 缺 -start-date/-stop-date def test_unknown_subcommand_exits(): with pytest.raises(SystemExit): cli.main(['bogus-cmd']) @patch('dw_base.datax.cli.JobConfigGenerator') def test_gen_json_speed_cli_args_flow_through(MockGen): """L3 CLI -channel/-byte/-record 传给 JobConfigGenerator cli_speed_overrides kwarg。""" mock_inst = MagicMock() MockGen.return_value = mock_inst cli.main(['gen-json', 'some.ini', '-start-date', '20260422', '-stop-date', '20260423', '-channel', '20', '-byte', '20971520', '-record', '50000']) kwargs = MockGen.call_args[1] assert kwargs['cli_speed_overrides'] == { 'channel': 20, 'byte': 20971520, 'record': 50000, } @patch('dw_base.datax.cli.JobConfigGenerator') def test_gen_json_speed_cli_default_none(MockGen): """CLI 没传 speed args 时,cli_speed_overrides 里全是 None。""" mock_inst = MagicMock() MockGen.return_value = mock_inst cli.main(['gen-json', 'some.ini', '-start-date', '20260422', '-stop-date', '20260423']) kwargs = MockGen.call_args[1] assert kwargs['cli_speed_overrides'] == { 'channel': None, 'byte': None, 'record': None, }