129e4ea1fc
New router intelligence modules (26 files): alert_ingest/store, audit_store, architecture_pressure, backlog_generator/store, cost_analyzer, data_governance, dependency_scanner, drift_analyzer, incident_* (5 files), llm_enrichment, platform_priority_digest, provider_budget, release_check_runner, risk_* (6 files), signature_state_store, sofiia_auto_router, tool_governance New services: - sofiia-console: Dockerfile, adapters/, monitor/nodes/ops/voice modules, launchd, react static - memory-service: integration_endpoints, integrations, voice_endpoints, static UI - aurora-service: full app suite (analysis, job_store, orchestrator, reporting, schemas, subagents) - sofiia-supervisor: new supervisor service - aistalk-bridge-lite: Telegram bridge lite - calendar-service: CalDAV calendar service with reminders - mlx-stt-service / mlx-tts-service: Apple Silicon speech services - binance-bot-monitor: market monitor service - node-worker: STT/TTS memory providers New tools (9): agent_email, browser_tool, contract_tool, observability_tool, oncall_tool, pr_reviewer_tool, repo_tool, safe_code_executor, secure_vault New crews: agromatrix_crew (10 modules: depth_classifier, doc_facts, doc_focus, farm_state, light_reply, llm_factory, memory_manager, proactivity, reflection_engine, session_context, style_adapter, telemetry) Tests: 85+ test files for all new modules Made-with: Cursor
264 lines
7.3 KiB
Python
264 lines
7.3 KiB
Python
"""
|
|
Tests for Knowledge Base Tool
|
|
"""
|
|
|
|
import pytest
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from services.router.tool_manager import ToolManager, ToolResult
|
|
|
|
|
|
class TestKnowledgeBaseTool:
|
|
"""Test KB tool functionality"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_requires_query(self):
|
|
"""Test that search requires query"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "search",
|
|
"params": {}
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "query" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_returns_results(self):
|
|
"""Test search returns results structure"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "search",
|
|
"params": {
|
|
"query": "test query",
|
|
"limit": 10
|
|
}
|
|
})
|
|
|
|
assert result.success is True
|
|
assert "results" in result.result
|
|
assert "summary" in result.result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_with_paths_filter(self):
|
|
"""Test search with paths filter"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "search",
|
|
"params": {
|
|
"query": "test",
|
|
"paths": ["docs"],
|
|
"limit": 5
|
|
}
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.result["count"] >= 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snippets_requires_query(self):
|
|
"""Test snippets requires query"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "snippets",
|
|
"params": {}
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "query" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snippets_returns_structure(self):
|
|
"""Test snippets returns proper structure"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "snippets",
|
|
"params": {
|
|
"query": "test",
|
|
"limit": 5
|
|
}
|
|
})
|
|
|
|
assert result.success is True
|
|
assert "results" in result.result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_open_requires_path(self):
|
|
"""Test open requires path"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "open",
|
|
"params": {}
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "path" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_open_blocks_traversal(self):
|
|
"""Test open blocks path traversal"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "open",
|
|
"params": {
|
|
"path": "../../../etc/passwd"
|
|
}
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "traversal" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_open_blocks_not_allowed_path(self):
|
|
"""Test open blocks not allowed paths"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "open",
|
|
"params": {
|
|
"path": "services/router/tool_manager.py"
|
|
}
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "not in allowed" in result.error.lower() or "not found" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sources_returns_indexed(self):
|
|
"""Test sources returns indexed sources"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "sources",
|
|
"params": {}
|
|
})
|
|
|
|
assert result.success is True
|
|
assert "sources" in result.result
|
|
assert "allowed_paths" in result.result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sources_with_paths_filter(self):
|
|
"""Test sources with paths filter"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "sources",
|
|
"params": {
|
|
"paths": ["docs"]
|
|
}
|
|
})
|
|
|
|
assert result.success is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_action_returns_error(self):
|
|
"""Test unknown action returns error"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "unknown_action",
|
|
"params": {}
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "unknown action" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_with_file_glob(self):
|
|
"""Test search with file glob filter"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "search",
|
|
"params": {
|
|
"query": "test",
|
|
"file_glob": "**/*.md",
|
|
"limit": 10
|
|
}
|
|
})
|
|
|
|
assert result.success is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snippets_context_lines(self):
|
|
"""Test snippets with context lines"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "snippets",
|
|
"params": {
|
|
"query": "test",
|
|
"context_lines": 2,
|
|
"limit": 3
|
|
}
|
|
})
|
|
|
|
assert result.success is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_open_with_line_range(self):
|
|
"""Test open with line range"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "open",
|
|
"params": {
|
|
"path": "docs/README.md",
|
|
"start_line": 1,
|
|
"end_line": 10
|
|
}
|
|
})
|
|
|
|
if result.success:
|
|
assert "content" in result.result
|
|
assert "start_line" in result.result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_result_structure(self):
|
|
"""Test search result has proper structure"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "search",
|
|
"params": {
|
|
"query": "documentation",
|
|
"limit": 5
|
|
}
|
|
})
|
|
|
|
assert result.success is True
|
|
if result.result["count"] > 0:
|
|
r = result.result["results"][0]
|
|
assert "path" in r
|
|
assert "score" in r
|
|
assert "highlights" in r
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redaction_of_secrets(self):
|
|
"""Test that secrets are redacted"""
|
|
tool_mgr = ToolManager({})
|
|
|
|
result = await tool_mgr._kb_tool({
|
|
"action": "snippets",
|
|
"params": {
|
|
"query": "API_KEY",
|
|
"limit": 1
|
|
}
|
|
})
|
|
|
|
if result.success and result.result["count"] > 0:
|
|
for snippet in result.result["results"]:
|
|
text = snippet.get("text", "")
|
|
assert "sk-***" in text or "API_KEY" not in text or "***" in text
|