tests/test_fast_reply_identity.py · hermescures1/splitbit-llm at main

splitbit-llm / tests /test_fast_reply_identity.py

Upload folder using huggingface_hub

948a05a verified 3 months ago

6.26 kB

"""Test fast reply cache, first-run naming, and 100-project mode."""

import sys
import os
import tempfile
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from splitbit_llm.memory.fast_cache import FastReplyCache
from splitbit_llm.identity import FirstRunManager


def test_fast_reply_cache():
    """Test fast reply cache — store, lookup, cache hit."""
    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
        db_path = f.name

    cache = FastReplyCache(db_path=db_path)

    # Store a response
    cache.store("What is Python?", "Python is a programming language.", confidence=0.96, response_time_s=0.5)

    # Exact match lookup
    result = cache.lookup("What is Python?")
    assert result is not None, "Expected cache hit"
    assert result["cache_hit"] is True
    assert result["cache_type"] == "exact"
    assert "Python is a programming language" in result["response"]
    print(f"  Exact hit: {result['cache_type']} (confidence: {result['confidence']:.2f})")

    # Miss for different query
    result2 = cache.lookup("What is JavaScript?")
    assert result2 is None, "Expected cache miss"
    print(f"  Miss for different query: OK")

    # Store more entries
    cache.store("What is JavaScript?", "JavaScript is a web programming language.", confidence=0.95)
    cache.store("How do I code?", "Write code in a text editor and run it.", confidence=0.90)
    cache.store("What is AI?", "AI is artificial intelligence.", confidence=0.97)

    # Semantic match (similar words)
    result3 = cache.lookup("What is Python programming?")
    # May or may not hit depending on threshold, but should not crash
    print(f"  Semantic lookup: {'hit' if result3 else 'miss'} (OK)")

    # Stats
    stats = cache.get_stats()
    assert stats["entries_stored"] >= 4
    assert stats["total_lookups"] >= 3
    print(f"  Stats: {stats['entries_stored']} entries, hit rate: {stats['hit_rate']:.1%}")

    # Persistence
    cache2 = FastReplyCache(db_path=db_path)
    stats2 = cache2.get_stats()
    assert stats2["entries_stored"] >= 4, f"Cache not persisted: {stats2}"
    print(f"  Persisted: {stats2['entries_stored']} entries")


def test_fast_cache_normalization():
    """Test query normalization for cache."""
    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
        db_path = f.name

    cache = FastReplyCache(db_path=db_path)
    cache.store("What is Python?", "Answer", confidence=0.96)

    # Normalized version should also hit
    result = cache.lookup("what is python?")
    assert result is not None, "Case difference should still hit"
    print(f"  Case-insensitive hit: OK")

    result2 = cache.lookup("What  is  Python?")
    assert result2 is not None, "Extra spaces should still hit"
    print(f"  Whitespace-normalized hit: OK")

    result3 = cache.lookup("What is Python!")
    assert result3 is not None, "Punctuation difference should still hit"
    print(f"  Punctuation-normalized hit: OK")


def test_first_run_naming():
    """Test first-run naming flow."""
    with tempfile.TemporaryDirectory() as tmpdir:
        # First run — no identity file
        identity = FirstRunManager(data_dir=tmpdir)
        assert identity.is_first_run() is True
        greeting = identity.get_greeting()
        assert "Incentives Inc." in greeting
        assert "name me" in greeting.lower()
        print(f"  First run greeting: {greeting}")

        # Set name
        identity.set_name("Jarvis")
        assert identity.is_first_run() is False
        assert identity.get_name() == "Jarvis"
        print(f"  Named: {identity.get_name()}")

        # System prompt suffix
        suffix = identity.get_system_prompt_suffix()
        assert "Jarvis" in suffix
        assert "Incentives Inc." in suffix
        print(f"  System prompt suffix: {suffix.strip()}")

        # Welcome back greeting
        greeting2 = identity.get_greeting()
        assert "Jarvis" in greeting2
        print(f"  Welcome back: {greeting2}")

        # Persistence — load again
        identity2 = FirstRunManager(data_dir=tmpdir)
        assert identity2.is_first_run() is False
        assert identity2.get_name() == "Jarvis"
        print(f"  Persisted name: {identity2.get_name()}")


def test_first_run_no_name():
    """Test first-run when user doesn't provide a name."""
    with tempfile.TemporaryDirectory() as tmpdir:
        identity = FirstRunManager(data_dir=tmpdir)
        assert identity.is_first_run()

        # Default name when not set
        assert identity.get_name() == "Incentives Inc. LLM"
        print(f"  Default name: {identity.get_name()}")

        # System prompt without name
        suffix = identity.get_system_prompt_suffix()
        assert "Incentives Inc." in suffix
        assert "Your name is" not in suffix
        print(f"  Default suffix: {suffix.strip()}")


def test_cache_hit_rate():
    """Test cache hit rate calculation."""
    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
        db_path = f.name

    cache = FastReplyCache(db_path=db_path)

    # Store some entries
    cache.store("Q1", "A1", confidence=0.96)
    cache.store("Q2", "A2", confidence=0.96)
    cache.store("Q3", "A3", confidence=0.96)

    # 3 hits, 2 misses = 60% hit rate
    cache.lookup("Q1")  # hit
    cache.lookup("Q2")  # hit
    cache.lookup("Q3")  # hit
    cache.lookup("Q4")  # miss
    cache.lookup("Q5")  # miss

    rate = cache.get_hit_rate()
    assert 0.5 < rate < 0.7, f"Hit rate should be ~0.6, got {rate}"
    print(f"  Hit rate: {rate:.1%} (5 lookups, 3 hits)")

    # Not fast ready yet (too few entries)
    assert cache.is_fast_ready() is False
    print(f"  Fast ready: {cache.is_fast_ready()} (needs 50+ entries)")


if __name__ == "__main__":
    print("Running fast cache, identity, and 100-project tests...")
    test_fast_reply_cache()
    print("  ✓ test_fast_reply_cache")
    test_fast_cache_normalization()
    print("  ✓ test_fast_cache_normalization")
    test_first_run_naming()
    print("  ✓ test_first_run_naming")
    test_first_run_no_name()
    print("  ✓ test_first_run_no_name")
    test_cache_hit_rate()
    print("  ✓ test_cache_hit_rate")
    print("\nAll fast reply & identity tests passed!")

	"""Test fast reply cache, first-run naming, and 100-project mode."""

	import sys
	import os
	import tempfile
	sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

	from splitbit_llm.memory.fast_cache import FastReplyCache
	from splitbit_llm.identity import FirstRunManager


	def test_fast_reply_cache():
	"""Test fast reply cache — store, lookup, cache hit."""
	with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
	db_path = f.name

	cache = FastReplyCache(db_path=db_path)

	# Store a response
	cache.store("What is Python?", "Python is a programming language.", confidence=0.96, response_time_s=0.5)

	# Exact match lookup
	result = cache.lookup("What is Python?")
	assert result is not None, "Expected cache hit"
	assert result["cache_hit"] is True
	assert result["cache_type"] == "exact"
	assert "Python is a programming language" in result["response"]
	print(f" Exact hit: {result['cache_type']} (confidence: {result['confidence']:.2f})")

	# Miss for different query
	result2 = cache.lookup("What is JavaScript?")
	assert result2 is None, "Expected cache miss"
	print(f" Miss for different query: OK")

	# Store more entries
	cache.store("What is JavaScript?", "JavaScript is a web programming language.", confidence=0.95)
	cache.store("How do I code?", "Write code in a text editor and run it.", confidence=0.90)
	cache.store("What is AI?", "AI is artificial intelligence.", confidence=0.97)

	# Semantic match (similar words)
	result3 = cache.lookup("What is Python programming?")
	# May or may not hit depending on threshold, but should not crash
	print(f" Semantic lookup: {'hit' if result3 else 'miss'} (OK)")

	# Stats
	stats = cache.get_stats()
	assert stats["entries_stored"] >= 4
	assert stats["total_lookups"] >= 3
	print(f" Stats: {stats['entries_stored']} entries, hit rate: {stats['hit_rate']:.1%}")

	# Persistence
	cache2 = FastReplyCache(db_path=db_path)
	stats2 = cache2.get_stats()
	assert stats2["entries_stored"] >= 4, f"Cache not persisted: {stats2}"
	print(f" Persisted: {stats2['entries_stored']} entries")


	def test_fast_cache_normalization():
	"""Test query normalization for cache."""
	with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
	db_path = f.name

	cache = FastReplyCache(db_path=db_path)
	cache.store("What is Python?", "Answer", confidence=0.96)

	# Normalized version should also hit
	result = cache.lookup("what is python?")
	assert result is not None, "Case difference should still hit"
	print(f" Case-insensitive hit: OK")

	result2 = cache.lookup("What is Python?")
	assert result2 is not None, "Extra spaces should still hit"
	print(f" Whitespace-normalized hit: OK")

	result3 = cache.lookup("What is Python!")
	assert result3 is not None, "Punctuation difference should still hit"
	print(f" Punctuation-normalized hit: OK")


	def test_first_run_naming():
	"""Test first-run naming flow."""
	with tempfile.TemporaryDirectory() as tmpdir:
	# First run — no identity file
	identity = FirstRunManager(data_dir=tmpdir)
	assert identity.is_first_run() is True
	greeting = identity.get_greeting()
	assert "Incentives Inc." in greeting
	assert "name me" in greeting.lower()
	print(f" First run greeting: {greeting}")

	# Set name
	identity.set_name("Jarvis")
	assert identity.is_first_run() is False
	assert identity.get_name() == "Jarvis"
	print(f" Named: {identity.get_name()}")

	# System prompt suffix
	suffix = identity.get_system_prompt_suffix()
	assert "Jarvis" in suffix
	assert "Incentives Inc." in suffix
	print(f" System prompt suffix: {suffix.strip()}")

	# Welcome back greeting
	greeting2 = identity.get_greeting()
	assert "Jarvis" in greeting2
	print(f" Welcome back: {greeting2}")

	# Persistence — load again
	identity2 = FirstRunManager(data_dir=tmpdir)
	assert identity2.is_first_run() is False
	assert identity2.get_name() == "Jarvis"
	print(f" Persisted name: {identity2.get_name()}")


	def test_first_run_no_name():
	"""Test first-run when user doesn't provide a name."""
	with tempfile.TemporaryDirectory() as tmpdir:
	identity = FirstRunManager(data_dir=tmpdir)
	assert identity.is_first_run()

	# Default name when not set
	assert identity.get_name() == "Incentives Inc. LLM"
	print(f" Default name: {identity.get_name()}")

	# System prompt without name
	suffix = identity.get_system_prompt_suffix()
	assert "Incentives Inc." in suffix
	assert "Your name is" not in suffix
	print(f" Default suffix: {suffix.strip()}")


	def test_cache_hit_rate():
	"""Test cache hit rate calculation."""
	with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
	db_path = f.name

	cache = FastReplyCache(db_path=db_path)

	# Store some entries
	cache.store("Q1", "A1", confidence=0.96)
	cache.store("Q2", "A2", confidence=0.96)
	cache.store("Q3", "A3", confidence=0.96)

	# 3 hits, 2 misses = 60% hit rate
	cache.lookup("Q1") # hit
	cache.lookup("Q2") # hit
	cache.lookup("Q3") # hit
	cache.lookup("Q4") # miss
	cache.lookup("Q5") # miss

	rate = cache.get_hit_rate()
	assert 0.5 < rate < 0.7, f"Hit rate should be ~0.6, got {rate}"
	print(f" Hit rate: {rate:.1%} (5 lookups, 3 hits)")

	# Not fast ready yet (too few entries)
	assert cache.is_fast_ready() is False
	print(f" Fast ready: {cache.is_fast_ready()} (needs 50+ entries)")


	if __name__ == "__main__":
	print("Running fast cache, identity, and 100-project tests...")
	test_fast_reply_cache()
	print(" ✓ test_fast_reply_cache")
	test_fast_cache_normalization()
	print(" ✓ test_fast_cache_normalization")
	test_first_run_naming()
	print(" ✓ test_first_run_naming")
	test_first_run_no_name()
	print(" ✓ test_first_run_no_name")
	test_cache_hit_rate()
	print(" ✓ test_cache_hit_rate")
	print("\nAll fast reply & identity tests passed!")