claudetools/test_context_compression_quick.py

"""Quick functional test for context compression utilities"""
# -*- coding: utf-8 -*-

import sys
import io

# Force UTF-8 output on Windows
if sys.platform == 'win32':
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')

from api.utils.context_compression import (
    compress_conversation_summary,
    create_context_snippet,
    compress_project_state,
    extract_key_decisions,
    calculate_relevance_score,
    merge_contexts,
    format_for_injection,
    extract_tags_from_text,
    compress_file_changes
)
from datetime import datetime, timezone
import json


def test_compress_conversation():
    print("Testing compress_conversation_summary...")
    messages = [
        {"role": "user", "content": "Build authentication with FastAPI"},
        {"role": "assistant", "content": "Completed auth endpoints. Working on testing."}
    ]
    result = compress_conversation_summary(messages)
    print(f"  Phase: {result['phase']}")
    print(f"  Completed: {result['completed']}")
    assert result['phase'] in ['api_development', 'testing']
    print("  ✓ Passed\n")


def test_create_snippet():
    print("Testing create_context_snippet...")
    snippet = create_context_snippet(
        "Using FastAPI for async support",
        snippet_type="decision",
        importance=8
    )
    print(f"  Type: {snippet['type']}")
    print(f"  Tags: {snippet['tags']}")
    print(f"  Relevance: {snippet['relevance_score']}")
    assert snippet['type'] == 'decision'
    assert 'fastapi' in snippet['tags']
    assert snippet['relevance_score'] > 0
    print("  ✓ Passed\n")


def test_extract_tags():
    print("Testing extract_tags_from_text...")
    text = "Using FastAPI with PostgreSQL database and Redis caching"
    tags = extract_tags_from_text(text)
    print(f"  Tags: {tags}")
    assert 'fastapi' in tags
    assert 'postgresql' in tags
    assert 'redis' in tags
    print("  ✓ Passed\n")


def test_extract_decisions():
    print("Testing extract_key_decisions...")
    text = "Decided to use FastAPI because it provides async support"
    decisions = extract_key_decisions(text)
    print(f"  Decisions found: {len(decisions)}")
    if decisions:
        print(f"  First decision: {decisions[0]['decision']}")
        assert 'fastapi' in decisions[0]['decision'].lower()
    print("  ✓ Passed\n")


def test_calculate_relevance():
    print("Testing calculate_relevance_score...")
    snippet = {
        "created_at": datetime.now(timezone.utc).isoformat(),
        "usage_count": 5,
        "importance": 8,
        "tags": ["critical", "api"],
        "last_used": datetime.now(timezone.utc).isoformat()
    }
    score = calculate_relevance_score(snippet)
    print(f"  Score: {score}")
    assert 0 <= score <= 10
    assert score > 8  # Should be boosted
    print("  ✓ Passed\n")


def test_merge_contexts():
    print("Testing merge_contexts...")
    ctx1 = {"phase": "api_dev", "completed": ["auth"]}
    ctx2 = {"phase": "api_dev", "completed": ["auth", "crud"]}
    merged = merge_contexts([ctx1, ctx2])
    print(f"  Merged completed: {merged['completed']}")
    assert "auth" in merged['completed']
    assert "crud" in merged['completed']
    print("  ✓ Passed\n")


def test_compress_project_state():
    print("Testing compress_project_state...")
    state = compress_project_state(
        {"name": "Test", "phase": "dev", "progress_pct": 50},
        "Building API",
        ["api/main.py", "tests/test_api.py"]
    )
    print(f"  Project: {state['project']}")
    print(f"  Files: {len(state['files'])}")
    assert state['project'] == "Test"
    assert state['progress'] == 50
    print("  ✓ Passed\n")


def test_compress_file_changes():
    print("Testing compress_file_changes...")
    files = ["api/auth.py", "tests/test_auth.py", "README.md"]
    compressed = compress_file_changes(files)
    print(f"  Compressed files: {len(compressed)}")
    for f in compressed:
        print(f"    {f['path']} -> {f['type']}")
    assert len(compressed) == 3
    assert compressed[0]['type'] == 'api'
    assert compressed[1]['type'] == 'test'
    assert compressed[2]['type'] == 'doc'
    print("  ✓ Passed\n")


def test_format_for_injection():
    print("Testing format_for_injection...")
    contexts = [
        {
            "type": "decision",
            "content": "Using FastAPI for async support",
            "tags": ["fastapi", "api"],
            "relevance_score": 8.5
        },
        {
            "type": "blocker",
            "content": "Need Redis setup",
            "tags": ["redis", "critical"],
            "relevance_score": 9.0
        }
    ]
    formatted = format_for_injection(contexts, max_tokens=500)
    print(f"  Output length: {len(formatted)} chars")
    print(f"  Contains 'Context Recall': {'Context Recall' in formatted}")
    assert "Context Recall" in formatted
    assert "blocker" in formatted.lower()
    print("  ✓ Passed\n")


def run_all_tests():
    print("=" * 60)
    print("CONTEXT COMPRESSION UTILITIES - FUNCTIONAL TESTS")
    print("=" * 60 + "\n")

    tests = [
        test_compress_conversation,
        test_create_snippet,
        test_extract_tags,
        test_extract_decisions,
        test_calculate_relevance,
        test_merge_contexts,
        test_compress_project_state,
        test_compress_file_changes,
        test_format_for_injection
    ]

    passed = 0
    failed = 0

    for test in tests:
        try:
            test()
            passed += 1
        except Exception as e:
            print(f"  ✗ Failed: {e}\n")
            failed += 1

    print("=" * 60)
    print(f"RESULTS: {passed} passed, {failed} failed")
    print("=" * 60)

    return failed == 0


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