# -*- 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))