Subagents delivered tests across all uncovered backend modules: Schemas (100%): agent, auth, health, profile, settings, validation Services (98-100%): auth, profile, health, llm, mapping, resource, security, git, superset_lookup, sql_table_extractor, rbac API routes (new): auth, admin, health, environments, plugins, dashboards (helpers, projection, actions, listing), git (config, deps, env, helpers) Clean Release (100%): DTO, facade, policy_engine, stages, repos, preparation, source_isolation, compliance Git services: base, remote_providers Agent module: app, run, middleware, langgraph_setup Core: trace, cleanup, ws_log_handler, timezone, auth (config/oauth/security), matching Reports: normalizer, report_service, type_profiles Notifications: service, providers Also: - .gitignore: add .coverage, *.cover, coverage-* dirs - src/schemas/auth.py: fix AD group DN regex (comma in CN=...) - Remove co-located src/services/__tests__/ (caused pytest module collision)
89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
# #region Test.AgentChat.Middleware [C:3] [TYPE Module] [SEMANTICS test,agent,middleware,audit]
|
|
# @BRIEF Tests for agent/middleware.py — log_tool_event.
|
|
# @RELATION BINDS_TO -> [AgentChat.Middleware]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
import pytest
|
|
|
|
|
|
# #region test_log_tool_event [C:2] [TYPE Function]
|
|
# @BRIEF Test log_tool_event for various event types.
|
|
class TestLogToolEvent:
|
|
@pytest.mark.asyncio
|
|
async def test_logs_tool_start(self):
|
|
from src.agent.middleware import log_tool_event
|
|
event = {
|
|
"event": "on_tool_start",
|
|
"name": "migrate",
|
|
"data": {"input": {"dashboard_id": "42"}},
|
|
}
|
|
with patch("src.agent.middleware.get_user_jwt", return_value="user-token"):
|
|
await log_tool_event(event, "conv-1")
|
|
# No exception = success
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logs_tool_end(self):
|
|
from src.agent.middleware import log_tool_event
|
|
event = {
|
|
"event": "on_tool_end",
|
|
"name": "migrate",
|
|
"data": {"output": "success"},
|
|
}
|
|
with patch("src.agent.middleware.get_user_jwt", return_value="user-token"):
|
|
await log_tool_event(event, "conv-1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logs_tool_error(self):
|
|
from src.agent.middleware import log_tool_event
|
|
event = {
|
|
"event": "on_tool_error",
|
|
"name": "migrate",
|
|
"data": {"error": "Connection failed"},
|
|
}
|
|
with patch("src.agent.middleware.get_user_jwt", return_value="user-token"):
|
|
await log_tool_event(event, "conv-1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logs_without_user_jwt(self):
|
|
from src.agent.middleware import log_tool_event
|
|
event = {
|
|
"event": "on_tool_start",
|
|
"name": "test_tool",
|
|
"data": {"input": {}},
|
|
}
|
|
with patch("src.agent.middleware.get_user_jwt", return_value=None):
|
|
await log_tool_event(event, "conv-1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handles_missing_data_key(self):
|
|
from src.agent.middleware import log_tool_event
|
|
event = {"event": "on_tool_start", "name": "test_tool"}
|
|
with patch("src.agent.middleware.get_user_jwt", return_value=None):
|
|
await log_tool_event(event, "conv-1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handles_unknown_event_kind(self):
|
|
from src.agent.middleware import log_tool_event
|
|
event = {"event": "on_custom_event", "name": "custom"}
|
|
with patch("src.agent.middleware.get_user_jwt", return_value=None):
|
|
await log_tool_event(event, "conv-1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_truncates_long_input(self):
|
|
from src.agent.middleware import log_tool_event
|
|
long_input = "x" * 1000
|
|
event = {
|
|
"event": "on_tool_start",
|
|
"name": "big_tool",
|
|
"data": {"input": long_input},
|
|
}
|
|
with patch("src.agent.middleware.get_user_jwt", return_value="token"):
|
|
await log_tool_event(event, "conv-1")
|
|
# #endregion test_log_tool_event
|
|
# #endregion Test.AgentChat.Middleware
|