File size: 6,264 Bytes
948a05a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
"""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!")