|
1 | | -# import pytest |
2 | | -# from unittest.mock import Mock |
3 | | -# from agent.memory_manager import MemoryManager |
4 | | -# from app_setup import DB_CONNECTION |
5 | | - |
6 | | - |
7 | | -# @pytest.fixture |
8 | | -# def setup(mocker): |
9 | | -# mocker.patch( |
10 | | -# "openai.ChatCompletion.create", |
11 | | -# return_value={"choices": [{"message": {"content": "summary"}}]}, |
12 | | -# ) |
13 | | -# memory_manager = MemoryManager( |
14 | | -# model="gpt-3.5-turbo-0613", |
15 | | -# table_name="test", |
16 | | -# db_connection=DB_CONNECTION, |
17 | | -# ) |
18 | | -# return memory_manager |
19 | | - |
20 | | - |
21 | | -# class TestMemoryManager: |
22 | | -# @pytest.fixture(autouse=True) |
23 | | -# def setup_method(self, setup): |
24 | | -# self.memory_manager = setup |
25 | | - |
26 | | -# def test_get_total_tokens_in_message(self): |
27 | | -# message = "This is a test message." |
28 | | -# tokens = self.memory_manager.get_total_tokens_in_message(message) |
29 | | -# print(f"TOKENS: {tokens}") |
30 | | -# assert isinstance(tokens, int) |
31 | | - |
32 | | -# # def test_add_message(self): |
33 | | -# # role = "user" |
34 | | -# # content = "What's the weather like in Boston?" |
35 | | -# # self.memory_manager.add_message(role, content) |
36 | | -# # messages = self.memory_manager.get_messages() |
37 | | -# # assert messages[0]["role"] == role |
38 | | -# # assert messages[0]["content"] == content |
39 | | - |
40 | | -# def test_get_total_tokens(self): |
41 | | -# tokens = self.memory_manager.get_total_tokens() |
42 | | -# assert isinstance(tokens, int) |
43 | | - |
44 | | -# def test_get_messages(self): |
45 | | -# messages = self.memory_manager.get_messages() |
46 | | -# assert isinstance(messages, list) |
47 | | - |
48 | | -# # def test_summarize(self): |
49 | | -# # message = "This is a test message." |
50 | | -# # summary, tokens = self.memory_manager.summarize(message) |
51 | | -# # assert isinstance(summary, str) |
52 | | -# # assert isinstance(tokens, int) |
53 | | - |
54 | | -# # def test_set_system(self): |
55 | | -# # input = { |
56 | | -# # "system": "You are an AI Pair Programmer and a world class python developer helping the Human work on a project." |
57 | | -# # } |
58 | | -# # self.memory_manager.set_system(input) |
59 | | -# # messages = self.memory_manager.get_messages() |
60 | | -# # assert messages[0]["role"] == "system" |
61 | | -# # assert messages[0]["content"] == input["system"] |
| 1 | +import pytest |
| 2 | +from unittest.mock import Mock |
| 3 | +from agent.memory_manager import MemoryManager |
| 4 | + |
| 5 | + |
| 6 | +@pytest.fixture |
| 7 | +def setup(): |
| 8 | + # Create a mock connection object |
| 9 | + conn = Mock() |
| 10 | + |
| 11 | + # Create a mock cursor object |
| 12 | + cursor = Mock() |
| 13 | + |
| 14 | + # Configure the mock connection to return the mock cursor when cursor() is called |
| 15 | + conn.cursor.return_value = cursor |
| 16 | + |
| 17 | + # Create an instance of MemoryManager with the mock connection |
| 18 | + memory_manager = MemoryManager( |
| 19 | + model="gpt-3.5-turbo-0613", |
| 20 | + table_name="test", |
| 21 | + db_connection=conn, |
| 22 | + ) |
| 23 | + return memory_manager |
| 24 | + |
| 25 | + |
| 26 | +class TestMemoryManager: |
| 27 | + @pytest.fixture(autouse=True) |
| 28 | + def setup_method(self, setup): |
| 29 | + self.memory_manager = setup |
| 30 | + |
| 31 | + def test_get_total_tokens_in_message(self): |
| 32 | + message = "This is a test message." |
| 33 | + tokens = self.memory_manager.get_total_tokens_in_message(message) |
| 34 | + print(f"TOKENS: {tokens}") |
| 35 | + assert isinstance(tokens, int) |
| 36 | + |
| 37 | + def test_get_total_tokens(self): |
| 38 | + tokens = self.memory_manager.get_total_tokens() |
| 39 | + assert ( |
| 40 | + tokens >= 0 |
| 41 | + ), f"Expected tokens to be greater than or equal to 0, but got {tokens}" |
| 42 | + assert isinstance(tokens, int) |
| 43 | + |
| 44 | + def test_get_messages(self): |
| 45 | + # Configure the mock cursor's fetchall method to return a list |
| 46 | + cursor = self.memory_manager.cur |
| 47 | + cursor.fetchall.side_effect = [ |
| 48 | + [("role1", "system_prompt")], # First call to fetchall |
| 49 | + [ # Second call to fetchall |
| 50 | + ("user", "full_user_message", "user_message", 10), |
| 51 | + ("assistant", "full_assistant_message", "assistant_message", 20), |
| 52 | + ], |
| 53 | + ] |
| 54 | + |
| 55 | + assert self.memory_manager.get_messages() == [ |
| 56 | + {"role": "role1", "content": "system_prompt"}, |
| 57 | + { |
| 58 | + "role": "assistant", |
| 59 | + "content": "assistant_message", |
| 60 | + "full_content": "full_assistant_message", |
| 61 | + }, |
| 62 | + { |
| 63 | + "role": "user", |
| 64 | + "content": "user_message", |
| 65 | + "full_content": "full_user_message", |
| 66 | + }, |
| 67 | + ] |
| 68 | + |
| 69 | + def test_set_system(self): |
| 70 | + input = { |
| 71 | + "system": "You are an AI Pair Programmer and a world class python developer helping the Human work on a project." |
| 72 | + } |
| 73 | + self.memory_manager.set_system(input) |
| 74 | + self.memory_manager.cur.execute.assert_called() # Check if execute was called on cursor |
| 75 | + |
| 76 | + def test_add_message(self): |
| 77 | + self.memory_manager.add_message("test_role", "test_content") |
| 78 | + self.memory_manager.cur.execute.assert_called() |
| 79 | + self.memory_manager.conn.commit.assert_called() |
| 80 | + call_args = self.memory_manager.cur.execute.call_args |
| 81 | + sql_query, params = call_args[0] |
| 82 | + assert type(params[0]) == str |
| 83 | + assert params[1] == "test_role" |
| 84 | + assert params[2] == "test_content" |
| 85 | + assert ( |
| 86 | + params[3] > 0 |
| 87 | + ), f"Expected tokens to be greater than 0, but got {params[3]}" |
| 88 | + assert params[4] is None |
| 89 | + assert params[5] is None |
| 90 | + assert params[6] is not None |
0 commit comments