LQAgentPlatform/core/debug/test_prompt_manager.py

"""
C04 PromptManager self-test script.

Tests all test cases from TC-C04-API-001 through TC-C04-ERROR-003.
Run from project root: python core/debug/test_prompt_manager.py
"""

import sys
import os
import json
import shutil
import traceback

PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if PROJECT_ROOT not in sys.path:
    sys.path.insert(0, PROJECT_ROOT)

from core.debug.prompt_manager import (
    PromptManager,
    PROMPT_DIR,
    VERSIONS_DIR,
    PROMPT_FILE_MAP,
    CHAINS,
    _extract_variables,
    _make_diff_lines,
)

PASS = 0
FAIL = 0
ERRORS = []


def check(cond, msg):
    global PASS, FAIL
    if cond:
        PASS += 1
        print(f'  [PASS] {msg}')
    else:
        FAIL += 1
        print(f'  [FAIL] {msg}')
        ERRORS.append(msg)


def section(title):
    print(f'\n{"=" * 60}')
    print(f'  {title}')
    print(f'{"=" * 60}')


# ================================================================
# Setup: backup existing versions directory
# ================================================================
section('Setup test environment')

versions_backup = None
if os.path.exists(VERSIONS_DIR):
    import tempfile
    versions_backup = VERSIONS_DIR + '_backup_' + os.urandom(4).hex()
    shutil.copytree(VERSIONS_DIR, versions_backup)
    shutil.rmtree(VERSIONS_DIR)
    print(f'  Backed up old versions dir to: {versions_backup}')

manager = PromptManager()

# Capture initial state for later verification (before any save/activate modifies it)
INITIAL_CC_V1 = manager._load_version_file('completeness_check', 'v1.0')
INITIAL_CC_MAIN = manager._read_current_from_main('completeness_check')

print(f'  PROMPT_DIR: {PROMPT_DIR}')
print(f'  VERSIONS_DIR: {VERSIONS_DIR}')
print(f'  Total prompts: {len(PROMPT_FILE_MAP)}')
print(f'  Chains: {CHAINS}')

# ================================================================
# TC-C04-API-001: Get prompt list with version info
# ================================================================
section('TC-C04-API-001: Get prompt list')

try:
    all_prompts = manager.get_all_prompts()
    check(len(all_prompts) >= 8, f'items count >= 8 (actual: {len(all_prompts)})')

    if all_prompts:
        item = all_prompts[0]
        check('name' in item, 'items[0] has name')
        check('version' in item, 'items[0] has version')
        check('time' in item, 'items[0] has time')
        check('chain' in item, 'items[0] has chain')
        check('is_current' in item, 'items[0] has is_current')
        check('note' in item, 'items[0] has note')

    names = [i['name'] for i in all_prompts]
    check('completeness_check' in names, 'list contains completeness_check')

    check(len(CHAINS) >= 7, f'chains count {len(CHAINS)} >= 7')
    check('完整性' in CHAINS, 'chains contains 完整性')
    check('时效性' in CHAINS, 'chains contains 时效性')
    check('规范性' in CHAINS, 'chains contains 规范性')
    check('敏感词' in CHAINS, 'chains contains 敏感词')
    check('语义逻辑' in CHAINS, 'chains contains 语义逻辑')
    check('语法' in CHAINS, 'chains contains 语法')
    check('专业性' in CHAINS, 'chains contains 专业性')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-API-001 exception: {e}')

# ================================================================
# TC-C04-API-002: Get prompt detail
# ================================================================
section('TC-C04-API-002: Get prompt detail')

try:
    detail = manager.get_prompt_detail('completeness_check')
    check(detail is not None, 'get_prompt_detail returns non-None')

    if detail:
        check('name' in detail and detail['name'] == 'completeness_check', 'detail has name')
        check('version' in detail, 'detail has version')
        check('time' in detail, 'detail has time')
        check('chain' in detail, 'detail has chain')
        check('is_current' in detail, 'detail has is_current')
        check('system_prompt' in detail and len(detail['system_prompt']) > 0, 'system_prompt not empty')
        check('user_prompt' in detail and len(detail['user_prompt']) > 0, 'user_prompt not empty')
        check('note' in detail, 'detail has note')
        check('variables' in detail, 'detail has variables')
        check('file_path' in detail, 'detail has file_path')

        variables = detail.get('variables', [])
        check(len(variables) > 0, f'variables has {len(variables)} items: {variables}')
        # review_content comes from user_prompt; review_references is in system_prompt only
        check('review_content' in variables, 'variables contains review_content')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-API-002 exception: {e}')

# ================================================================
# TC-C04-API-003: Save new version and set as current
# ================================================================
section('TC-C04-API-003: Save new version')

try:
    detail_before = manager.get_prompt_detail('completeness_check')
    print(f'  Version before save: {detail_before["version"] if detail_before else "N/A"}')

    new_system = '你是一个专业的施工方案完整性审查专家。测试新版本内容。'
    new_user = '请审查以下施工方案内容的完整性：\n方案内容：{review_content}\n参考依据：{review_references}'
    result = manager.save_new_version(
        'completeness_check',
        system_prompt=new_system,
        user_prompt=new_user,
        note='测试保存新版本',
        set_current=True,
    )

    check(result is not None, 'save_new_version returns non-None')
    version = result.get('version', '')
    check(version.startswith('v'), f'version format: {version}')

    major = int(version.lstrip('v').split('.')[0])
    check(major >= 2, f'version incremented: {version} (major >= 2)')

    check('name' in result and result['name'] == 'completeness_check', 'result has name')

    import glob
    ver_files = glob.glob(os.path.join(VERSIONS_DIR, 'completeness_check', '*.yaml'))
    check(len(ver_files) >= 2, f'version files >= 2 (actual: {len(ver_files)})')

    detail_after = manager.get_prompt_detail('completeness_check')
    if detail_after:
        check(detail_after['system_prompt'] == new_system, 'main file system_prompt updated')
        check(detail_after['user_prompt'] == new_user, 'main file user_prompt updated')

    new_ver_data = manager._load_version_file('completeness_check', version)
    if new_ver_data:
        check(new_ver_data.get('is_current') == True, f'version file {version} is_current=true')

    print(f'  Version after save: {version}')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-API-003 exception: {e}')

# ================================================================
# TC-C04-API-004: Version comparison (Diff)
# ================================================================
section('TC-C04-API-004: Version comparison (Diff)')

try:
    versions = manager.get_versions('completeness_check')
    check(len(versions) >= 2, f'at least 2 versions (actual: {len(versions)})')

    if len(versions) >= 2:
        v1 = versions[-1]['version']
        v2 = versions[0]['version']
        diff_result = manager.compare_versions('completeness_check', v1, v2)

        check(diff_result is not None, 'compare_versions returns non-None')
        check('name' in diff_result, 'diff has name')
        check('base_version' in diff_result, 'diff has base_version')
        check('target_version' in diff_result, 'diff has target_version')
        check('diffs' in diff_result, 'diff has diffs')

        diffs = diff_result.get('diffs', [])
        check(len(diffs) == 2, f'diffs has 2 sections (actual: {len(diffs)})')

        for d in diffs:
            check('section' in d, 'diff item has section')
            check('type' in d, 'diff item has type')
            check('lines' in d, 'diff item has lines')
            check(d['type'] == 'text_diff', f'diff type is text_diff')

        sys_diff = [d for d in diffs if d['section'] == 'system_prompt']
        if sys_diff:
            lines = sys_diff[0]['lines']
            check(len(lines) > 0, f'system_prompt diff lines > 0 (actual: {len(lines)})')

            for line in lines[:3]:
                check('type' in line and line['type'] in ('add', 'del', 'ctx'),
                      f'line has valid type ({line.get("type")})')
                check('text' in line, 'line has text')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-API-004 exception: {e}')

# ================================================================
# TC-C04-API-005: Activate version
# ================================================================
section('TC-C04-API-005: Activate version')

try:
    versions = manager.get_versions('completeness_check')
    check(len(versions) >= 2, f'at least 2 versions (actual: {len(versions)})')

    if len(versions) >= 2:
        oldest_ver = versions[-1]['version']
        print(f'  Activating version: {oldest_ver}')

        old_detail = manager.get_prompt_detail('completeness_check', version=oldest_ver)
        check(old_detail is not None, f'can read {oldest_ver} version')

        if old_detail:
            old_system = old_detail['system_prompt']

        act_result = manager.activate_version('completeness_check', oldest_ver)
        check(act_result.get('success') == True, 'activate_version returns success=true')
        check(act_result.get('name') == 'completeness_check', 'result has name')
        check(act_result.get('version') == oldest_ver, f'result version is {oldest_ver}')

        current = manager.get_prompt_detail('completeness_check')
        if current and old_detail:
            check(current['system_prompt'] == old_detail['system_prompt'],
                  'main file system_prompt matches activated version')

        if old_detail:
            ver_data = manager._load_version_file('completeness_check', oldest_ver)
            if ver_data:
                check(ver_data.get('is_current') == True,
                      f'{oldest_ver} has is_current=true')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-API-005 exception: {e}')

# ================================================================
# TC-C04-API-006: Rollback version
# ================================================================
section('TC-C04-API-006: Rollback version')

try:
    rollback_result = manager.rollback_version('completeness_check', 'v1.0')
    check(rollback_result.get('success') == True, 'rollback_version returns success=true')
    check(rollback_result.get('name') == 'completeness_check', 'result has name')
    check(rollback_result.get('version') == 'v1.0', 'result version is v1.0')

    current = manager.get_prompt_detail('completeness_check')
    v1_detail = manager.get_prompt_detail('completeness_check', version='v1.0')
    if current and v1_detail:
        check(current['system_prompt'] == v1_detail['system_prompt'],
              'after rollback, main file system_prompt matches v1.0')
        check(current['user_prompt'] == v1_detail['user_prompt'],
              'after rollback, main file user_prompt matches v1.0')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-API-006 exception: {e}')

# ================================================================
# TC-C04-EDGE-001: No version specified returns current
# ================================================================
section('TC-C04-EDGE-001: No version returns current')

try:
    result = manager.save_new_version(
        'completeness_check',
        system_prompt='EDGE-001 test new version content',
        user_prompt='Review content: {review_content}',
        note='EDGE-001 test',
        set_current=True,
    )
    print(f'  Current version: {result["version"]}')

    detail_no_ver = manager.get_prompt_detail('completeness_check')
    if detail_no_ver:
        check(detail_no_ver['is_current'] == True, 'no version param: is_current=True')
        check(detail_no_ver['version'] == result['version'],
              f'returns current version ({detail_no_ver["version"]})')

        detail_with_ver = manager.get_prompt_detail('completeness_check', version='v1.0')
        if detail_with_ver:
            check(detail_with_ver['is_current'] == False,
                  'with v1.0 param: is_current=False (current is later version)')
            check(detail_with_ver['version'] == 'v1.0',
                  f'with v1.0 param: version=v1.0')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-EDGE-001 exception: {e}')

# ================================================================
# TC-C04-EDGE-002: Special characters in prompt name
# ================================================================
section('TC-C04-EDGE-002: Special chars in prompt name')

try:
    detail_special = manager.get_prompt_detail('non_parameter_compliance_check')
    check(detail_special is not None,
          'get_prompt_detail("non_parameter_compliance_check") returns non-None')

    if detail_special:
        check(len(detail_special.get('system_prompt', '')) > 0, 'system_prompt not empty')
        check(len(detail_special.get('user_prompt', '')) > 0, 'user_prompt not empty')

    ver_files = manager._list_version_files('non_parameter_compliance_check')
    check(len(ver_files) >= 1,
          f'version files >= 1 (actual: {len(ver_files)})')

    expected_dir = os.path.join(VERSIONS_DIR, 'non_parameter_compliance_check')
    check(os.path.exists(expected_dir),
          f'version dir exists: {expected_dir}')

    if ver_files:
        ver_path = os.path.join(expected_dir, ver_files[0])
        check(os.path.exists(ver_path), f'version file exists: {ver_files[0]}')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-EDGE-002 exception: {e}')

# ================================================================
# TC-C04-ERROR-001: Empty system prompt handled by API layer
# ================================================================
section('TC-C04-ERROR-001: Empty system prompt')

try:
    try:
        manager.save_new_version(
            name='nonexistent_prompt',
            system_prompt='test',
            user_prompt='test',
            note='',
        )
        check(False, 'nonexistent prompt should raise ValueError')
    except ValueError:
        check(True, 'nonexistent prompt raises ValueError')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-ERROR-001 exception: {e}')

# ================================================================
# TC-C04-ERROR-002: Non-existent prompt name returns None/empty
# ================================================================
section('TC-C04-ERROR-002: Non-existent prompt name')

try:
    detail = manager.get_prompt_detail('nonexistent_check')
    check(detail is None, 'nonexistent prompt detail returns None')

    versions = manager.get_versions('nonexistent_check')
    check(isinstance(versions, list), 'nonexistent prompt versions returns list')
    check(len(versions) == 0, 'nonexistent prompt versions returns empty list')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-ERROR-002 exception: {e}')

# ================================================================
# TC-C04-ERROR-003: Corrupted version file gracefully skipped
# ================================================================
section('TC-C04-ERROR-003: Corrupted version file')

try:
    completeness_dir = os.path.join(VERSIONS_DIR, 'completeness_check')

    bad_file = os.path.join(completeness_dir, 'corrupt.yaml')
    with open(bad_file, 'w', encoding='utf-8') as f:
        f.write('{invalid: yaml: content\n  broken indent\n')

    try:
        versions = manager.get_versions('completeness_check')
        check(True, f'corrupted file skipped gracefully ({len(versions)} valid versions)')

        for v in versions:
            check(v['version'] != 'corrupt', 'corrupt file not in version list')
    except Exception as e:
        check(False, f'corrupted file should not raise exception: {e}')

    if os.path.exists(bad_file):
        os.remove(bad_file)

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'TC-C04-ERROR-003 exception: {e}')

# ================================================================
# Extra: _extract_variables
# ================================================================
section('Extra: _extract_variables utility')

try:
    vars = _extract_variables('Hello {name}, age is {age}')
    check('name' in vars and 'age' in vars, f'extracted: {vars}')

    vars_empty = _extract_variables('no variables')
    check(len(vars_empty) == 0, 'no variables returns empty list')

    vars_multi = _extract_variables('{a} {b} {c} {a}')
    check(len(set(vars_multi)) == 3, f'3 unique vars: {set(vars_multi)}')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'_extract_variables exception: {e}')

# ================================================================
# Extra: _make_diff_lines
# ================================================================
section('Extra: _make_diff_lines utility')

try:
    a = 'line1\nline2\nline3'
    b = 'line1\nline2_modified\nline3'

    lines = _make_diff_lines(a, b)
    check(len(lines) > 0, f'Diff has {len(lines)} lines')

    has_ctx = any(l['type'] == 'ctx' for l in lines)
    has_del = any(l['type'] == 'del' for l in lines)
    has_add = any(l['type'] == 'add' for l in lines)
    check(has_ctx, 'Diff has ctx lines')
    check(has_del, 'Diff has del lines')
    check(has_add, 'Diff has add lines')

    same_diff = _make_diff_lines('abc', 'abc')
    check(len(same_diff) == 0, 'identical content => empty diff')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'_make_diff_lines exception: {e}')

# ================================================================
# Extra: get_all_prompts filtering
# ================================================================
section('Extra: get_all_prompts filtering')

try:
    all_items = manager.get_all_prompts()

    completeness_items = manager.get_all_prompts(chain_filter='完整性')
    for item in completeness_items:
        check(item['chain'] == '完整性', f'filtered chain 完整性 (actual: {item["chain"]})')

    search_items = manager.get_all_prompts(search='completeness')
    for item in search_items:
        check('completeness' in item['name'].lower(),
              f'search result contains completeness (actual: {item["name"]})')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'Filtering exception: {e}')

# ================================================================
# Extra: First run initialization (run early, before state changes)
# ================================================================
section('Extra: First run initialization')

try:
    for prompt_name in PROMPT_FILE_MAP:
        version_dir = os.path.join(VERSIONS_DIR, prompt_name)
        check(os.path.exists(version_dir), f'{prompt_name} version dir exists')

        v1_file = os.path.join(version_dir, 'v1.0.yaml')
        check(os.path.exists(v1_file), f'{prompt_name} v1.0.yaml exists')

    # Verify v1.0 matches main file at initialization time
    # (Note: run this BEFORE save/activate tests that modify main file)
    if INITIAL_CC_V1 and INITIAL_CC_MAIN:
        check(INITIAL_CC_V1['system_prompt'] == INITIAL_CC_MAIN['system_prompt'],
              'v1.0 system_prompt matches main file at init')
        check(INITIAL_CC_V1['user_prompt_template'] == INITIAL_CC_MAIN['user_prompt'],
              'v1.0 user_prompt matches main file at init')

except Exception as e:
    print(f'  [EXCEPTION] {e}')
    traceback.print_exc()
    FAIL += 1
    ERRORS.append(f'Initialization exception: {e}')

# ================================================================
# Summary
# ================================================================
section('Test Results')

total = PASS + FAIL
print(f'  Total: {total}  Passed: {PASS}  Failed: {FAIL}')
if ERRORS:
    print(f'\n  Failed details:')
    for i, err in enumerate(ERRORS, 1):
        print(f'    {i}. {err}')

print(f'\n  {"=" * 20} {"ALL PASSED!" if FAIL == 0 else "SOME FAILED!"} {"=" * 20}')

# ================================================================
# Cleanup: restore original versions directory
# ================================================================
if versions_backup and os.path.exists(versions_backup):
    if os.path.exists(VERSIONS_DIR):
        shutil.rmtree(VERSIONS_DIR)
    shutil.copytree(versions_backup, VERSIONS_DIR)
    shutil.rmtree(versions_backup)
    print(f'\n  Restored original versions directory')
else:
    print(f'\n  Note: test-created version files left at {VERSIONS_DIR}')

sys.exit(0 if FAIL == 0 else 1)