zeekerWatchmen/ref/ir_generation_new_skill/tests/test_step2.py

"""
Tests for Stage 2 (IR Extraction).

Validates that ir_fragments.json meets quality and structural requirements:
- All fragments have non-empty rules
- All rules have source references with logic tree nodes
- All trigger conditions have signal/operator/value
- No duplicate rule_ids (across all fragments)
"""

import json
import sys
from pathlib import Path
from collections import Counter

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


PASS = "[PASS]"
FAIL = "[FAIL]"
WARN = "[WARN]"


def load_fragments():
    """Load ir_fragments.json."""
    try:
        return config.load_json(config.IR_FRAGMENTS_JSON)
    except FileNotFoundError:
        print(f"{FAIL} ir_fragments.json 未找到: {config.IR_FRAGMENTS_JSON}")
        print("  请先运行 step2_ir_extraction.py")
        sys.exit(1)


def check_non_empty_rules(fragments: list[dict]) -> list[str]:
    """Every fragment must have at least one rule."""
    errors = []
    for f in fragments:
        uid = f.get("unit_id", "?")
        rules = f.get("rules", [])
        if not rules:
            if f.get("error"):
                errors.append(f"{uid}: 提取失败 — {f['error']}")
            else:
                errors.append(f"{uid}: rules 为空")
    return errors


def check_sources_have_logic_tree_nodes(fragments: list[dict]) -> list[str]:
    """Every rule should reference at least one logic tree node in its sources."""
    errors = []
    for f in fragments:
        uid = f.get("unit_id", "?")
        for j, rule in enumerate(f.get("rules", [])):
            rid = rule.get("rule_id", f"rule[{j}]")
            sources = rule.get("sources", [])
            has_logic_tree = any(
                src.get("type") == "logic_tree" and src.get("node_ids")
                for src in sources
            )
            # NOTE: Some rules might only reference text (e.g., switch-off rules),
            # so we flag as warning rather than error
            if not has_logic_tree:
                has_text = any(
                    src.get("type") in ("table", "para") for src in sources
                )
                if not has_text:
                    errors.append(f"{rid}: sources 中既无逻辑树引用也无文字引用")
    return errors


def check_trigger_conditions(fragments: list[dict]) -> list[str]:
    """Every trigger condition must have signal, operator, value."""
    errors = []
    for f in fragments:
        uid = f.get("unit_id", "?")
        for j, rule in enumerate(f.get("rules", [])):
            rid = rule.get("rule_id", f"rule[{j}]")
            trigger = rule.get("trigger", {})
            conditions = trigger.get("conditions", [])

            # Check if trigger uses 'event' instead of conditions
            if trigger.get("event") is not None:
                continue  # event-based trigger is valid

            for k, cond in enumerate(conditions):
                signal = cond.get("signal", "")
                operator = cond.get("operator", "")
                # value can be 0, False, "", so check with 'in'
                has_value = "value" in cond

                if not signal:
                    errors.append(f"{rid}.condition[{k}]: 缺少 signal")
                if not operator:
                    errors.append(f"{rid}.condition[{k}]: 缺少 operator")
                if not has_value:
                    errors.append(f"{rid}.condition[{k}]: 缺少 value")

    return errors


def check_duplicate_rule_ids(fragments: list[dict]) -> list[str]:
    """Check for duplicate rule_ids across all fragments."""
    all_rule_ids = []
    for f in fragments:
        for rule in f.get("rules", []):
            rid = rule.get("rule_id", "")
            if rid:
                all_rule_ids.append(rid)

    duplicates = [rid for rid, count in Counter(all_rule_ids).items() if count > 1]
    errors = []
    if duplicates:
        errors.append(f"重复 rule_id: {duplicates}")
    return errors


def check_action_types(fragments: list[dict]) -> list[str]:
    """Verify that actions have valid types."""
    valid_types = {"system", "user_interaction"}
    errors = []
    for f in fragments:
        for j, rule in enumerate(f.get("rules", [])):
            rid = rule.get("rule_id", f"rule[{j}]")
            for k, action in enumerate(rule.get("actions", [])):
                atype = action.get("type", "")
                if atype not in valid_types:
                    errors.append(
                        f"{rid}.action[{k}]: type='{atype}' 无效, "
                        f"应为 {valid_types}"
                    )
                if atype == "user_interaction" and "content" not in action:
                    errors.append(
                        f"{rid}.action[{k}]: user_interaction 类型缺少 content 字段"
                    )
    return errors


def run_all_tests():
    print("=" * 60)
    print("Step 2 自检测试")
    print("=" * 60)

    fragments = load_fragments()
    all_errors = []
    total_units = len(fragments)
    total_rules = sum(len(f.get("rules", [])) for f in fragments)

    # Test 1: Non-empty rules
    errors = check_non_empty_rules(fragments)
    if errors:
        print(f"\n{FAIL} 非空规则检查: {len(errors)} 个错误")
        for e in errors:
            print(f"  - {e}")
        all_errors.extend(errors)
    else:
        print(f"\n{PASS} 非空规则检查: 全部通过 ({total_units} 个片段)")

    # Test 2: Sources have logic tree references
    errors = check_sources_have_logic_tree_nodes(fragments)
    if errors:
        print(f"\n{FAIL} 来源节点引用: {len(errors)} 个规则缺少来源引用")
        for e in errors[:10]:  # Show first 10
            print(f"  - {e}")
        if len(errors) > 10:
            print(f"  ... 还有 {len(errors) - 10} 个")
        all_errors.extend(errors)
    else:
        print(f"\n{PASS} 来源节点引用: 全部通过")

    # Test 3: Trigger conditions completeness
    errors = check_trigger_conditions(fragments)
    if errors:
        print(f"\n{FAIL} 触发条件完整性: {len(errors)} 个条件不完整")
        for e in errors[:10]:
            print(f"  - {e}")
        if len(errors) > 10:
            print(f"  ... 还有 {len(errors) - 10} 个")
        all_errors.extend(errors)
    else:
        print(f"\n{PASS} 触发条件完整性: 全部通过")

    # Test 4: No duplicate rule_ids
    errors = check_duplicate_rule_ids(fragments)
    if errors:
        print(f"\n{FAIL} rule_id 唯一性: 发现重复")
        for e in errors:
            print(f"  - {e}")
        all_errors.extend(errors)
    else:
        print(f"\n{PASS} rule_id 唯一性: 全部通过")

    # Test 5: Valid action types
    errors = check_action_types(fragments)
    if errors:
        print(f"\n{FAIL} 动作类型检查: {len(errors)} 个问题")
        for e in errors[:10]:
            print(f"  - {e}")
        all_errors.extend(errors)
    else:
        print(f"\n{PASS} 动作类型检查: 全部通过")

    # Summary
    print(f"\n{'='*60}")
    total_failures = len(all_errors)

    if total_failures == 0:
        print(f"{PASS} 所有测试通过!")
    else:
        print(f"{FAIL} 测试失败: {total_failures} 个错误")
        print("\n建议:")
        print("  1. 检查 ir_fragments.json 中出错的规则")
        print("  2. 如果某些功能单元的规则为空，检查上下文包是否丢失了关键信息")
        print("  3. 调整 Prompt（prompts/step2_ir_extraction.txt）或上下文提取逻辑后重新运行")

    print(f"\n统计:")
    print(f"  功能单元数: {total_units}")
    print(f"  规则总数: {total_rules}")
    error_units = sum(1 for f in fragments if f.get("error"))
    if error_units:
        print(f"  提取失败的单元: {error_units}")

    return total_failures == 0


if __name__ == "__main__":
    success = run_all_tests()
    sys.exit(0 if success else 1)