EvalHarness/tests/test_fingerprint.py

"""Offline regression tests for the fp_fusion fingerprint module.

零网络: 只测探针电池定义/归一化/信号打分/裁决阶梯/参考解析/CLI 接线。
Run:  python tests/test_fingerprint.py   (or pytest tests/test_fingerprint.py -q)
"""

import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

from evalharness.fingerprint import REFERENCES_DIR, main as fp_main  # noqa: E402
from evalharness.fingerprint.battery import (ALL_CELL_DEFS, ALL_TEXT_PROBES,  # noqa: E402
                                             CORE16_CELLS, TEXT_PRUNED_V7)
from evalharness.fingerprint.engine import (build_d_normalized,  # noqa: E402
                                            normalize_answer)
from evalharness.fingerprint.run_fp_fusion import (_assemble_probes,  # noqa: E402
                                                   resolve_reference)
from evalharness.fingerprint.scorer import (build_report, identity_signal,  # noqa: E402
                                            load_aliases, meta_signal,
                                            requested_family, verdict_from_score)


# ---------------------------------------------------------------- 电池定义 ----

def test_battery_definitions():
    ids = [p['id'] for p in ALL_TEXT_PROBES]
    assert len(ids) == 36 and len(set(ids)) == 36, '文本探针应 36 条且 id 唯一'
    assert set(TEXT_PRUNED_V7) <= set(ids), '剪枝集必须 ⊆ 全量探针'
    assert len(TEXT_PRUNED_V7) == 7
    universe = {f"{c['id']}:{lang}" for c in ALL_CELL_DEFS for lang in ('en', 'zh')}
    assert set(CORE16_CELLS) <= universe, 'core16 必须 ⊆ cell 宇宙(26 cell)'
    for c in ALL_CELL_DEFS:  # 每个 cell 定义必须带中英改写池
        assert c['par'].get('en') and c['par'].get('zh'), c['id']


def test_probe_assembly_counts():
    """剪枝口径与介绍文档一致: 文本 29 = 36 - 7; V 层 7 条; full 模式含对抗组."""
    slim = _assemble_probes('verify', None, set(TEXT_PRUNED_V7))
    assert len(slim) == 29
    assert not ({p['id'] for p in slim} & set(TEXT_PRUNED_V7))
    full = _assemble_probes('verify', None, None)
    assert len(full) == 36
    variant = _assemble_probes('variant', None, set(TEXT_PRUNED_V7))
    v_ids = [p['id'] for p in variant if p['layer'] == 'V']
    assert len(v_ids) == 7
    adv = _assemble_probes('adversarial', 'You are Kimi.', set(TEXT_PRUNED_V7))
    assert any(p['id'].startswith('adv_role_') for p in adv)
    assert any(p['id'].startswith('adv_challenge_') for p in adv)
    # 显式 --impersonate 时对抗组只保留 角色组+挑战组(风格模仿组不跑)
    assert not any(p['id'].startswith('adv_style_') for p in adv)


# ---------------------------------------------------------------- 归一化 ----

def test_normalize_answer():
    assert normalize_answer('42', ('int', 1, 100)) == ('42', 'valid')
    assert normalize_answer('七。', ('int', 1, 100))[0] == '7'
    assert normalize_answer('Blue.', ('color',)) == ('blue', 'valid')
    assert normalize_answer('violet', ('color',))[0] == 'purple'
    assert normalize_answer('灰色', ('color',))[0] == '灰', 'zh 色词保留中文去尾字'
    assert normalize_answer('礼拜三。', ('enum', ['monday']))[0] == 'wednesday'
    assert normalize_answer('I cannot answer that', ('int', 1, 100))[1] == 'refusal'
    assert normalize_answer('', ('int', 1, 100))[1] == 'empty'


# ---------------------------------------------------------------- 信号打分 ----

def _rec(layer, pid, response, error=False, **meta):
    return {'layer': layer, 'id': pid, 'response': response, 'error': error,
            'meta': meta, 'latency_ms': 100, 'completion_tokens': 3}


def test_identity_and_meta_signals():
    aliases = load_aliases()  # 默认解析到包内 family_aliases.json
    assert requested_family('ZhipuAi/GLM-5.3', aliases) == 'glm'
    i_ok = [_rec('I', f'i_direct_en{i}', f'I am GLM-{i}, made by Zhipu AI.') for i in range(4)]
    idn = identity_signal(i_ok, aliases, 'glm')
    assert idn['s_idn'] == 1.0 and idn['consistent'] == 4
    i_bad = i_ok + [_rec('I', 'i_outlier', 'I am Claude, made by Anthropic.')]
    idn2 = identity_signal(i_bad, aliases, 'glm')
    assert idn2['s_idn'] < 1.0 and idn2['outliers'], '离群自称应被记录'

    k_recs = [_rec('K', 'k_cutoff1', 'My knowledge cutoff is January 2025.'),
              _rec('K', 'k_params', 'I do not have access to that information.',
                   metacog=True)]
    meta = meta_signal(k_recs, [])
    assert meta['s_meta'] == 1.0 and len(meta['cutoffs_unique']) == 1
    k_messy = k_recs + [_rec('K', 'k_cutoff2', 'Training data updated until June 2024.')]
    assert meta_signal(k_messy, [])['s_meta'] < 1.0, '口径混乱应扣分'


def test_verdict_ladder():
    assert verdict_from_score(0.90, True) == 'VERIFIED'
    assert verdict_from_score(0.90, False) == 'LIKELY_MATCH', '无参考不得验明正身'
    assert verdict_from_score(0.72, True) == 'LIKELY_MATCH'
    assert verdict_from_score(0.55, True) == 'INCONCLUSIVE'
    assert verdict_from_score(0.32, False) == 'SUSPECTED_MISMATCH'
    assert verdict_from_score(0.10, True) == 'MISMATCH'


def test_build_report_synthetic():
    """合成 clean 记录走全管线: 质量门 SUFFICIENT, 分数与裁决齐备."""
    recs = ([_rec('I', f'i_direct_en{i}', 'I am GLM, made by Zhipu AI.') for i in range(4)]
            + [_rec('I', 'i_zh_direct', '我是智谱AI开发的GLM大模型。')]
            + [_rec('K', 'k_cutoff1', 'My knowledge cutoff is January 2025.')]
            + [_rec('K', 'k_params', 'I do not know my parameter count.', metacog=True)]
            + [_rec('C', f'c_refuse_l{i}', 'I cannot help with that.') for i in (1, 2)]
            + [_rec('S', 's_list', '1. health\n2. sleep')])
    d_norm = build_d_normalized([
        {'layer': 'D', 'cell': 'random-number-1-100:en', 'response': '42',
         'error': False, 'arrival': i} for i in range(5)])
    assert len(d_norm) == 5 and all(s['norm'] == '42' for s in d_norm)
    aliases = load_aliases()
    dist_cmp = {'mean_jsd': None, 'split_half': 0.03, 'baseline_p50': 500}
    rep = build_report(recs, d_norm, dist_cmp, 'ZhipuAi/GLM-5.3', None,
                       aliases, {'input': 100, 'output': 200}, 12.5, mode='verify')
    assert rep['gate']['quality'] == 'SUFFICIENT'
    assert rep['verdict'] == 'LIKELY_MATCH', '无参考模式裁决上限 LIKELY_MATCH'
    assert 0.0 <= rep['score'] <= 1.0
    assert rep['signals']['identity']['s_idn'] == 1.0
    assert rep['signals']['meta']['s_meta'] == 1.0


# ------------------------------------------------------- 参考库与 CLI 接线 ----

def test_reference_resolution():
    r = resolve_reference('glm53')
    assert r.endswith('glm53_fusion_reference.json') and Path(r).exists()
    from evalharness.fingerprint.engine import load_reference

    ref = load_reference(r)  # bundled 参考可正常加载(detector schema)
    assert ref['model'] and ref['cells']
    missing = resolve_reference('no_such_model_xyz')
    assert missing == 'no_such_model_xyz', '未知名原样返回交由 load_reference 报错'


def test_bundled_references_present():
    fusion = list(REFERENCES_DIR.glob('*_fusion_reference.json'))
    assert len(fusion) >= 12, f'内置 fp_fusion 参考应 ≥12 个, 实际 {len(fusion)}'


def test_cli_wiring():
    import contextlib
    import io

    from evalharness.cli import build_parser

    args = build_parser().parse_args(
        ['fingerprint', 'run', '--api-url', 'http://x/v1'])
    assert args.fp_args == ['run', '--api-url', 'http://x/v1'], 'REMAINDER 应完整透传'
    with contextlib.redirect_stdout(io.StringIO()) as buf:
        assert fp_main(['list']) == 0
    assert 'glm53' in buf.getvalue()


def test_cli_help_lists_fingerprint():
    """顶层 help 应包含 fingerprint 子命令(惰性导入不拖垮 CLI 启动)."""
    root = Path(__file__).parent.parent
    out = subprocess.run([sys.executable, '-m', 'evalharness', '--help'],
                         capture_output=True, text=True, cwd=root)
    assert out.returncode == 0 and 'fingerprint' in out.stdout


if __name__ == '__main__':
    fails = 0
    for name, fn in sorted({k: v for k, v in globals().items()
                            if k.startswith('test_') and callable(v)}.items()):
        try:
            fn()
            print(f'PASS {name}')
        except AssertionError as e:
            fails += 1
            print(f'FAIL {name}: {e}')
        except Exception as e:
            fails += 1
            print(f'ERROR {name}: {type(e).__name__}: {e}')
    sys.exit(1 if fails else 0)