poyee-data-warehouse/tests/unit/tracking/test_mask.py

# -*- coding:utf-8 -*-
import pytest

from dw_base.tracking.mask import apply_mask, apply_method, load_mask_conf


# ---- apply_method：6 种方法语义（对齐 datax/mask.py） ----

def test_method_md5():
    assert apply_method('md5', 'abc') == '900150983cd24fb0d6963f7d28e17f72'


def test_method_mask_middle():
    assert apply_method('mask_middle', '17301839727') == '173****9727'


def test_method_mask_middle_short_unchanged():
    assert apply_method('mask_middle', '1234567') == '1234567'  # len<8 不脱敏


def test_method_keep_first():
    assert apply_method('keep_first_2', 'abcdef') == 'ab****'


def test_method_keep_last():
    assert apply_method('keep_last_4', '17301839727') == '****9727'


def test_method_month_trunc():
    assert apply_method('month_trunc', '2024-05-13 10:00:00') == '2024-05'


def test_method_month_trunc_non_date_unchanged():
    assert apply_method('month_trunc', 'not-a-date') == 'not-a-date'


def test_method_none_passthrough():
    assert apply_method('md5', None) is None


def test_method_unknown_raises():
    with pytest.raises(ValueError, match='未知脱敏方法'):
        apply_method('bogus', 'x')


# ---- apply_mask：drop / mask / 兜底 / 优先级 / 嵌套 ----

CONF = {
    'PayOrder': {'drop': ['receiverName', 'receiverTelephone']},
    'Login': {'mask': {'phone': 'keep_last_4'}},
    'Mixed': {'drop': ['x'], 'mask': {'x': 'md5'}},
}


def test_drop_in_params_layer():
    props = {'storeId': '31', 'params': {'receiverName': 'Ethan', 'orderNo': 'A1'}}
    out = apply_mask('PayOrder', props, CONF)
    assert 'receiverName' not in out['params']
    assert out['params']['orderNo'] == 'A1'


def test_apply_mask_does_not_mutate_input():
    props = {'params': {'receiverName': 'Ethan'}}
    apply_mask('PayOrder', props, CONF)
    assert props['params']['receiverName'] == 'Ethan'


def test_undeclared_event_passthrough():
    props = {'params': {'whatever': '1'}}
    assert apply_mask('$AppStart', props, CONF) == props


def test_mask_top_level_field():
    out = apply_mask('Login', {'phone': '17301839727'}, CONF)
    assert out['phone'] == '****9727'


def test_drop_beats_mask_same_field():
    out = apply_mask('Mixed', {'x': 'secret'}, CONF)
    assert 'x' not in out  # drop 优先：字段消失而非被 md5


# ---- load_mask_conf：解析 / fail-fast ----

def test_load_conf_parses_drop_and_mask(tmp_path):
    p = tmp_path / 'tracking-mask.ini'
    p.write_text(
        '[event:PayOrder]\n'
        'drop = receiverName, receiverTelephone\n'
        'mask = receiverArea:mask_middle\n',
        encoding='utf-8')
    conf = load_mask_conf(str(p))
    assert conf['PayOrder']['drop'] == ['receiverName', 'receiverTelephone']
    assert conf['PayOrder']['mask'] == {'receiverArea': 'mask_middle'}


def test_load_conf_ignores_non_event_sections(tmp_path):
    p = tmp_path / 'm.ini'
    p.write_text('[DEFAULT]\nfoo = bar\n[event:X]\ndrop = a\n', encoding='utf-8')
    assert set(load_mask_conf(str(p))) == {'X'}


def test_load_conf_missing_file_raises(tmp_path):
    with pytest.raises(FileNotFoundError):
        load_mask_conf(str(tmp_path / 'nope.ini'))


def test_load_conf_unknown_method_raises(tmp_path):
    p = tmp_path / 'm.ini'
    p.write_text('[event:X]\nmask = f:bogus\n', encoding='utf-8')
    with pytest.raises(ValueError, match='未知脱敏方法'):
        load_mask_conf(str(p))


def test_load_conf_bad_mask_format_raises(tmp_path):
    p = tmp_path / 'm.ini'
    p.write_text('[event:X]\nmask = noColon\n', encoding='utf-8')
    with pytest.raises(ValueError, match='field:method'):
        load_mask_conf(str(p))