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 compute_partition_dt, 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_compute_partition_dt_stop_minus_1():
    assert compute_partition_dt('20260423') == '20260422'


def test_compute_partition_dt_month_boundary():
    assert compute_partition_dt('20260401') == '20260331'


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, stop_date='20260423')
    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, stop_date='20260423') is None


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


def test_parse_dt_aligns_with_stop_minus_1_for_multiday(tmp_path):
    # 多日范围 start=20260401 stop=20260410：dt 应 = stop-1 = 20260409
    # 避免老脚本 START_DATE=dt 假设引入的多日范围分区错位
    ini = _write_ini(tmp_path, '''\
        [writer]
        path = /user/hive/warehouse/db1.db/t1/dt=${dt}/
    ''')
    ddl = parse_ini_partition(ini, stop_date='20260410')
    assert 'PARTITION(dt=20260409)' 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()