Add: test for services

tests/unit/test_services.py

@@ -0,0 +1,134 @@
+import uuid
+from unittest.mock import MagicMock
+
+import pandas as pd
+import pytest
+
+# Import the service module and DB connector
+import app.services.gemini_service as gemini_module
+from app.services.gemini_service import generate_gemini_response, refine
+
+
+# Dummy implementations
+class DummyDB:
+    def __init__(self):
+        self.updated = []
+        self.deleted = []
+        self.stored = []
+
+    def update_dataframe(self, iter_rows):
+        for idx, row in iter_rows:
+            # Convert row (Series) to list for comparison
+            self.updated.append((idx, row.tolist()))
+
+    def delete_chat(self, chat_id):
+        self.deleted.append(chat_id)
+
+    def store_chat(self, question, answer, uid, user_id):
+        self.stored.append((question, answer, uid, user_id))
+
+
+class DummyClientModels:
+    def __init__(self, text, candidates=True):
+        self._text = text
+        self._candidates = [1] if candidates else []
+
+    def generate_content(self, model, contents):
+        class Response:
+            def __init__(self, text, candidates):
+                self.text = text
+                self.candidates = candidates
+
+        return Response(self._text, self._candidates)
+
+
+class DummyGeminiClient:
+    def __init__(self, text, candidates=True):
+        self.models = DummyClientModels(text, candidates)
+
+
+class DummyChatItem:
+    def __init__(self, id, question, answer):
+        self.id = id
+        self.question = question
+        self.answer = answer
+
+
+@pytest.fixture(autouse=True)
+def patch_db_and_client(monkeypatch):
+    # Patch DB in service to DummyDB
+    monkeypatch.setattr(gemini_module, "DB", DummyDB)
+    # Patch gemini_client in service
+    dummy_client = MagicMock()
+    monkeypatch.setattr(gemini_module, "gemini_client", dummy_client)
+    yield
+
+
+def test_refine_with_no_chats():
+    # Should not error when chats list is empty
+    refine([])
+
+
+def test_refine_with_chats(monkeypatch):
+    # Prepare chat items
+    chats = [DummyChatItem(1, "Q1", "A1"), DummyChatItem(2, "Q2", "A2")]
+    # Prepare a refined DataFrame with extra column
+    df = pd.DataFrame([["Q1", "A1", "X"], ["Q2", "A2", "Y"]])
+    # Patch the refine method of gemini_client
+    gemini_module.gemini_client.refine.return_value = df
+    # Use DummyDB instance
+    db = DummyDB()
+    monkeypatch.setattr(gemini_module, "DB", lambda: db)
+    # Call refine
+    refine(chats)
+    # Check update_dataframe called with df rows
+    assert db.updated == [(0, ["Q1", "A1", "X"]), (1, ["Q2", "A2", "Y"])]
+    # Check delete_chat for each id
+    assert db.deleted == [1, 2]
+
+
+def test_generate_gemini_response_success(monkeypatch):
+    prompt = "Hello"
+    uid = str(uuid.uuid4())
+    metadata = {"k": "v"}
+    user_id = uuid.uuid4()
+    # Set up DummyGeminiClient with candidates
+    dummy_client = DummyGeminiClient("resp", candidates=True)
+    monkeypatch.setattr(
+        gemini_module.gemini_client, "get_gemini_client", lambda: dummy_client
+    )
+    # Patch find_best_passage
+    gemini_module.gemini_client.find_best_passage.return_value = "best"
+    # Patch generate_prompt to return final and storable prompts
+    monkeypatch.setattr(
+        gemini_module, "generate_prompt", lambda p, pas, md: ("FP:" + p, "SP:" + p)
+    )
+    # Patch DB
+    db = DummyDB()
+    monkeypatch.setattr(gemini_module, "DB", lambda: db)
+
+    res = generate_gemini_response(prompt, uid, metadata, user_id)
+    assert res == "resp"
+    # Verify stored chat
+    assert db.stored == [("SP:" + prompt, "resp", uid, user_id)]
+
+
+def test_generate_gemini_response_no_candidates(monkeypatch):
+    prompt = "Hi"
+    uid = str(uuid.uuid4())
+    metadata = None
+    user_id = uuid.uuid4()
+    # Dummy client without candidates
+    dummy_client = DummyGeminiClient("txt", candidates=False)
+    monkeypatch.setattr(
+        gemini_module.gemini_client, "get_gemini_client", lambda: dummy_client
+    )
+    gemini_module.gemini_client.find_best_passage.return_value = ""
+    monkeypatch.setattr(gemini_module, "generate_prompt", lambda p, pas, md: ("F", "S"))
+    db = DummyDB()
+    monkeypatch.setattr(gemini_module, "DB", lambda: db)
+
+    res = generate_gemini_response(prompt, uid, metadata, user_id)
+    assert res == "txt"
+    # Should not store when no candidates present
+    assert db.stored == []