File size: 3,463 Bytes
79d4fd5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
import pytest

from integration.context_manager import (
    ContextManager,
    get_context_manager,
    reset_context_manager,
    MAX_HISTORY_MESSAGES,
)


class TestCreditFileIdTracking:

    def setup_method(self):
        self.cm = ContextManager()

    def test_initially_none(self):
        assert self.cm.get_active_credit_file_id() is None

    def test_set_and_get(self):
        self.cm.set_active_credit_file_id("26000046")
        assert self.cm.get_active_credit_file_id() == "26000046"

    def test_strips_whitespace(self):
        self.cm.set_active_credit_file_id("  26000046  ")
        assert self.cm.get_active_credit_file_id() == "26000046"

    def test_ignores_empty_string(self):
        self.cm.set_active_credit_file_id("26000046")
        self.cm.set_active_credit_file_id("")
        assert self.cm.get_active_credit_file_id() == "26000046"

    def test_ignores_none(self):
        self.cm.set_active_credit_file_id("26000046")
        self.cm.set_active_credit_file_id(None)
        assert self.cm.get_active_credit_file_id() == "26000046"

    def test_updates_to_new_id(self):
        self.cm.set_active_credit_file_id("26000046")
        self.cm.set_active_credit_file_id("99999999")
        assert self.cm.get_active_credit_file_id() == "99999999"


class TestPrepareContext:

    def setup_method(self):
        self.cm = ContextManager()

    def test_empty_history(self):
        result = self.cm.prepare_context("test", [])
        assert result["chat_history"] == []
        assert result["resolved_query"] == "test"

    def test_trims_to_max(self):
        history = [{"role": "user", "content": f"msg {i}"} for i in range(20)]
        result = self.cm.prepare_context("q", history)
        assert len(result["chat_history"]) == MAX_HISTORY_MESSAGES

    def test_keeps_all_when_under_max(self):
        history = [{"role": "user", "content": f"msg {i}"} for i in range(5)]
        result = self.cm.prepare_context("q", history)
        assert len(result["chat_history"]) == 5

    def test_keeps_most_recent(self):
        history = [{"role": "user", "content": f"msg {i}"} for i in range(15)]
        result = self.cm.prepare_context("q", history)
        last_content = result["chat_history"][-1]["content"]
        assert last_content == "msg 14"

    def test_metadata_present(self):
        result = self.cm.prepare_context("q", [{"role": "user", "content": "x"}])
        assert "metadata" in result
        assert result["metadata"]["filtering_method"] == "recent_window"
        assert result["metadata"]["kept"] == 1

    def test_none_history_treated_as_empty(self):
        result = self.cm.prepare_context("q", None)
        assert result["chat_history"] == []

    def test_intent_param_accepted(self):
        result = self.cm.prepare_context("q", [], intent="database_query")
        assert result["resolved_query"] == "q"


class TestSingleton:

    def setup_method(self):
        reset_context_manager()

    def teardown_method(self):
        reset_context_manager()

    def test_returns_instance(self):
        cm = get_context_manager()
        assert isinstance(cm, ContextManager)

    def test_returns_same_instance(self):
        cm1 = get_context_manager()
        cm2 = get_context_manager()
        assert cm1 is cm2

    def test_reset_creates_new(self):
        cm1 = get_context_manager()
        reset_context_manager()
        cm2 = get_context_manager()
        assert cm1 is not cm2