poyee-data-warehouse/tests/unit/datax/test_runner.py

# -*- coding:utf-8 -*-
from unittest.mock import patch

import pytest

from dw_base.datax.runner import run_job


class _RC:
    def __init__(self, code): self.returncode = code


@patch('dw_base.datax.runner.subprocess.run')
def test_run_job_local_two_subprocess_calls(mock_run, tmp_path):
    mock_run.return_value = _RC(0)
    rc = run_job(
        ini_path=str(tmp_path / 'x.ini'),
        start_date='20260422', stop_date='20260423',
        worker_host='cdhmaster02', current_host='cdhmaster02',
        base_dir=str(tmp_path), python3_path='/usr/bin/python3',
        datax_home='/opt/datax',
    )
    assert rc == 0
    assert mock_run.call_count == 2  # 生成 json + 执行 datax.py
    first_argv = mock_run.call_args_list[0][0][0]
    assert first_argv[0] == '/usr/bin/python3'
    assert 'datax-job-config-generator.py' in first_argv[2]


@patch('dw_base.datax.runner.subprocess.run')
def test_run_job_remote_uses_ssh(mock_run, tmp_path):
    mock_run.return_value = _RC(0)
    rc = run_job(
        ini_path=str(tmp_path / 'x.ini'),
        start_date='20260422', stop_date='20260423',
        worker_host='cdhnode02', current_host='cdhmaster02',
        base_dir=str(tmp_path), python3_path='/usr/bin/python3',
        datax_home='/opt/datax',
    )
    assert rc == 0
    first_argv = mock_run.call_args_list[0][0][0]
    assert first_argv[0] == 'ssh'
    assert first_argv[1] == 'cdhnode02'
    # remote_cmd 是单字符串参数
    assert 'datax-job-config-generator.py' in first_argv[2]


@patch('dw_base.datax.runner.subprocess.run')
def test_run_job_gen_failure_raises(mock_run, tmp_path):
    mock_run.return_value = _RC(1)
    with pytest.raises(RuntimeError, match='生成 DataX json 失败'):
        run_job(
            ini_path=str(tmp_path / 'x.ini'),
            start_date='20260422', stop_date='20260423',
            worker_host='cdhmaster02', current_host='cdhmaster02',
            base_dir=str(tmp_path), python3_path='/usr/bin/python3',
            datax_home='/opt/datax',
        )


@patch('dw_base.datax.runner.subprocess.run')
def test_run_job_skip_datax_only_runs_gen(mock_run, tmp_path):
    mock_run.return_value = _RC(0)
    rc = run_job(
        ini_path=str(tmp_path / 'x.ini'),
        start_date='20260422', stop_date='20260423',
        worker_host='cdhmaster02', current_host='cdhmaster02',
        base_dir=str(tmp_path), python3_path='/usr/bin/python3',
        datax_home='/opt/datax',
        skip_datax=True,
    )
    assert rc == 0
    assert mock_run.call_count == 1