repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
public Clawd ADK gateway launch mirror
stars
latest
clone command
git clone gitlawb://did:key:z6Mkq5mY...iFZ5/my-project-publ...git clone gitlawb://did:key:z6Mkq5mY.../my-project-publ...2fa351d6docs: add automaton and perps launch sources16d ago| #1 | import pytest |
| #2 | |
| #3 | from embedchain.memory.base import ChatHistory |
| #4 | from embedchain.memory.message import ChatMessage |
| #5 | |
| #6 | |
| #7 | # Fixture for creating an instance of ChatHistory |
| #8 | @pytest.fixture |
| #9 | def chat_memory_instance(): |
| #10 | return ChatHistory() |
| #11 | |
| #12 | |
| #13 | def test_add_chat_memory(chat_memory_instance): |
| #14 | app_id = "test_app" |
| #15 | session_id = "test_session" |
| #16 | human_message = "Hello, how are you?" |
| #17 | ai_message = "I'm fine, thank you!" |
| #18 | |
| #19 | chat_message = ChatMessage() |
| #20 | chat_message.add_user_message(human_message) |
| #21 | chat_message.add_ai_message(ai_message) |
| #22 | |
| #23 | chat_memory_instance.add(app_id, session_id, chat_message) |
| #24 | |
| #25 | assert chat_memory_instance.count(app_id, session_id) == 1 |
| #26 | chat_memory_instance.delete(app_id, session_id) |
| #27 | |
| #28 | |
| #29 | def test_get(chat_memory_instance): |
| #30 | app_id = "test_app" |
| #31 | session_id = "test_session" |
| #32 | |
| #33 | for i in range(1, 7): |
| #34 | human_message = f"Question {i}" |
| #35 | ai_message = f"Answer {i}" |
| #36 | |
| #37 | chat_message = ChatMessage() |
| #38 | chat_message.add_user_message(human_message) |
| #39 | chat_message.add_ai_message(ai_message) |
| #40 | |
| #41 | chat_memory_instance.add(app_id, session_id, chat_message) |
| #42 | |
| #43 | recent_memories = chat_memory_instance.get(app_id, session_id, num_rounds=5) |
| #44 | |
| #45 | assert len(recent_memories) == 5 |
| #46 | |
| #47 | all_memories = chat_memory_instance.get(app_id, fetch_all=True) |
| #48 | |
| #49 | assert len(all_memories) == 6 |
| #50 | |
| #51 | |
| #52 | def test_delete_chat_history(chat_memory_instance): |
| #53 | app_id = "test_app" |
| #54 | session_id = "test_session" |
| #55 | |
| #56 | for i in range(1, 6): |
| #57 | human_message = f"Question {i}" |
| #58 | ai_message = f"Answer {i}" |
| #59 | |
| #60 | chat_message = ChatMessage() |
| #61 | chat_message.add_user_message(human_message) |
| #62 | chat_message.add_ai_message(ai_message) |
| #63 | |
| #64 | chat_memory_instance.add(app_id, session_id, chat_message) |
| #65 | |
| #66 | session_id_2 = "test_session_2" |
| #67 | |
| #68 | for i in range(1, 6): |
| #69 | human_message = f"Question {i}" |
| #70 | ai_message = f"Answer {i}" |
| #71 | |
| #72 | chat_message = ChatMessage() |
| #73 | chat_message.add_user_message(human_message) |
| #74 | chat_message.add_ai_message(ai_message) |
| #75 | |
| #76 | chat_memory_instance.add(app_id, session_id_2, chat_message) |
| #77 | |
| #78 | chat_memory_instance.delete(app_id, session_id) |
| #79 | |
| #80 | assert chat_memory_instance.count(app_id, session_id) == 0 |
| #81 | assert chat_memory_instance.count(app_id) == 5 |
| #82 | |
| #83 | chat_memory_instance.delete(app_id) |
| #84 | |
| #85 | assert chat_memory_instance.count(app_id) == 0 |
| #86 | |
| #87 | |
| #88 | @pytest.fixture |
| #89 | def close_connection(chat_memory_instance): |
| #90 | yield |
| #91 | chat_memory_instance.close_connection() |
| #92 |