Merge pull request #16 from jethronap/8_testing

8 testing

Author: jethronap

local_test_pipeline.sh

@@ -0,0 +1,31 @@
+#!/bin/bash
+PYTHONPATH="${PYTHONPATH}:$(realpath "./src")"
+export PYTHONPATH
+
+# Navigate to the script's directory (project root)
+cd "$(dirname "$0")" || exit
+
+echo "Running base agent tests"
+pytest tests/test_base_agent.py
+echo "Done..."
+echo "==============================================="
+
+echo "Running dummy agent tests"
+pytest tests/test_dummy_agent.py
+echo "Done..."
+echo "==============================================="
+
+echo "Running llm wrapper tests"
+pytest tests/test_llm_wrapper.py
+echo "Done..."
+echo "==============================================="
+
+echo "Running memory store tests"
+pytest tests/test_memory_store.py
+echo "Done..."
+echo "==============================================="
+
+echo "Running ollama client tests"
+pytest tests/test_ollama_client.py
+echo "Done..."
+echo "==============================================="

pyproject.toml

@@ -8,6 +8,7 @@ dependencies = [
     "loguru>=0.7.3",
     "pre-commit>=4.2.0",
     "pydantic-settings>=2.9.1",
+    "pytest>=8.3.5",
     "requests>=2.32.3",
     "ruff>=0.11.7",
     "sqlmodel>=0.0.24",

src/agents/dummy_agent.py

@@ -47,4 +47,5 @@ def act(self, action: str, context: Dict[str, Any]) -> Any:
         logger.info(f"{self.name} executing action: {action} with context: {context}")
         result = f"result_of_{action}"
         logger.info(f"{self.name} executed action: {action} -> {result}")
+        print(f"{self.name} executed action: {action} -> {result}")
         return result

src/config/settings.py

@@ -10,12 +10,18 @@ class OllamaSettings(BaseSettings):
     Base settings for Ollama interaction
     """
 
-    base_url: str = Field(description="The Ollama base url")
-    timeout_seconds: float
+    base_url: str = Field(
+        default="http://localhost:11434/api/generate", description="The Ollama base url"
+    )
+    timeout_seconds: float = Field(
+        default=30.0, description="Timeout for calling Ollama"
+    )
     stream: bool = Field(default=False, description="Flag to denote chunked streaming.")
-    model: str = Field(description="Ollama model name.")
+    model: str = Field(default="gemma2", description="Ollama model name.")
 
-    model_config = SettingsConfigDict(env_file=".env", env_prefix="OLLAMA_")
+    model_config = SettingsConfigDict(
+        env_file=".env", env_prefix="OLLAMA_", extra="allow"
+    )
 
 
 class DatabaseSettings(BaseSettings):
@@ -36,16 +42,18 @@ class LoggingSettings(BaseSettings):
     Configuration for Loguru logging sinks.
     """
 
-    level: str = Field(description="The log level")
+    level: str = Field(default="DEBUG", description="The log level")
     console: bool = Field(default=True, description="Show logs in console")
     enable_file: bool = Field(
         default=False, description="Flag to denote persistence of logs"
     )
     filepath: Optional[Path] = Field(
         default=None, description="Optional file path for logs"
     )
-    rotation: str = Field(description="Roll log after this size")
-    retention: str = Field(description="Keep logs for this amount of time")
-    compression: str = Field(description="Compress old logs")
+    rotation: str = Field(default="10 MB", description="Roll log after this size")
+    retention: str = Field(
+        default="7 days", description="Keep logs for this amount of time"
+    )
+    compression: str = Field(default="zip", description="Compress old logs")
 
     model_config = SettingsConfigDict(env_file=".env", env_prefix="LOG_", extra="allow")

src/tools/llm_wrapper.py

@@ -42,18 +42,26 @@ def generate_response(self, prompt: str, **kwargs: Any) -> str:
 
         # Extract generated text from response
         text = ""
-        if "response" in response:
-            text = response.get("response", "").strip()
-            logger.debug("Extracted 'response' field from Ollama response")
-        elif "choices" in response:
-            text = response.get("choices", [{}][0].get("text", "").strip())
-            logger.debug("Extracted 'choices' field from Ollama response")
-        elif "results" in response:
-            text = response.get("results", [{}])[0].get("text", "").strip()
-            logger.debug("Extracted 'results' field from Ollama response")
+        if isinstance(response, dict):
+            if "response" in response and isinstance(response["response"], str):
+                text = response["response"].strip()
+                logger.debug("Extracted 'response' field from Ollama response")
+            elif "choices" in response and isinstance(response["choices"], list):
+                choice = response["choices"][0]
+                text = (
+                    choice.get("text", "").strip() if isinstance(choice, dict) else ""
+                )
+                logger.debug("Extracted 'choices' field from Ollama response")
+            elif "results" in response and isinstance(response["results"], list):
+                result = response["results"][0]
+                text = (
+                    result.get("text", "").strip() if isinstance(result, dict) else ""
+                )
+                logger.debug("Extracted 'results' field from Ollama response")
+            else:
+                logger.warning(f"Unexpected response format: {response}")
         else:
-            logger.warning(f"Unexpected response format: {response}")
-            text = str(response).strip()
+            logger.warning(f"Invalid response type: {type(response)}")
 
         if not text:
             logger.warning("Received empty response from LLM")

tests/conftest.py

@@ -0,0 +1,33 @@
+import pytest
+from src.config.settings import OllamaSettings, DatabaseSettings
+
+
+@pytest.fixture
+def ollama_settings():
+    return OllamaSettings(
+        base_url="http://test-server/api/generate",
+        timeout_seconds=0.1,
+        model="test-model",
+    )
+
+
+@pytest.fixture
+def db_settings(tmp_path):
+    return DatabaseSettings(
+        url=f"sqlite:///{tmp_path / 'test_memory.db'}",
+        echo=False,
+    )
+
+
+@pytest.fixture(autouse=True)
+def patch_client(monkeypatch):
+    # Replace OllamaClient used inside LocalLLM with our stub
+    monkeypatch.setattr("src.tools.llm_wrapper.OllamaClient", StubClient)
+
+
+class StubClient:
+    def __init__(self, settings):
+        pass
+
+    def __call__(self, prompt, **kwargs):
+        return self._response

tests/test_base_agent.py

@@ -0,0 +1,24 @@
+import pytest
+from src.agents.base_agent import Agent
+
+
+def test_base_agent_abstract():
+    with pytest.raises(TypeError):
+        Agent(name="X", tools={}, memory=None)
+
+
+class SimpleAgent(Agent):
+    def perceive(self, input_data):
+        return {"in": input_data}
+
+    def plan(self, observation):
+        return ["step"]
+
+    def act(self, action, context):
+        return "ok"
+
+
+def test_simple_agent_runs():
+    agent = SimpleAgent(name="Simple", tools={}, memory=None)
+    # Should run without errors
+    agent.achieve_goal("DATA")

tests/test_dummy_agent.py

@@ -0,0 +1,11 @@
+from src.agents.dummy_agent import DummyAgent
+
+
+def test_dummy_agent_cycle(capsys):
+    agent = DummyAgent(name="TestDummy")
+    agent.achieve_goal("INPUT")
+    captured = capsys.readouterr().out.strip().splitlines()
+    # Expect three lines for dummy_step1..3
+    assert len(captured) == 3
+    for i, line in enumerate(captured, start=1):
+        assert f"executed action: dummy_step{i}" in line

tests/test_llm_wrapper.py

@@ -0,0 +1,31 @@
+from src.config.settings import OllamaSettings
+from src.tools.llm_wrapper import LocalLLM
+from tests.conftest import StubClient
+
+
+def make_wrapper(resp):
+    settings = OllamaSettings(base_url="http://x", timeout_seconds=1, model="m")
+    stub = StubClient(settings)
+    stub._response = resp
+    # ensure LocalLLM uses our stub instance
+    wrapper = LocalLLM(settings)
+    wrapper.client = stub
+    return wrapper
+
+
+def test_generate_from_response_field():
+    wrapper = make_wrapper({"response": "hi there"})
+    assert wrapper.generate_response("x") == "hi there"
+
+
+def test_generate_from_choices_field():
+    wrapper = make_wrapper({"choices": [{"text": "foo "}]})
+    assert wrapper.generate_response("x") == "foo"
+
+
+def test_generate_empty_and_warn(caplog):
+    wrapper = make_wrapper({})
+    caplog.set_level("WARNING")
+    out = wrapper.generate_response("x")
+    assert out == ""
+    assert "" in caplog.text.lower()

tests/test_memory_store.py

@@ -0,0 +1,29 @@
+from src.memory.store import MemoryStore
+from src.memory.models import Memory
+
+
+def test_store_and_load_roundtrip(db_settings):
+    store = MemoryStore(db_settings)
+    # store 2 memories for AgentA
+    m1 = store.store("AgentA", "step1", "data1")
+    m2 = store.store("AgentA", "step2", "data2")
+    assert isinstance(m1, Memory) and isinstance(m2, Memory)
+
+    records = store.load("AgentA")
+    assert len(records) == 2
+    # ensure content matches
+    contents = {r.content for r in records}
+    assert contents == {"data1", "data2"}
+
+
+def test_isolated_databases(tmp_path):
+    # two separate stores shouldn't see each other's data
+    from src.config.settings import DatabaseSettings
+
+    s1 = DatabaseSettings(url=f"sqlite:///{tmp_path / 'a.db'}")
+    s2 = DatabaseSettings(url=f"sqlite:///{tmp_path / 'b.db'}")
+    st1 = MemoryStore(s1)
+    st2 = MemoryStore(s2)
+    st1.store("X", "s", "d")
+    assert len(st1.load("X")) == 1
+    assert st2.load("X") == []