Jelajahi Sumber

feat(datax): 补齐 3 项老入口功能(阶段 print + 串行 tee / HDFS 源 check / 默认日期)+ kb/94 标 ✅

- entry/runner: 补关键阶段 print(worker 选中 / gen 起止 / exec 起止 / 任务起止),串行模式走 runner._run_with_tee 写独立 log + 父 stdout(对齐老 | tee LOG_FILE)
- runner: 加 _hdfs_src_check,reader dataSource 为 hdfs/ 的 ini 在 gen json 前跑 hadoop fs -test -e + du -s,missing/empty 跳过 datax(对齐老 check_data_exists)
- bin: -start-date / -stop-date 不传时默认昨天/今天(对齐老 sh YESTERDAY/TODAY)
- tests: runner 加 5 条单测(hdfs n/a × 2 / missing / empty / run_job 跳过)
- kb/94 §1.2: 3 项补齐标 ✅

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
tianyu.chu 1 Minggu lalu
induk
melakukan
5e10a82151

+ 12 - 2
bin/datax-hdfs-export-starter.py

@@ -13,6 +13,7 @@ DataX hdfs-export 入口:源=HDFS(Hive 表数据),目标=外部系统(
 import argparse
 import os
 import sys
+from datetime import date, timedelta
 
 project_root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 sys.path.append(project_root_dir)
@@ -29,8 +30,10 @@ def main():
                         help='DataX ini 单文件(可多次)')
     parser.add_argument('-inis', action='append', default=[], metavar='DIR',
                         help='DataX ini 目录,非递归扫 *.ini(可多次)')
-    parser.add_argument('-start-date', required=True, metavar='YYYYMMDD')
-    parser.add_argument('-stop-date', required=True, metavar='YYYYMMDD')
+    parser.add_argument('-start-date', default=None, metavar='YYYYMMDD',
+                        help='默认昨天(对齐老 sh 行为)')
+    parser.add_argument('-stop-date', default=None, metavar='YYYYMMDD',
+                        help='默认今天(对齐老 sh 行为)')
     parser.add_argument('-host', default=None, metavar='HOSTNAME',
                         help='显式指定 worker(优先于 -random)')
     parser.add_argument('-random', action='store_true', dest='use_random',
@@ -41,9 +44,16 @@ def main():
                         help='只生成 json 不执行 datax.py')
     args = parser.parse_args()
 
+    # 默认日期:昨天 → 今天(对齐老 datax-single-job-starter.sh:207-219 行为)
+    if not args.start_date:
+        args.start_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
+    if not args.stop_date:
+        args.stop_date = date.today().strftime('%Y%m%d')
+
     print('{script} 收到参数: {argv}'.format(
         script=os.path.basename(__file__), argv=' '.join(sys.argv[1:]),
     ))
+    print('  start_date={s} stop_date={e}'.format(s=args.start_date, e=args.stop_date))
 
     exporter = DataxExport(
         base_dir=project_root_dir,

+ 12 - 2
bin/datax-hive-import-starter.py

@@ -17,6 +17,7 @@ DataX hive-import 入口:目标=Hive(自动预建分区),对应 jobs/raw
 import argparse
 import os
 import sys
+from datetime import date, timedelta
 
 project_root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 sys.path.append(project_root_dir)
@@ -33,8 +34,10 @@ def main():
                         help='DataX ini 单文件(可多次)')
     parser.add_argument('-inis', action='append', default=[], metavar='DIR',
                         help='DataX ini 目录,非递归扫 *.ini(可多次)')
-    parser.add_argument('-start-date', required=True, metavar='YYYYMMDD')
-    parser.add_argument('-stop-date', required=True, metavar='YYYYMMDD')
+    parser.add_argument('-start-date', default=None, metavar='YYYYMMDD',
+                        help='默认昨天(对齐老 sh 行为)')
+    parser.add_argument('-stop-date', default=None, metavar='YYYYMMDD',
+                        help='默认今天(对齐老 sh 行为)')
     parser.add_argument('-host', default=None, metavar='HOSTNAME',
                         help='显式指定 worker(优先于 -random)')
     parser.add_argument('-random', action='store_true', dest='use_random',
@@ -50,9 +53,16 @@ def main():
                         help='显式追加需建分区的 Hive 表(可多次)')
     args = parser.parse_args()
 
+    # 默认日期:昨天 → 今天(对齐老 datax-single-job-starter.sh:207-219 行为)
+    if not args.start_date:
+        args.start_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
+    if not args.stop_date:
+        args.stop_date = date.today().strftime('%Y%m%d')
+
     print('{script} 收到参数: {argv}'.format(
         script=os.path.basename(__file__), argv=' '.join(sys.argv[1:]),
     ))
+    print('  start_date={s} stop_date={e}'.format(s=args.start_date, e=args.stop_date))
 
     importer = DataxImport(
         base_dir=project_root_dir,

+ 16 - 9
dw_base/datax/entry.py

@@ -80,11 +80,12 @@ class _BaseDatax:
             job_name = path_utils.job_name_from_ini(ini_path)
             log_file = path_utils.log_path(self.log_root_dir, self.log_module, start_date, job_name)
             os.makedirs(os.path.dirname(log_file), exist_ok=True)
+            print('[datax] ini={j} worker={w} log={lf}'.format(j=job_name, w=w, lf=log_file))
             # 并行:每任务独立 log 文件(输出不回父 stdout,对齐老 > LOG 2>&1 &)
-            # 串行:继承父 stdout(用户可 tail 文件或靠外层 bash tee
+            # 串行:tee 到独立 log 文件 + 父 stdout(对齐老 | tee LOG_FILE
             if parallel:
                 with open(log_file, 'a', encoding='utf-8') as fh:
-                    return runner.run_job(
+                    rc = runner.run_job(
                         ini_path=ini_path, start_date=start_date, stop_date=stop_date,
                         worker_host=w, current_host=self.current_host,
                         base_dir=self.base_dir, python3_path=self.python3_path,
@@ -92,13 +93,19 @@ class _BaseDatax:
                         skip_datax=skip_datax,
                         stdout=fh, stderr=fh,
                     )
-            return runner.run_job(
-                ini_path=ini_path, start_date=start_date, stop_date=stop_date,
-                worker_host=w, current_host=self.current_host,
-                base_dir=self.base_dir, python3_path=self.python3_path,
-                datax_home=self.datax_home,
-                skip_datax=skip_datax,
-            )
+                print('[datax] ini={j} done rc={rc}'.format(j=job_name, rc=rc))
+                return rc
+            with open(log_file, 'a', encoding='utf-8') as fh:
+                rc = runner.run_job(
+                    ini_path=ini_path, start_date=start_date, stop_date=stop_date,
+                    worker_host=w, current_host=self.current_host,
+                    base_dir=self.base_dir, python3_path=self.python3_path,
+                    datax_home=self.datax_home,
+                    skip_datax=skip_datax,
+                    tee_to=fh,
+                )
+            print('[datax] ini={j} done rc={rc}'.format(j=job_name, rc=rc))
+            return rc
         return _run_one
 
 

+ 100 - 32
dw_base/datax/runner.py

@@ -3,19 +3,68 @@
 DataX 单任务执行器(从 bin/datax-single-job-starter.sh 搬迁 Python 版)。
 
 流程(对齐老脚本 generate_job_config + run_single_datax_job):
-1. 生成 json(调 bin/datax-job-config-generator.py,本机 subprocess / 远端 ssh)
-2. 执行 datax.py(同样本机 / 远端分发)
+1. (hdfs reader 场景)HDFS 源路径存在性 check,对齐老 check_data_exists
+2. 生成 json (调 -m dw_base.datax.cli gen-json,本机 subprocess / 远端 ssh)
+3. 执行 datax.py(同样本机 / 远端分发)
 
-生成 json 通过 `python3 -m dw_base.datax.cli gen-json` 调用(替代老 bin shim);
-worker 选择由 worker.select_worker 提供,本模块只接收最终 worker_host。
+stdout 输出模式三选一:
+- 默认 stdout=None, tee_to=None            → 继承父进程(最简)
+- stdout=fh, tee_to=None                   → 只写 fh(并行模式,不回父 stdout,对齐老 > LOG 2>&1 &)
+- stdout=None, tee_to=fh                   → tee 到 fh + 父 stdout(串行模式,对齐老 | tee LOG_FILE)
 """
 import os
 import shlex
 import subprocess
+import sys
+from configparser import ConfigParser
 
 from dw_base.datax import path_utils
 
 
+def _hdfs_src_check(ini_path: str, start_date: str) -> str:
+    """
+    HDFS 源路径存在性 + 空检查(对齐老 datax-single-job-starter.sh:128-146 check_data_exists)。
+
+    仅对 reader 是 hdfs 的 ini 触发;其他 reader 返回 'n/a'。
+
+    Returns:
+        'ok'       - 路径存在且有数据
+        'missing'  - 路径不存在(hadoop fs -test -e 失败)
+        'empty'    - 路径存在但空(hadoop fs -du -s 返回 0)
+        'n/a'      - 非 hdfs reader,不需要检查
+    """
+    cp = ConfigParser()
+    cp.read(ini_path)
+    if not cp.has_option('reader', 'dataSource'):
+        return 'n/a'
+    ds = cp.get('reader', 'dataSource')
+    if not ds.startswith('hdfs/'):
+        return 'n/a'
+    if not cp.has_option('reader', 'path'):
+        return 'n/a'
+    path = cp.get('reader', 'path')
+    # 替换占位符(对齐 hdfs_reader.load_others L27-32)
+    path = path.replace('${start_date}', start_date)
+    path = path.replace('${start-date}', start_date)
+    path = path.replace('${dt}', start_date)
+
+    # hadoop fs -test -e
+    if subprocess.run(['hadoop', 'fs', '-test', '-e', path],
+                      stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode != 0:
+        return 'missing'
+    # hadoop fs -du -s → 第一列是字节数
+    r = subprocess.run(['hadoop', 'fs', '-du', '-s', path],
+                       stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    if r.returncode == 0:
+        try:
+            size = int(r.stdout.decode('utf-8', errors='replace').strip().split()[0])
+            if size == 0:
+                return 'empty'
+        except (IndexError, ValueError):
+            pass
+    return 'ok'
+
+
 def run_job(ini_path: str,
             start_date: str,
             stop_date: str,
@@ -26,28 +75,20 @@ def run_job(ini_path: str,
             datax_home: str,
             skip_datax: bool = False,
             stdout=None,
-            stderr=None) -> int:
+            stderr=None,
+            tee_to=None) -> int:
     """
-    单任务执行:生成 json → 执行 datax.py(本机或 ssh 到远端)。
-
-    Args:
-        ini_path: DataX ini 绝对路径
-        start_date / stop_date: yyyyMMdd
-        worker_host: 已由 worker.select_worker 决定的目标节点
-        current_host: hostname -s
-        base_dir: 项目根绝对路径
-        python3_path: python3 可执行路径(conf/env.sh 的 PYTHON3_PATH)
-        datax_home: DataX 安装路径(conf/env.sh 的 DATAX_HOME)
-        skip_datax: 只生成 json 不执行
-        stdout / stderr: 传给 subprocess.run 的重定向 target(文件句柄等);
-                        None = 继承父进程。batch 并行模式传文件句柄隔离每任务日志
-
-    Returns:
-        datax.py 的 returncode(0 = 成功);skip_datax=True 时返回 0
-
-    Raises:
-        RuntimeError: 生成 json 失败
+    单任务执行。
     """
+    # 1. HDFS 源路径 check(对齐老 check_data_exists;仅 hdfs reader 触发)
+    src_state = _hdfs_src_check(ini_path, start_date)
+    if src_state == 'missing':
+        print('[datax] {ini} HDFS 源路径不存在,跳过 datax'.format(ini=ini_path))
+        return 0
+    if src_state == 'empty':
+        print('[datax] {ini} HDFS 源路径空,跳过 datax'.format(ini=ini_path))
+        return 0
+
     json_path = path_utils.json_output_path(base_dir, ini_path)
     os.makedirs(os.path.dirname(json_path), exist_ok=True)
 
@@ -58,12 +99,15 @@ def run_job(ini_path: str,
         '-start-date', start_date,
         '-stop-date', stop_date,
     ]
+    print('[datax] {ini} 开始生成 json @ worker={w}'.format(ini=ini_path, w=worker_host))
     gen_rc = _run_local_or_remote(gen_argv, worker_host, current_host, base_dir,
-                                  stdout=stdout, stderr=stderr)
+                                  stdout=stdout, stderr=stderr, tee_to=tee_to)
     if gen_rc != 0:
         raise RuntimeError('生成 DataX json 失败: ' + ini_path)
+    print('[datax] {ini} json 生成完成'.format(ini=ini_path))
 
     if skip_datax:
+        print('[datax] {ini} skip_datax=True,跳过执行'.format(ini=ini_path))
         return 0
 
     exec_argv = [
@@ -71,21 +115,45 @@ def run_job(ini_path: str,
         os.path.join(datax_home, 'bin', 'datax.py'),
         json_path,
     ]
-    return _run_local_or_remote(exec_argv, worker_host, current_host, base_dir,
-                                stdout=stdout, stderr=stderr)
+    print('[datax] {ini} 开始执行 datax.py @ worker={w}'.format(ini=ini_path, w=worker_host))
+    rc = _run_local_or_remote(exec_argv, worker_host, current_host, base_dir,
+                              stdout=stdout, stderr=stderr, tee_to=tee_to)
+    print('[datax] {ini} datax.py 执行完成 rc={rc}'.format(ini=ini_path, rc=rc))
+    return rc
 
 
 def _run_local_or_remote(argv, worker_host: str, current_host: str, base_dir: str,
-                         stdout=None, stderr=None) -> int:
+                         stdout=None, stderr=None, tee_to=None) -> int:
     """
-    本机:subprocess.run 直跑 + PYTHONPATH 注入 base_dir(让 python -m dw_base.datax.cli 能找到包)
-    远端:ssh worker_host 'cd <base_dir> && <cmd>'(cwd 为项目根,同样保证 -m 能找到 dw_base)
-    stdout/stderr 默认继承父进程;batch 层可传文件句柄做每任务独立日志。
+    本机:subprocess.run 直跑 + PYTHONPATH 注入 base_dir
+    远端:ssh worker_host 'cd <base_dir> && <cmd>'
+    tee_to 优先于 stdout/stderr:提供 tee_to 时走 Popen 行循环 tee(文件 + 父 stdout)
     """
     if worker_host == current_host:
         env = os.environ.copy()
         existing = env.get('PYTHONPATH', '')
         env['PYTHONPATH'] = base_dir + (os.pathsep + existing if existing else '')
+        if tee_to is not None:
+            return _run_with_tee(argv, tee_to, env=env)
         return subprocess.run(argv, stdout=stdout, stderr=stderr, env=env).returncode
     remote_cmd = 'cd ' + shlex.quote(base_dir) + ' && ' + ' '.join(shlex.quote(a) for a in argv)
-    return subprocess.run(['ssh', worker_host, remote_cmd], stdout=stdout, stderr=stderr).returncode
+    ssh_argv = ['ssh', worker_host, remote_cmd]
+    if tee_to is not None:
+        return _run_with_tee(ssh_argv, tee_to)
+    return subprocess.run(ssh_argv, stdout=stdout, stderr=stderr).returncode
+
+
+def _run_with_tee(argv, log_fh, env=None) -> int:
+    """
+    Popen + 行循环 tee:subprocess 的 stdout/stderr 合并后同时写 log_fh 和 sys.stdout。
+    对齐老脚本 bash 层的 `| tee LOG_FILE` 行为(串行模式独立 log 文件)。
+    """
+    proc = subprocess.Popen(argv,
+                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                            env=env, bufsize=1, universal_newlines=True)
+    for line in proc.stdout:
+        sys.stdout.write(line)
+        sys.stdout.flush()
+        log_fh.write(line)
+        log_fh.flush()
+    return proc.wait()

+ 8 - 4
kb/94-重构对比.md

@@ -22,8 +22,8 @@
 | 11 | 字段脱敏 | 不支持;需要在 ini 里自己手写 querySql 用 SQL 表达式脱敏 | `[mask]` 声明式段自动生成 querySql;PG 5 种脱敏(3 静态 `month_trunc` / `md5` / `mask_middle` + 2 动态 `keep_first_{n}` / `keep_last_{n}`),列名白名单防 SQL 注入 |
 | 12 | Workers 配置 | 硬编码在 `init.sh:13-28`(老 m3/d1-d4 列表 + 权重 map) | 外配 `conf/workers.ini`(新 cdhmaster02/cdhnode01-03) |
 | 13 | 日志路径 | 多维派生:`${LOG_ROOT_DIR}/datax/${SRC_DST}/${PROJECT_LAYER_ENV}/${DB_ENV}/${GROUP}/${START_DATE}/${JOB_NAME}.log`,路径含老 `conf/datax/config/` 残留语义 | 扁平化:`${LOG_ROOT_DIR}/datax/${dt}/${job_name}.log`(对齐 kb/90 §7.2.1) |
-| 14 | 日志内容(stdout)| `pretty_print` **带 ANSI 颜色**:`NORM_MGT`(紫)+ `NORM_GRN`(绿)+ `NORM_RED`(红)+ `NORM_YEL`(黄)分区;驱动常量来自 `bin/common/print-constants.sh` + `dw_base/__init__.py` | Python 原生 `print` **无颜色**(退化);DataX JVM 自身日志输出不变;异常改用 `raise RuntimeError` 携堆栈 |
-| 15 | Tee / 独立日志文件 | bash 层 `\| tee LOG`(串行)/ `> LOG 2>&1 &`(并行),shell pipe 天然做到 | Python 层 `subprocess.run(stdout=file_handle)`,并行模式每任务独立 log 文件;串行模式 stdout 继承父进程、无独立日志文件(需上游 bash 套 tee) |
+| 14 | 日志内容(stdout)| `pretty_print` **带 ANSI 颜色**:`NORM_MGT`(紫)+ `NORM_GRN`(绿)+ `NORM_RED`(红)+ `NORM_YEL`(黄)分区;驱动常量来自 `bin/common/print-constants.sh` + `dw_base/__init__.py` | Python 原生 `print` **无颜色**(颜色丢失仍未补);**关键阶段 print 已补齐**(worker 选中 / gen 起止 / exec 起止 / 任务起止,2026-04-24 落地);DataX JVM 自身日志输出不变;异常改用 `raise RuntimeError` 携堆栈 |
+| 15 | Tee / 独立日志文件 | bash 层 `\| tee LOG`(串行)/ `> LOG 2>&1 &`(并行),shell pipe 天然做到 | Python 层 `runner._run_with_tee` Popen + 行循环 tee(串行 `tee_to=fh` 写文件 + 父 stdout;并行 `stdout=fh` 只写文件)。**2026-04-24 补齐串行独立 log 文件**,行为对齐老 `\| tee LOG_FILE` |
 | 16 | 错误处理 | `set -e` + exit code 透传、出错位置信息含糊 | Python Exception + `subprocess.returncode`,抛 `RuntimeError` 含上下文(ini 名 / 阶段) |
 | 17 | 分布式 | `-random` + `workers.ini` 加权随机 + ssh 分发到 worker | 同(待 kb/93 ADR-02 拍板后可砍 ssh 分发,DS worker group 接管) |
 | 18 | 字段变换 | 无(需源库做或下游 Spark SQL 做) | ini `[mask]` 声明式 + `mask.py` 翻译为 PG `querySql` 内 SQL 表达式,源库端脱敏(合规硬约束:敏感不出业务库) |
@@ -36,8 +36,12 @@
 
 对比表里**新不如老**的项:
 
-- **#14 日志颜色退化**:新入口 Python `print` 无 ANSI。用户控制台看到的执行日志从带颜色的"分级可读"变成单色纯文本。如要恢复用 `colorama` 或自己写 20 行 ANSI escape helper 即可,本轮不做
-- **#15 串行模式日志文件缺失**:新入口串行跑时 stdout 只继承父进程、不单独落 log 文件,需要上游 bash 套 `| tee`。并行模式保留了独立 log 文件。影响面小(DS 调度默认带 stdout 捕获),本轮不做
+- ~~**#14 日志颜色退化**~~ / ~~**关键阶段 print 缺失**~~ —— 2026-04-24 **阶段 print 已补齐**(worker 选中 / gen 起止 / exec 起止 / 任务起止);颜色仍退化(仅单色文本),如要恢复用 `colorama` 或自写 ANSI escape helper,当前不做
+- ~~**#15 串行模式日志文件缺失**~~ —— 2026-04-24 **已补齐**,`runner.run_job(tee_to=fh)` Popen 行循环 tee 到文件 + 父 stdout,行为对齐老 `| tee LOG_FILE`
+- **HDFS 源路径存在性 check 未搬迁** —— 2026-04-24 **已补齐**,`runner._hdfs_src_check` 对 `reader.dataSource` 是 `hdfs/...` 的 ini 在 gen json 前跑 `hadoop fs -test -e` + `hadoop fs -du -s`,路径不存在 / 空则跳过 datax(对齐老 `datax-single-job-starter.sh:128-146` check_data_exists)
+- **默认日期未做** —— 2026-04-24 **已补齐**,入口层 `-start-date` / `-stop-date` 不传时默认昨天/今天(对齐老 sh 的 YESTERDAY/TODAY 默认)
+
+**仍未做**:hdfs-kafka 特殊 writer 说明(老 sh `datax-single-job-starter.sh:232-245` 10 行 columnType/columnMapping 业务说明),本项目用 kafka 概率低,用到再补
 
 ### 1.3 收益点概述
 

+ 72 - 0
tests/unit/datax/test_runner.py

@@ -90,3 +90,75 @@ def test_run_job_skip_datax_only_runs_gen(mock_run, tmp_path):
     )
     assert rc == 0
     assert mock_run.call_count == 1
+
+
+import textwrap
+from dw_base.datax.runner import _hdfs_src_check
+
+
+def _ini_with_reader(tmp_path, reader_body):
+    p = tmp_path / 'reader.ini'
+    p.write_text(textwrap.dedent(reader_body), encoding='utf-8')
+    return str(p)
+
+
+def test_hdfs_src_check_na_for_non_hdfs_reader(tmp_path):
+    ini = _ini_with_reader(tmp_path, '''\
+        [reader]
+        dataSource = postgresql/dev-x
+        path = /some/path
+    ''')
+    assert _hdfs_src_check(ini, '20260422') == 'n/a'
+
+
+def test_hdfs_src_check_na_when_no_reader(tmp_path):
+    ini = tmp_path / 'empty.ini'
+    ini.write_text('', encoding='utf-8')
+    assert _hdfs_src_check(str(ini), '20260422') == 'n/a'
+
+
+@patch('dw_base.datax.runner.subprocess.run')
+def test_hdfs_src_check_missing(mock_run, tmp_path):
+    # hadoop fs -test -e 返回非 0 → missing
+    mock_run.return_value = _RC(1)
+    ini = _ini_with_reader(tmp_path, '''\
+        [reader]
+        dataSource = hdfs/prd-ha
+        path = /user/hive/warehouse/x/dt=${dt}/
+    ''')
+    assert _hdfs_src_check(ini, '20260422') == 'missing'
+
+
+@patch('dw_base.datax.runner.subprocess.run')
+def test_hdfs_src_check_empty_when_du_size_zero(mock_run, tmp_path):
+    # 第一次调用 test -e → 0 成功,第二次调 du -s 返回 "0  .." → empty
+    call_count = {'n': 0}
+
+    def side_effect(*args, **kwargs):
+        call_count['n'] += 1
+        if call_count['n'] == 1:
+            return _RC(0)
+        return type('R', (), {'returncode': 0, 'stdout': b'0   0   /path\n', 'stderr': b''})()
+
+    mock_run.side_effect = side_effect
+    ini = _ini_with_reader(tmp_path, '''\
+        [reader]
+        dataSource = hdfs/prd-ha
+        path = /user/hive/warehouse/x/dt=${dt}/
+    ''')
+    assert _hdfs_src_check(ini, '20260422') == 'empty'
+
+
+@patch('dw_base.datax.runner._hdfs_src_check', return_value='missing')
+@patch('dw_base.datax.runner.subprocess.run')
+def test_run_job_skips_when_hdfs_src_missing(mock_run, _mock_check, tmp_path):
+    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
+    # HDFS missing → 不调 gen / exec
+    assert mock_run.call_count == 0