Unit tests for dependencies.py

CHANGELOG.md

@@ -11,11 +11,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 - Fixed a bug that prevented AsyncSqlite checkpoint to access the DB in streamed endpoints.
+- Fixed a bug that caused some unit tests to fail due to a change in how httpx_mock works in version 0.32
 
 ## [0.1.0] - 19.09.2024
 
 ### Added
 - Update readme
+- Extra unit tests for dependencies.py
 
 ### Removed
 - Github action to create the docs.

src/neuroagent/app/dependencies.py

@@ -101,7 +101,6 @@ def get_engine(
         if "sqlite" in settings.db.prefix:  # type: ignore
             # https://fastapi.tiangolo.com/tutorial/sql-databases/#create-the-sqlalchemy-engine
             engine_kwargs["connect_args"] = {"check_same_thread": False}
-
         engine = create_engine(**engine_kwargs)
     else:
         logger.warning("The SQL db_prefix needs to be set to use the SQL DB.")

tests/agents/test_simple_chat_agent.py

@@ -9,6 +9,7 @@
 from neuroagent.agents import AgentOutput, AgentStep, SimpleChatAgent
 
 
+@pytest.mark.httpx_mock(can_send_already_matched_responses=True)
 @pytest.mark.asyncio
 async def test_arun(fake_llm_with_tools, httpx_mock):
     llm, tools, fake_responses = await anext(fake_llm_with_tools)
@@ -64,6 +65,7 @@ async def test_arun(fake_llm_with_tools, httpx_mock):
         assert len(messages_list) == 10
 
 
+@pytest.mark.httpx_mock(can_send_already_matched_responses=True)
 @pytest.mark.asyncio
 async def test_astream(fake_llm_with_tools, httpx_mock):
     llm, tools, fake_responses = await anext(fake_llm_with_tools)

tests/app/test_dependencies.py

@@ -4,9 +4,10 @@
 import os
 from pathlib import Path
 from typing import AsyncIterator
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import pytest
+from fastapi import HTTPException
 from httpx import AsyncClient
 from langchain_openai import ChatOpenAI
 from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
@@ -19,13 +20,18 @@
     get_brain_region_resolver_tool,
     get_cell_types_kg_hierarchy,
     get_chat_agent,
+    get_connection_string,
     get_electrophys_feature_tool,
+    get_engine,
     get_httpx_client,
     get_kg_morpho_feature_tool,
+    get_kg_token,
     get_language_model,
     get_literature_tool,
     get_morpho_tool,
     get_morphology_feature_tool,
+    get_session,
+    get_settings,
     get_traces_tool,
     get_update_kg_hierarchy,
     get_user_id,
@@ -38,6 +44,18 @@
     LiteratureSearchTool,
     MorphologyFeatureTool,
 )
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+
+def test_get_settings(monkeypatch):
+    monkeypatch.setenv("NEUROAGENT_TOOLS__LITERATURE__URL", "https://localhost1")
+    monkeypatch.setenv("NEUROAGENT_KNOWLEDGE_GRAPH__BASE_URL", "https://localhost2")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__USERNAME", "fake_username")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__PASSWORD", "fake_password")
+    settings = get_settings()
+    assert settings.tools.literature.url == "https://localhost1"
+    assert settings.knowledge_graph.url == "https://localhost2/search/query/"
 
 
 @pytest.mark.asyncio
@@ -361,3 +379,113 @@ async def test_get_cell_types_kg_hierarchy(
     )
 
     assert os.path.exists(settings.knowledge_graph.ct_saving_path)
+
+
+def test_get_connection_string_full(monkeypatch):
+    monkeypatch.setenv("NEUROAGENT_TOOLS__LITERATURE__URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_KNOWLEDGE_GRAPH__BASE_URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_DB__PREFIX", "http://")
+    monkeypatch.setenv("NEUROAGENT_DB__USER", "John")
+    monkeypatch.setenv("NEUROAGENT_DB__PASSWORD", "Doe")
+    monkeypatch.setenv("NEUROAGENT_DB__HOST", "localhost")
+    monkeypatch.setenv("NEUROAGENT_DB__PORT", "5000")
+    monkeypatch.setenv("NEUROAGENT_DB__NAME", "test")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__USERNAME", "fake_username")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__PASSWORD", "fake_password")
+
+    settings = Settings()
+    result = get_connection_string(settings)
+    assert (
+        result == "http://John:Doe@localhost:5000/test"
+    ), "must return fully formed connection string"
+
+
+def test_get_connection_string_no_prefix(monkeypatch):
+    monkeypatch.setenv("NEUROAGENT_TOOLS__LITERATURE__URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_KNOWLEDGE_GRAPH__BASE_URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_DB__PREFIX", "")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__USERNAME", "fake_username")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__PASSWORD", "fake_password")
+
+    settings = Settings()
+
+    result = get_connection_string(settings)
+    assert result is None, "should return None when prefix is not set"
+
+
+@patch("neuroagent.app.dependencies.create_engine")
+def test_get_engine(create_engine_mock, monkeypatch):
+    create_engine_mock.return_value = Mock()
+
+    monkeypatch.setenv("NEUROAGENT_TOOLS__LITERATURE__URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_KNOWLEDGE_GRAPH__BASE_URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_DB__PREFIX", "prefix")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__USERNAME", "fake_username")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__PASSWORD", "fake_password")
+
+    settings = Settings()
+
+    connection_string = "https://localhost"
+    retval = get_engine(settings=settings, connection_string=connection_string)
+    assert retval is not None
+
+
+@patch("neuroagent.app.dependencies.create_engine")
+def test_get_engine_no_connection_string(create_engine_mock, monkeypatch):
+    create_engine_mock.return_value = Mock()
+
+    monkeypatch.setenv("NEUROAGENT_TOOLS__LITERATURE__URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_KNOWLEDGE_GRAPH__BASE_URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_DB__PREFIX", "prefix")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__USERNAME", "fake_username")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__PASSWORD", "fake_password")
+
+    settings = Settings()
+
+    retval = get_engine(settings=settings, connection_string=None)
+    assert retval is None
+
+
+@patch("sqlalchemy.orm.Session")
+def test_get_session_success(_):
+    database_url = "sqlite:///:memory:"
+    engine = create_engine(database_url)
+    result = next(get_session(engine))
+    assert isinstance(result, Session)
+
+
+def test_get_session_no_engine():
+    with pytest.raises(HTTPException):
+        next(get_session(None))
+
+
+def test_get_kg_token_with_token(monkeypatch):
+    monkeypatch.setenv("NEUROAGENT_TOOLS__LITERATURE__URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_KNOWLEDGE_GRAPH__BASE_URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_DB__PREFIX", "prefix")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__USERNAME", "fake_username")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__PASSWORD", "fake_password")
+
+    settings = Settings()
+
+    token = "Test_Token"
+    result = get_kg_token(settings, token)
+    assert result == "Test_Token"
+
+
+def test_get_kg_token_with_settings_knowledge_graph_token(monkeypatch):
+    monkeypatch.setenv("NEUROAGENT_TOOLS__LITERATURE__URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_KNOWLEDGE_GRAPH__BASE_URL", "http://localhost")
+    monkeypatch.setenv("NEUROAGENT_DB__PREFIX", "prefix")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__USERNAME", "fake_username")
+    monkeypatch.setenv("NEUROAGENT_KEYCLOAK__PASSWORD", "fake_password")
+    monkeypatch.setenv("NEUROAGENT_KNOWLEDGE_GRAPH__USE_TOKEN", "true")
+    monkeypatch.setenv("NEUROAGENT_KNOWLEDGE_GRAPH__TOKEN", "Test_kg_Token")
+
+    settings = Settings()
+
+    token = None
+
+    result = get_kg_token(settings, token)
+
+    assert result == "Test_kg_Token"

tests/test_utils.py

@@ -313,6 +313,7 @@ async def test_get_kg_data_errors(httpx_mock):
     )
 
 
+@pytest.mark.httpx_mock(can_send_already_matched_responses=True)
 @pytest.mark.asyncio
 async def test_get_kg_data(httpx_mock):
     url = "http://fake_url"

tests/tools/test_electrophys_tool.py

@@ -15,6 +15,7 @@
 
 
 class TestElectrophysTool:
+    @pytest.mark.httpx_mock(can_send_already_matched_responses=True)
     @pytest.mark.asyncio
     async def test_arun(self, httpx_mock):
         url = "http://fake_url"

tests/tools/test_traces_tool.py

@@ -11,6 +11,7 @@
 
 
 class TestTracesTool:
+    @pytest.mark.httpx_mock(can_send_already_matched_responses=True)
     @pytest.mark.asyncio
     async def test_arun(self, httpx_mock, brain_region_json_path):
         url = "http://fake_url"