From b153daa989ea696846bec86a363e68eb1d324b48 Mon Sep 17 00:00:00 2001 From: Dystroyer8 Date: Thu, 30 Jul 2026 17:00:45 +0200 Subject: [PATCH] test: cover chat core and session storage --- tests/unit/test_chat_service.py | 57 +++++++++++++++++++++++++++++ tests/unit/test_cli.py | 40 +++++++++++++++++++++ tests/unit/test_ollama_provider.py | 56 +++++++++++++++++++++++++++++ tests/unit/test_settings.py | 39 ++++++++++++++++++++ tests/unit/test_sqlite_store.py | 58 ++++++++++++++++++++++++++++++ 5 files changed, 250 insertions(+) create mode 100644 tests/unit/test_chat_service.py create mode 100644 tests/unit/test_cli.py create mode 100644 tests/unit/test_ollama_provider.py create mode 100644 tests/unit/test_settings.py create mode 100644 tests/unit/test_sqlite_store.py diff --git a/tests/unit/test_chat_service.py b/tests/unit/test_chat_service.py new file mode 100644 index 0000000..015ed1a --- /dev/null +++ b/tests/unit/test_chat_service.py @@ -0,0 +1,57 @@ +import tempfile +import unittest +from pathlib import Path + +from javis.core.chat_service import ChatService +from javis.memory.sqlite_store import SQLiteSessionStore +from javis.providers.base import ChatMessage, LocalModelProvider + + +class RecordingProvider: + name = "ollama" + model = "test-model" + + def __init__(self) -> None: + self.calls: list[list[ChatMessage]] = [] + + def chat(self, messages: list[ChatMessage]) -> str: + self.calls.append(messages) + return f"Antwort {len(self.calls)}" + + +class ChatServiceTests(unittest.TestCase): + def setUp(self) -> None: + self.temporary_directory = tempfile.TemporaryDirectory() + database = Path(self.temporary_directory.name) / "sessions.sqlite3" + self.provider = RecordingProvider() + self.service = ChatService(SQLiteSessionStore(database), self.provider) + + def tearDown(self) -> None: + self.temporary_directory.cleanup() + + def test_test_double_implements_provider_contract(self) -> None: + self.assertIsInstance(self.provider, LocalModelProvider) + + def test_send_passes_and_persists_conversation_history(self) -> None: + session = self.service.new_session() + + first = self.service.send(session.id, "Hallo") + second = self.service.send(session.id, "Und jetzt?") + + self.assertEqual(first, "Antwort 1") + self.assertEqual(second, "Antwort 2") + self.assertEqual(len(self.provider.calls[1]), 3) + loaded = self.service.load_session(session.id) + self.assertEqual( + [message.content for message in loaded.messages], + [ + "Hallo", + "Antwort 1", + "Und jetzt?", + "Antwort 2", + ], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py new file mode 100644 index 0000000..dc65f97 --- /dev/null +++ b/tests/unit/test_cli.py @@ -0,0 +1,40 @@ +import tempfile +import unittest +from pathlib import Path + +from javis.core.chat_service import ChatService +from javis.interface.cli import run_chat +from javis.memory.sqlite_store import SQLiteSessionStore +from javis.providers.base import ChatMessage + + +class _FakeProvider: + name = "ollama" + model = "test-model" + + def chat(self, messages: list[ChatMessage]) -> str: + return f"Echo: {messages[-1].content}" + + +class CliTests(unittest.TestCase): + def test_basic_chat_commands(self) -> None: + with tempfile.TemporaryDirectory() as directory: + store = SQLiteSessionStore(Path(directory) / "sessions.sqlite3") + service = ChatService(store, _FakeProvider()) + inputs = iter(["Hallo", "/sessions", "/new", "/clear", "/exit"]) + output: list[str] = [] + + result = run_chat( + service, + input_fn=lambda _prompt: next(inputs), + output=output.append, + ) + + self.assertEqual(result, 0) + self.assertTrue(any(line == "Javis: Echo: Hallo" for line in output)) + self.assertTrue(any("Nachrichten" in line for line in output)) + self.assertEqual(output[-1], "Chat beendet.") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_ollama_provider.py b/tests/unit/test_ollama_provider.py new file mode 100644 index 0000000..71c6170 --- /dev/null +++ b/tests/unit/test_ollama_provider.py @@ -0,0 +1,56 @@ +import json +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import ClassVar + +from javis.providers.base import ChatMessage +from javis.providers.ollama import OllamaProvider + + +class _OllamaHandler(BaseHTTPRequestHandler): + request_payload: ClassVar[dict[str, object]] = {} + + def do_POST(self) -> None: + length = int(self.headers["Content-Length"]) + type(self).request_payload = json.loads(self.rfile.read(length)) + body = json.dumps( + { + "message": {"role": "assistant", "content": "Lokale Antwort"}, + "done": True, + } + ).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args: object) -> None: + return + + +class OllamaProviderTests(unittest.TestCase): + def test_provider_uses_local_chat_endpoint(self) -> None: + server = ThreadingHTTPServer(("127.0.0.1", 0), _OllamaHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + provider = OllamaProvider( + "test-model", + f"http://127.0.0.1:{server.server_port}", + 2, + ) + response = provider.chat([ChatMessage("user", "Hallo")]) + finally: + server.shutdown() + server.server_close() + thread.join() + + self.assertEqual(response, "Lokale Antwort") + self.assertEqual(_OllamaHandler.request_payload["model"], "test-model") + self.assertFalse(_OllamaHandler.request_payload["stream"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_settings.py b/tests/unit/test_settings.py new file mode 100644 index 0000000..f98a4e1 --- /dev/null +++ b/tests/unit/test_settings.py @@ -0,0 +1,39 @@ +import unittest +from pathlib import Path + +from javis.config.settings import ConfigurationError, Settings, default_data_dir + + +class SettingsTests(unittest.TestCase): + def test_windows_default_uses_local_app_data(self) -> None: + result = default_data_dir( + {"LOCALAPPDATA": "C:\\Local"}, + platform_name="nt", + home=Path("C:\\Users\\test"), + ) + self.assertEqual(result, Path("C:\\Local") / "Javis") + + def test_posix_default_uses_xdg_data_home(self) -> None: + result = default_data_dir( + {"XDG_DATA_HOME": "/tmp/data"}, + platform_name="posix", + home=Path("/home/test"), + ) + self.assertEqual(result, Path("/tmp/data/javis")) + + def test_explicit_data_dir_must_be_absolute(self) -> None: + with self.assertRaises(ConfigurationError): + Settings.from_env({"JAVIS_DATA_DIR": "relative"}) + + def test_provider_url_must_remain_local(self) -> None: + with self.assertRaises(ConfigurationError): + Settings.from_env( + { + "JAVIS_DATA_DIR": str(Path.cwd()), + "JAVIS_OLLAMA_URL": "https://example.com", + } + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_sqlite_store.py b/tests/unit/test_sqlite_store.py new file mode 100644 index 0000000..5f312ce --- /dev/null +++ b/tests/unit/test_sqlite_store.py @@ -0,0 +1,58 @@ +import tempfile +import unittest +from pathlib import Path + +from javis.memory.sqlite_store import SessionNotFoundError, SQLiteSessionStore + + +class SQLiteSessionStoreTests(unittest.TestCase): + def setUp(self) -> None: + self.temporary_directory = tempfile.TemporaryDirectory() + database = Path(self.temporary_directory.name) / "sessions.sqlite3" + self.store = SQLiteSessionStore(database) + + def tearDown(self) -> None: + self.temporary_directory.cleanup() + + def test_create_list_and_load_session(self) -> None: + created = self.store.create_session("ollama", "test-model") + loaded = self.store.get_session(created.id) + + self.assertEqual(loaded.id, created.id) + self.assertEqual(loaded.provider, "ollama") + self.assertEqual(loaded.model, "test-model") + self.assertEqual(self.store.list_sessions()[0].id, created.id) + + def test_messages_keep_exchange_order(self) -> None: + session = self.store.create_session("ollama", "test-model") + self.store.append_exchange(session.id, "eins", "zwei") + self.store.append_exchange(session.id, "drei", "vier") + + messages = self.store.get_messages(session.id) + + self.assertEqual( + [(message.role, message.content) for message in messages], + [ + ("user", "eins"), + ("assistant", "zwei"), + ("user", "drei"), + ("assistant", "vier"), + ], + ) + + def test_unknown_session_raises_clear_error(self) -> None: + with self.assertRaises(SessionNotFoundError): + self.store.get_session("missing") + + def test_clear_keeps_session_and_removes_messages(self) -> None: + session = self.store.create_session("ollama", "test-model") + self.store.append_exchange(session.id, "eins", "zwei") + + self.store.clear_messages(session.id) + + self.assertEqual(self.store.get_messages(session.id), []) + self.assertEqual(self.store.get_session(session.id).message_count, 0) + + +if __name__ == "__main__": + unittest.main()