AiDOPWarehouse/tools/integration-simulator/tests/test_report.py

"""F5 全链路报告：只存摘要、保存/导出前脱敏、Markdown 结构与渠道校验（任务书 §10 / §24）。"""
from __future__ import annotations

import json

from app import CHANNELS, REPORT_FIELDS, REPORTS, build_report_markdown
from clients.redaction import REDACTED

SECRET = "S3CRET-VALUE-DO-NOT-LEAK"
PASSWORD = "pw-123456"
ACCESS_KEY = "TESTKPARTNER"


def _fields(**kw) -> dict:
    base = {
        "runId": "SIM-RPT-001",
        "channel": "DB_SYNC",
        "environment": "开发库 / 租户 T1",
        "objectCode": "ITEM",
        "simPrep": "建表并装载 3 行",
        "manualSteps": "在 Ai-DOP 登记 SIM_MODE3_DB 来源并触发同步",
        "simEvidence": "源表 3 行，最大增量 2026-10-12",
        "aidopConn": "测连成功",
        "flowIds": "SYNC-1001",
        "stagingCheck": "贴源 3 行，与源端一致",
        "targetResult": "目标表可查 3 行",
        "conclusion": "全链路打通",
        "gaps": "",
        "cleanup": "删除 SIM_MODE3_DB 来源",
    }
    base.update(kw)
    return base


def test_report_fields_cover_taskbook_sections():
    labels = [label for _, label in REPORT_FIELDS]
    for expected in ("运行编号", "集成方式", "环境与测试租户", "模拟端准备结果",
                     "人工执行步骤", "模拟端证据", "Ai-DOP 测连结果",
                     "同步/入站流水编号", "贴源核验结果", "目标业务结果",
                     "结论", "未完成项", "清理提示"):
        assert any(expected in label for label in labels), f"report missing section: {expected}"


def test_report_save_redacts_secret_and_password():
    # 真实运行中密码来自环境变量 _secrets()；这里把密码作为已知 Secret 传入，模拟同一路径
    entry = REPORTS.save(_fields(simPrep=f"用密码 {PASSWORD} 连接，Secret={SECRET} 初始化"),
                         secrets=[SECRET, PASSWORD])
    dumped = json.dumps(entry, ensure_ascii=False)
    assert SECRET not in dumped
    assert PASSWORD not in dumped
    assert REDACTED in dumped


def test_report_save_masks_chinese_password_label():
    # 测试人员自由文本写「密码 xxx」时也须遮蔽（任务书 §24），无需知道真实密码
    entry = REPORTS.save(_fields(simPrep=f"数据库连接：密码 {PASSWORD}，端口 3306"))
    dumped = json.dumps(entry, ensure_ascii=False)
    assert PASSWORD not in dumped
    assert "3306" in dumped  # 非敏感部分保留


def test_report_save_masks_jwt_and_connection_string():
    entry = REPORTS.save(_fields(
        flowIds="token=eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.abcdEFGH1234，"
                "conn=mysql://root:pw-123456@127.0.0.1:3306/aidop_integration_sim"))
    dumped = json.dumps(entry, ensure_ascii=False)
    assert "eyJhbGciOiJIUzI1NiJ9" not in dumped
    assert "pw-123456" not in dumped
    assert "aidop_integration_sim" in dumped  # 非敏感部分保留


def test_report_save_keeps_only_last_four_of_access_key():
    entry = REPORTS.save(_fields(environment=f"开放身份 AccessKey={ACCESS_KEY}"),
                         access_keys=[ACCESS_KEY])
    dumped = json.dumps(entry, ensure_ascii=False)
    assert ACCESS_KEY not in dumped
    assert "****TNER" in dumped


def test_report_markdown_contains_all_sections_and_redaction_note():
    entry = REPORTS.save(_fields())
    md = build_report_markdown(entry)
    for label in ("运行编号 runId", "方式甲 DB_SYNC", "环境与测试租户", "对象",
                  "模拟端准备结果", "人工执行步骤", "模拟端证据", "Ai-DOP 测连结果",
                  "同步/入站流水编号", "贴源核验结果", "目标业务结果", "结论",
                  "未完成项", "清理提示", "脱敏说明"):
        assert label in md
    assert entry["runId"] in md
    assert "SIM-RPT-001" in md


def test_report_markdown_uses_channel_label():
    md = build_report_markdown(_fields(channel="API_INBOUND"))
    assert "方式丙 API_INBOUND（接口推送）" in md


def test_channels_are_the_three_integration_modes():
    assert set(CHANNELS) == {"DB_SYNC", "API_PULL", "API_INBOUND"}


def test_report_store_limit_and_delete():
    store = REPORTS
    entry = store.save(_fields(runId="SIM-RPT-DEL"))
    assert store.get(entry["id"]) is not None
    assert store.delete(entry["id"]) is True
    assert store.get(entry["id"]) is None
    assert store.delete(entry["id"]) is False