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

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

from dw_base.datax.partition import execute_ddls, parse_ini_partition


def _write_ini(tmp_path: Path, content: str) -> str:
    p = tmp_path / 'x.ini'
    p.write_text(textwrap.dedent(content))
    return str(p)


def test_parse_partitioned_writer(tmp_path):
    ini = _write_ini(tmp_path, '''\
        [writer]
        dataSource = hdfs/prd-ha
        path = /user/hive/warehouse/test.db/raw_usr_app_user_cert_info_inc_d/dt=${dt}/
    ''')
    ddl = parse_ini_partition(ini, start_date='20260422')
    assert ddl == ('ALTER TABLE test.raw_usr_app_user_cert_info_inc_d '
                   'ADD IF NOT EXISTS PARTITION(dt=20260422);')


def test_parse_non_partitioned(tmp_path):
    ini = _write_ini(tmp_path, '''\
        [writer]
        dataSource = hdfs/xx
        path = /user/hive/warehouse/test.db/non_partitioned/
    ''')
    assert parse_ini_partition(ini, start_date='20260422') is None


def test_parse_no_writer_path(tmp_path):
    ini = _write_ini(tmp_path, '''\
        [writer]
        dataSource = mongo/xx
    ''')
    assert parse_ini_partition(ini, start_date='20260422') is None


def test_parse_dt_aligns_with_start_date_for_multiday(tmp_path):
    # 多日范围 start=20260401 stop=20260410：dt 应 = start_date = 20260401（业务日）
    # 与 HDFS writer 对齐，分区内允许含 [start, stop) 范围内所有数据
    ini = _write_ini(tmp_path, '''\
        [writer]
        path = /user/hive/warehouse/db1.db/t1/dt=${dt}/
    ''')
    ddl = parse_ini_partition(ini, start_date='20260401')
    assert 'PARTITION(dt=20260401)' in ddl


@patch('dw_base.datax.partition.subprocess.run')
def test_execute_ddls_empty_is_noop(mock_run):
    execute_ddls([])
    mock_run.assert_not_called()