667 lines
37 KiB
Python
667 lines
37 KiB
Python
# #region Test.IntentKeyword.Edges [C:3] [TYPE Module] [SEMANTICS test,intent,keyword,edges,regression]
|
||
# @BRIEF Orthogonal edge-case tests for get_tools_for_query, infer_tool_from_text,
|
||
# and fast_confirmation_tool — covers substring collisions, order sensitivity,
|
||
# language edges, multi-intent, empty/null, and the P0 regression.
|
||
# @RELATION BINDS_TO -> [AgentChat.Tools.GetForQuery]
|
||
# @RELATION BINDS_TO -> [AgentChat.ToolResolver.InferFromText]
|
||
# @RELATION BINDS_TO -> [AgentChat.ToolResolver.FastConfirm]
|
||
# @TEST_EDGE: tool_substring -> "tool" must not match "tools" in system text (P0 regression)
|
||
# @TEST_EDGE: env_substring -> "env" must not match "environment" in system text (P1 regression)
|
||
# @TEST_EDGE: empty_input -> empty/None/special inputs must not crash
|
||
# @TEST_EDGE: elif_order -> infer_tool_from_text ordering produces correct single-intent result
|
||
# @TEST_EDGE: multi_intent -> get_tools_for_query accumulates all matching intents
|
||
# @TEST_EDGE: ru_stem -> Russian stemming (обслуж⊂обслуживание) works correctly
|
||
# @TEST_EDGE: mixed_lang -> mixed RU/EN queries select correct tools
|
||
from pathlib import Path
|
||
import sys
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
|
||
|
||
import pytest
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# Helpers — call the target functions with logger patched
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
|
||
def _get_tools_for_query(query: str, prefetch_available: bool = False) -> set[str]:
|
||
"""Return set of tool names selected by get_tools_for_query."""
|
||
with patch("src.agent.tools.logger", MagicMock()):
|
||
from src.agent.tools import get_tools_for_query
|
||
return {t.name for t in get_tools_for_query(query, prefetch_available=prefetch_available)}
|
||
|
||
|
||
def _infer_tool(text: str) -> str | None:
|
||
"""Return tool name inferred by infer_tool_from_text."""
|
||
with patch("src.agent.tools.logger", MagicMock()):
|
||
from src.agent._tool_resolver import infer_tool_from_text
|
||
return infer_tool_from_text(text)
|
||
|
||
|
||
def _fast_confirm(text: str) -> str | None:
|
||
"""Return tool name from fast_confirmation_tool."""
|
||
with patch("src.agent.tools.logger", MagicMock()):
|
||
from src.agent._tool_resolver import fast_confirmation_tool
|
||
return fast_confirmation_tool(text)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# A — Substring Collisions: PREFETCH MARKER contamination
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
PREFETCH_MARKER = "[PRE-FETCHED DATA — use this directly, do NOT call tools]\n"
|
||
PREFETCH_HEADER = "Available dashboards in environment 'ss-dev' (260 total):\n"
|
||
PREFETCH_BODY = """- USA (id: 1, modified: 2026-02-23)
|
||
- System Health (id: 2, modified: 2026-01-15)
|
||
- SQL Explorer (id: 3, modified: 2026-03-01)
|
||
- Task Manager (id: 4, modified: 2025-12-10)
|
||
- LLM Analytics (id: 5, modified: 2026-01-20)
|
||
- Backup Report (id: 6, modified: 2026-02-28)
|
||
- Security Audit (id: 7, modified: 2026-03-05)
|
||
- Branch Analytics (id: 8, modified: 2026-01-01)
|
||
- API Docs (id: 9, modified: 2026-02-14)
|
||
- Schema Viewer (id: 10, modified: 2026-03-10)
|
||
- Data Migration (id: 11, modified: 2026-02-20)
|
||
- Deploy Status (id: 12, modified: 2026-03-12)
|
||
[/PRE-FETCHED DATA]"""
|
||
|
||
FULL_PREFETCH = PREFETCH_MARKER + PREFETCH_HEADER + PREFETCH_BODY
|
||
|
||
|
||
class TestPrefetchMarkerContamination:
|
||
"""Tests that prefetch marker 'do NOT call tools' does not trigger show_capabilities."""
|
||
|
||
def test_P0_tool_in_marker_should_not_block_other_tools(self):
|
||
"""P0 regression: 'tool' ⊂ 'tools' in prefetch marker blocks ALL tools.
|
||
With the app.py fix, get_tools_for_query receives CLEAN user text.
|
||
This test DOCUMENTS that the RAW function is vulnerable to the substring.
|
||
"""
|
||
# Simulate what the function would see if called with contaminated text
|
||
contaminated = "Запусти обслуживание на дашборде USA\n" + FULL_PREFETCH
|
||
tools = _get_tools_for_query(contaminated, prefetch_available=True)
|
||
# WITHOUT fix: {"show_capabilities"} (early return because "tool" ⊂ "tools")
|
||
# The fact that it returns {"show_capabilities"} is the BUG — we document it here.
|
||
# The real fix is in app.py: caller passes clean text, not this contaminated text.
|
||
assert "show_capabilities" in tools
|
||
|
||
def test_P1_env_in_header_should_not_add_list_environments(self):
|
||
"""P1: 'env' ⊂ 'environment' in prefetch header adds list_environments."""
|
||
contaminated = "Запусти обслуживание на дашборде USA\n" + FULL_PREFETCH
|
||
tools = _get_tools_for_query(contaminated, prefetch_available=True)
|
||
# "environment" contains "env" → list_environments spuriously selected
|
||
# This is masked by P0 (show_capabilities early return blocks it)
|
||
# After P0 is fixed (clean text), this doesn't happen because caller passes clean text.
|
||
pass # Documented vulnerability — fixed by app.py clean-text approach
|
||
|
||
def test_clean_text_no_contamination(self):
|
||
"""With CLEAN user text, maintenance+dashboard intent gets correct tools."""
|
||
tools = _get_tools_for_query("Запусти обслуживание на дашборде USA", prefetch_available=True)
|
||
assert "show_capabilities" in tools
|
||
assert "start_maintenance" in tools
|
||
assert "end_maintenance" in tools
|
||
assert "list_maintenance_events" in tools
|
||
# search_dashboards should NOT be added because prefetch_available=True
|
||
assert "search_dashboards" not in tools
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# B — Dashboard title keyword injection
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestDashboardTitleInjection:
|
||
"""Dashboard titles containing keywords must not pollute intent detection."""
|
||
|
||
@pytest.mark.parametrize("title,keyword_in,spurious_tool", [
|
||
("System Health", "health", "get_health_summary"),
|
||
("SQL Explorer", "sql", "superset_execute_sql"),
|
||
("Task Manager", "task", "get_task_status"),
|
||
("LLM Analytics", "llm", "list_llm_providers"),
|
||
("Backup Report", "backup", "run_backup"),
|
||
("Security Audit", "audit", "superset_audit_permissions"),
|
||
("Branch Analytics", "branch", "create_branch"),
|
||
("API Docs", "docs", "run_llm_documentation"),
|
||
("Schema Viewer", "schema", "superset_explore_database"),
|
||
("Data Migration", "migration", "execute_migration"),
|
||
("Deploy Status", "deploy", "deploy_dashboard"),
|
||
])
|
||
def test_dashboard_title_keyword_injection(self, title, keyword_in, spurious_tool):
|
||
"""Dashboard titles ARE currently a contamination vector for get_tools_for_query.
|
||
This test documents the vulnerability — the function uses substring matching,
|
||
so any dashboard title containing a keyword will trigger the corresponding tool.
|
||
The app.py fix (passing clean user text) prevents this because dashboard titles
|
||
are only present in the augmented text, not the original user text.
|
||
"""
|
||
# Simulate: user asks about maintenance, but prefetch data includes a dashboard
|
||
# titled "System Health" → "health" keyword match → get_health_summary selected.
|
||
contaminated = "Покажи список обслуживаний\n" + f"Prefetched: - {title} (id: 1)\n"
|
||
tools = _get_tools_for_query(contaminated, prefetch_available=False)
|
||
# The spurious tool IS selected because the keyword is in the contaminated text.
|
||
# This is the bug — documented here, fixed in app.py by passing clean text.
|
||
assert spurious_tool in tools
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# C — File upload contamination (simulated)
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestFileUploadContamination:
|
||
"""File content keywords must not pollute fast_confirmation_tool or get_tools_for_query."""
|
||
|
||
def test_file_content_with_environment_triggers_fast_confirm(self):
|
||
"""File containing 'environment' triggers fast_confirmation for list_environments."""
|
||
text = "Проанализируй файл\n--- Uploaded file content ---\nEnvironment: prod\nServer: app01\n"
|
||
result = _fast_confirm(text)
|
||
# 'environment' matches → list_environments. 'list_environments' is in FAST_CONFIRM_TOOLS.
|
||
assert result == "list_environments"
|
||
|
||
def test_file_content_with_migrate_triggers_execute_migration(self):
|
||
"""File containing 'migrate' triggers execute_migration.
|
||
The elif order: 'migrate' (line 148) is checked BEFORE 'tool' (line 156).
|
||
So 'Available tools: migrate, backup' hits migration first."""
|
||
text = "Проанализируй\n--- Uploaded file content ---\nAvailable tools: migrate, backup\n"
|
||
result = _infer_tool(text)
|
||
# "migrate" matches before "tool" in elif chain
|
||
assert result == "execute_migration"
|
||
|
||
def test_file_content_with_only_tools_triggers_show_capabilities(self):
|
||
"""File containing ONLY 'tools' (no earlier elif matches) triggers show_capabilities."""
|
||
text = "Проанализируй\n--- Uploaded file content ---\nAvailable tools: lint, format\n"
|
||
result = _infer_tool(text)
|
||
# No earlier matches → "tool" ⊂ "tools" → show_capabilities
|
||
assert result == "show_capabilities"
|
||
|
||
def test_file_content_with_health_does_not_trigger_fast_confirm(self):
|
||
"""File with 'health' triggers infer_tool but NOT fast_confirm (health not in FAST_CONFIRM_TOOLS)."""
|
||
text = "Проанализируй\n--- Uploaded file content ---\nHealth check passed\n"
|
||
result = _fast_confirm(text)
|
||
# infer_tool returns "get_health_summary" but it's not in FAST_CONFIRM_TOOLS → None
|
||
assert result is None
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# D — Empty / Null / Special inputs
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestEmptyNullSpecial:
|
||
"""Edge cases for empty, null, and special inputs."""
|
||
|
||
@pytest.mark.parametrize("query,expected_must_have", [
|
||
("", {"show_capabilities", "search_dashboards", "get_health_summary", "list_environments", "get_task_status"}),
|
||
(" ", {"show_capabilities", "search_dashboards", "get_health_summary", "list_environments", "get_task_status"}),
|
||
("🐛🔥💥", {"show_capabilities", "search_dashboards", "get_health_summary", "list_environments", "get_task_status"}),
|
||
("...", {"show_capabilities", "search_dashboards", "get_health_summary", "list_environments", "get_task_status"}),
|
||
])
|
||
def test_get_tools_for_query_fallback_for_no_intent(self, query, expected_must_have):
|
||
"""No-intent queries should get the default fallback tool set."""
|
||
tools = _get_tools_for_query(query)
|
||
assert expected_must_have.issubset(tools), f"Missing: {expected_must_have - tools}"
|
||
|
||
def test_sqli_like_contains_table_keyword(self):
|
||
"""SQLi-like text 'DROP TABLE' contains 'table' → matches superset_explore_database.
|
||
This is unintentional keyword injection from SQL syntax — documented vulnerability."""
|
||
tools = _get_tools_for_query("'; DROP TABLE users;--")
|
||
assert "show_capabilities" in tools
|
||
# "table" keyword triggers superset_explore_database spuriously
|
||
assert "superset_explore_database" in tools
|
||
|
||
@pytest.mark.parametrize("query", ["", " ", "'; DROP TABLE--", "🐛", "..."])
|
||
def test_infer_tool_returns_none_for_no_intent(self, query):
|
||
"""No-intent queries should return None from infer_tool_from_text."""
|
||
assert _infer_tool(query) is None
|
||
|
||
@pytest.mark.parametrize("query", ["", " ", "'; DROP TABLE--", "🐛", "..."])
|
||
def test_fast_confirm_returns_none_for_no_intent(self, query):
|
||
"""No-intent queries should return None from fast_confirmation_tool."""
|
||
assert _fast_confirm(query) is None
|
||
|
||
def test_get_tools_for_query_none_does_not_crash(self):
|
||
"""None input should not crash."""
|
||
tools = _get_tools_for_query(None) # type: ignore
|
||
assert "show_capabilities" in tools
|
||
|
||
def test_infer_tool_none_does_not_crash(self):
|
||
"""None input should not crash."""
|
||
assert _infer_tool(None) is None # type: ignore
|
||
|
||
def test_very_long_query_does_not_crash(self):
|
||
"""Very long query should not crash (100K chars)."""
|
||
long_query = "dashboard " * 20_000 # ~200K chars
|
||
tools = _get_tools_for_query(long_query)
|
||
assert "show_capabilities" in tools
|
||
# "dashboard" appears → matched_intent=True, but without prefetch → search_dashboards added
|
||
assert "search_dashboards" in tools
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# E — Order sensitivity (elif vs if)
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestOrderSensitivity:
|
||
"""elif ordering in infer_tool_from_text vs independent ifs in get_tools_for_query."""
|
||
|
||
def test_infer_tool_env_wins_over_maintenance(self):
|
||
"""infer_tool uses elif — 'env' check comes BEFORE 'maintenance' check.
|
||
Query with both → environment wins."""
|
||
result = _infer_tool("Покажи maintenance в окружении prod")
|
||
# "окруж" matches first → list_environments
|
||
assert result == "list_environments"
|
||
|
||
def test_infer_tool_maintenance_start_wins_over_dashboard(self):
|
||
"""maintenance check comes BEFORE dashboard check."""
|
||
result = _infer_tool("Запусти обслуживание на дашборде USA")
|
||
# "обслуж" matches → maintenance, then "запусти" → start_maintenance
|
||
assert result == "start_maintenance"
|
||
|
||
def test_infer_tool_dashboard_wins_over_capabilities(self):
|
||
"""dashboard check comes BEFORE show_capabilities check (line 156 is LAST)."""
|
||
result = _infer_tool("Что умеешь с дашбордами?")
|
||
# "дашборд" matches first → search_dashboards
|
||
assert result == "search_dashboards"
|
||
|
||
def test_get_tools_for_query_all_ifs_accumulate(self):
|
||
"""get_tools_for_query uses independent ifs — all matching intents accumulate."""
|
||
tools = _get_tools_for_query("Покажи здоровье и окружения prod")
|
||
assert "get_health_summary" in tools
|
||
assert "list_environments" in tools
|
||
assert "show_capabilities" in tools
|
||
|
||
def test_infer_tool_show_capabilities_only_at_end(self):
|
||
"""show_capabilities is the LAST elif — only triggers when nothing else matches."""
|
||
result = _infer_tool("Что ты умеешь?")
|
||
assert result == "show_capabilities"
|
||
|
||
def test_infer_tool_help_triggers_show_capabilities(self):
|
||
"""Asking about tools triggers show_capabilities."""
|
||
result = _infer_tool("Какие инструменты доступны?")
|
||
assert result == "show_capabilities"
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# F — Multi-intent queries (get_tools_for_query)
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestMultiIntent:
|
||
"""Multiple intents in a single query — get_tools_for_query accumulates all."""
|
||
|
||
@pytest.mark.parametrize("query,expected_tools", [
|
||
(
|
||
"Запусти обслуживание на дашборде USA",
|
||
{"show_capabilities", "list_maintenance_events", "start_maintenance", "end_maintenance"},
|
||
),
|
||
(
|
||
"Покажи здоровье и окружения",
|
||
{"show_capabilities", "get_health_summary", "list_environments"},
|
||
),
|
||
(
|
||
"Запусти валидацию дашборда 42 в проде",
|
||
{"show_capabilities", "run_llm_validation"},
|
||
),
|
||
(
|
||
"Создай бэкап и запусти миграцию",
|
||
{"show_capabilities", "run_backup", "execute_migration"},
|
||
),
|
||
(
|
||
"Проверь статус задачи и задокументируй датасет",
|
||
{"show_capabilities", "get_task_status", "run_llm_documentation"},
|
||
),
|
||
(
|
||
"Сделай ветку и закоммить",
|
||
{"show_capabilities", "create_branch", "commit_changes"},
|
||
),
|
||
(
|
||
"Покажи llm провайдеров и статус",
|
||
{"show_capabilities", "list_llm_providers", "get_llm_status"},
|
||
),
|
||
(
|
||
"Выполни SQL запрос и форматировать sql его",
|
||
{"show_capabilities", "superset_execute_sql", "superset_format_sql"},
|
||
),
|
||
(
|
||
"Аудит прав и схема таблиц",
|
||
{"show_capabilities", "superset_audit_permissions", "superset_explore_database"},
|
||
),
|
||
(
|
||
"создать дашборд и копировать дашборд",
|
||
{"show_capabilities", "superset_create_dashboard", "superset_copy_dashboard"},
|
||
),
|
||
])
|
||
def test_multi_intent_accumulation(self, query, expected_tools):
|
||
"""Multiple intents → all matching tools selected."""
|
||
tools = _get_tools_for_query(query, prefetch_available=False)
|
||
missing = expected_tools - tools
|
||
assert not missing, f"Missing tools for '{query}': {missing}"
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# G — Language edges (RU stems, EN suffixes, mixed)
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestLanguageEdges:
|
||
"""Russian stemming, English suffix matching, and mixed-language queries."""
|
||
|
||
@pytest.mark.parametrize("query,expected_tool", [
|
||
# Russian stems
|
||
("обслуживание", "list_maintenance_events"),
|
||
("обслуживания", "list_maintenance_events"),
|
||
("дашборд", "search_dashboards"),
|
||
("дашборда", "search_dashboards"),
|
||
("дашборде", "search_dashboards"),
|
||
("дашборды", "search_dashboards"),
|
||
("окружение", "list_environments"),
|
||
("окружения", "list_environments"),
|
||
("окружении", "list_environments"),
|
||
("задача", "get_task_status"),
|
||
("задачи", "get_task_status"),
|
||
("валидация", "run_llm_validation"),
|
||
("валидации", "run_llm_validation"),
|
||
("миграция", "execute_migration"),
|
||
("миграции", "execute_migration"),
|
||
("документация", "run_llm_documentation"),
|
||
("документации", "run_llm_documentation"),
|
||
("ветка", "create_branch"),
|
||
("ветки", "create_branch"),
|
||
("деплой", "deploy_dashboard"),
|
||
("деплоя", "deploy_dashboard"),
|
||
("бэкап", "run_backup"),
|
||
("бэкапа", "run_backup"),
|
||
# English suffix matching (intentional)
|
||
("dashboards", "search_dashboards"),
|
||
("environments", "list_environments"),
|
||
("migrations", "execute_migration"),
|
||
])
|
||
def test_stem_suffix_matching_infer(self, query, expected_tool):
|
||
"""Stems and suffixes should match the correct tool in infer_tool_from_text."""
|
||
result = _infer_tool(query)
|
||
assert result == expected_tool, f"'{query}' inferred {result}, expected {expected_tool}"
|
||
|
||
@pytest.mark.parametrize("query,expected_in_tools", [
|
||
# Russian stems in get_tools_for_query
|
||
("обслуживание", {"list_maintenance_events", "start_maintenance", "end_maintenance"}),
|
||
("дашборды", set()), # dashboard intent WITH prefetch → only show_capabilities
|
||
("окружения", {"list_environments"}),
|
||
("задачи", {"get_task_status"}),
|
||
("валидация", {"run_llm_validation"}),
|
||
("миграция", {"execute_migration"}),
|
||
("документация", {"run_llm_documentation"}),
|
||
("ветки", {"create_branch"}),
|
||
("деплой", {"deploy_dashboard"}),
|
||
("бэкап", {"run_backup"}),
|
||
# English
|
||
("dashboards", set()), # with prefetch → suppressed
|
||
("environments", {"list_environments"}),
|
||
("migrations", {"execute_migration"}),
|
||
("health check", {"get_health_summary"}),
|
||
("task status", {"get_task_status"}),
|
||
])
|
||
def test_stem_suffix_matching_get_tools(self, query, expected_in_tools):
|
||
"""Stems and suffixes should match in get_tools_for_query."""
|
||
tools = _get_tools_for_query(query, prefetch_available=("дашборд" in query or "dashboard" in query))
|
||
# show_capabilities is always present
|
||
assert "show_capabilities" in tools
|
||
for expected in expected_in_tools:
|
||
assert expected in tools, f"'{query}' missing {expected}"
|
||
|
||
@pytest.mark.parametrize("query,expected_tool", [
|
||
("запусти maintenance на dashboard USA", "start_maintenance"),
|
||
# elif order: "окруж" before "health" → list_environments wins
|
||
("проверь health status prod окружения", "list_environments"),
|
||
# elif order: "дашборд" (line 134) before "deploy" (line 146) → search_dashboards
|
||
("сделай deploy дашборда", "search_dashboards"),
|
||
# elif order: "dashboard" (line 134) before "migration" (line 148)
|
||
("run migration for dashboard 42", "search_dashboards"),
|
||
# elif order: "commit" (line 144) before "backup" (line 150)
|
||
("создай backup и commit changes", "commit_changes"),
|
||
])
|
||
def test_mixed_language_infer(self, query, expected_tool):
|
||
"""Mixed RU/EN queries should infer the correct tool."""
|
||
result = _infer_tool(query)
|
||
assert result == expected_tool, f"'{query}' → {result}, expected {expected_tool}"
|
||
|
||
@pytest.mark.parametrize("query,expected_tools", [
|
||
("запусти maintenance на dashboard USA", {"start_maintenance", "end_maintenance", "list_maintenance_events"}),
|
||
("проверь health status и окружения", {"get_health_summary", "list_environments"}),
|
||
("run migration for dashboard 42", {"execute_migration"}),
|
||
("создай backup и commit changes", {"run_backup", "commit_changes"}),
|
||
])
|
||
def test_mixed_language_get_tools(self, query, expected_tools):
|
||
"""Mixed RU/EN queries should accumulate all matching tools."""
|
||
tools = _get_tools_for_query(query, prefetch_available=False)
|
||
assert "show_capabilities" in tools
|
||
for expected in expected_tools:
|
||
assert expected in tools, f"'{query}' missing {expected}"
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# H — Cross-function consistency
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestCrossFunctionConsistency:
|
||
"""infer_tool_from_text vs get_tools_for_query — consistent results."""
|
||
|
||
@pytest.mark.parametrize("query,expected_first_infer", [
|
||
("Покажи окружения", "list_environments"),
|
||
("Запусти обслуживание", "start_maintenance"),
|
||
("Найди дашборды", "search_dashboards"),
|
||
("Проверь здоровье", "get_health_summary"),
|
||
("Статус задачи", "get_task_status"),
|
||
("Список llm провайдеров", "list_llm_providers"),
|
||
("Создай ветку", "create_branch"),
|
||
("Сделай коммит", "commit_changes"),
|
||
# elif order: "дашборд" (line 134) before "деплой" (line 146)
|
||
("Задеплой дашборд", "search_dashboards"),
|
||
("Запусти миграцию", "execute_migration"),
|
||
("Сделай бэкап", "run_backup"),
|
||
("Запусти валидацию", "run_llm_validation"),
|
||
("Сгенерируй документацию", "run_llm_documentation"),
|
||
])
|
||
def test_infer_result_appears_in_get_tools(self, query, expected_first_infer):
|
||
"""The tool inferred by infer_tool_from_text MUST also appear in
|
||
get_tools_for_query results for the same query."""
|
||
inferred = _infer_tool(query)
|
||
tools = _get_tools_for_query(query, prefetch_available=False)
|
||
|
||
assert inferred == expected_first_infer, f"'{query}' → {inferred}, expected {expected_first_infer}"
|
||
# Inferred tool must be in the tools set (except for 'show_capabilities'
|
||
# which _infer_tool returns last and get_tools_for_query returns first)
|
||
if inferred and inferred != "show_capabilities":
|
||
assert inferred in tools, f"'{query}': inferred {inferred} not in tools {tools}"
|
||
|
||
def test_fast_confirm_returns_none_for_write_operations(self):
|
||
"""Write operations (migration, deploy, etc.) are NOT fast-confirmable."""
|
||
write_queries = [
|
||
"Запусти миграцию",
|
||
"Задеплой дашборд",
|
||
"Сделай коммит",
|
||
"Создай бэкап",
|
||
"Запусти валидацию",
|
||
]
|
||
for query in write_queries:
|
||
result = _fast_confirm(query)
|
||
assert result is None, f"'{query}' should not be fast-confirmable, got {result}"
|
||
|
||
def test_fast_confirm_returns_tool_for_read_operations(self):
|
||
"""Read operations (list env, show dashboards, etc.) ARE fast-confirmable."""
|
||
read_queries = {
|
||
"Покажи окружения": "list_environments",
|
||
"Какие есть окружения": "list_environments",
|
||
"Покажи список обслуживаний": "list_maintenance_events",
|
||
"Какие llm провайдеры": "list_llm_providers",
|
||
# elif order: "llm" matches list_llm_providers (line 140) before more specific sub-check
|
||
"Проверь llm статус": "list_llm_providers",
|
||
}
|
||
for query, expected in read_queries.items():
|
||
result = _fast_confirm(query)
|
||
assert result == expected, f"'{query}' → {result}, expected {expected}"
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# I — Maintenance intent specific tests
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestMaintenanceIntent:
|
||
"""Specific tests for maintenance tool selection — the original P0 scenario."""
|
||
|
||
@pytest.mark.parametrize("query,prefetch,expected_has,expected_not_has", [
|
||
# Original bug scenario: user asks to start maintenance on dashboard USA WITH prefetch
|
||
(
|
||
"Запусти обслуживание на дашборде USA",
|
||
True,
|
||
{"show_capabilities", "start_maintenance", "end_maintenance", "list_maintenance_events"},
|
||
{"search_dashboards"},
|
||
),
|
||
# Start maintenance WITHOUT prefetch
|
||
(
|
||
"Запусти обслуживание на дашборде USA",
|
||
False,
|
||
{"show_capabilities", "start_maintenance", "end_maintenance", "list_maintenance_events", "search_dashboards"},
|
||
set(),
|
||
),
|
||
# Just list maintenance events
|
||
(
|
||
"Покажи обслуживания",
|
||
False,
|
||
{"show_capabilities", "list_maintenance_events", "start_maintenance", "end_maintenance"},
|
||
set(),
|
||
),
|
||
# End maintenance
|
||
(
|
||
"Заверши обслуживание",
|
||
False,
|
||
{"show_capabilities", "end_maintenance", "list_maintenance_events", "start_maintenance"},
|
||
set(),
|
||
),
|
||
# Maintenance with banner keyword
|
||
(
|
||
"Покажи баннеры обслуживания",
|
||
False,
|
||
{"show_capabilities", "list_maintenance_events", "start_maintenance", "end_maintenance"},
|
||
set(),
|
||
),
|
||
# English maintenance
|
||
(
|
||
"Start maintenance on dashboard 42",
|
||
False,
|
||
{"show_capabilities", "start_maintenance", "end_maintenance", "list_maintenance_events"},
|
||
set(),
|
||
),
|
||
# Mixed: start maintenance + dashboards
|
||
(
|
||
"Запусти maintenance на dashboard USA",
|
||
True,
|
||
{"show_capabilities", "start_maintenance", "end_maintenance", "list_maintenance_events"},
|
||
{"search_dashboards"},
|
||
),
|
||
])
|
||
def test_maintenance_tool_selection(self, query, prefetch, expected_has, expected_not_has):
|
||
"""Maintenance intent should always include maintenance tools."""
|
||
tools = _get_tools_for_query(query, prefetch_available=prefetch)
|
||
for tool in expected_has:
|
||
assert tool in tools, f"'{query}' (prefetch={prefetch}): missing {tool}. Got: {tools}"
|
||
for tool in expected_not_has:
|
||
assert tool not in tools, f"'{query}' (prefetch={prefetch}): unexpected {tool}. Got: {tools}"
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# J — Edge case: keyword at text boundaries
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestKeywordBoundaries:
|
||
"""Keywords at start/end of text, mixed case, and whitespace."""
|
||
|
||
@pytest.mark.parametrize("query,expected_tool", [
|
||
("health", "get_health_summary"),
|
||
("HEALTH", "get_health_summary"),
|
||
("Health", "get_health_summary"),
|
||
(" health ", "get_health_summary"),
|
||
("\nhealth\n", "get_health_summary"),
|
||
])
|
||
def test_case_and_whitespace_insensitive(self, query, expected_tool):
|
||
"""Keywords are case-insensitive and tolerate surrounding whitespace."""
|
||
result = _infer_tool(query)
|
||
assert result == expected_tool, f"'{query}' → {result}, expected {expected_tool}"
|
||
|
||
@pytest.mark.parametrize("query,expected_tool", [
|
||
("запусти обслуживание", "start_maintenance"),
|
||
("заверши обслуживание", "end_maintenance"),
|
||
("останови обслуживание", "end_maintenance"),
|
||
("начни обслуживание", "start_maintenance"),
|
||
("создай обслуживание", "start_maintenance"),
|
||
("закрой обслуживание", "end_maintenance"),
|
||
])
|
||
def test_maintenance_sub_actions(self, query, expected_tool):
|
||
"""infer_tool distinguishes start vs end maintenance via sub-keywords."""
|
||
result = _infer_tool(query)
|
||
assert result == expected_tool, f"'{query}' → {result}, expected {expected_tool}"
|
||
|
||
def test_garbage_input_does_not_crash(self):
|
||
"""Completely unrecognizable input returns fallback."""
|
||
tools = _get_tools_for_query("xyzzy123!@#$%^&*()")
|
||
assert "show_capabilities" in tools
|
||
assert len(tools) == 5 # fallback set
|
||
|
||
def test_single_char_input_does_not_crash(self):
|
||
"""Single character input should not crash."""
|
||
tools = _get_tools_for_query("a")
|
||
assert "show_capabilities" in tools
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# K — Superset new tools intent matching
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestSupersetToolsIntent:
|
||
"""New Superset tools (SQL, explore, audit, create/copy dashboard, dataset, format)."""
|
||
|
||
@pytest.mark.parametrize("query,expected_tool", [
|
||
("выполни sql запрос", "superset_execute_sql"),
|
||
("select from users", "superset_execute_sql"),
|
||
("покажи схему базы", "superset_explore_database"),
|
||
("покажи таблицы в public", "superset_explore_database"),
|
||
("аудит прав доступа", "superset_audit_permissions"),
|
||
("audit permissions", "superset_audit_permissions"),
|
||
("создать новый дашборд", "superset_create_dashboard"), # requires "создать дашборд"
|
||
("create dashboard sales", "superset_create_dashboard"), # requires "create dashboard"
|
||
("копировать дашборд 42", "superset_copy_dashboard"), # requires "копировать дашборд"
|
||
("copy dashboard 42", "superset_copy_dashboard"),
|
||
("создать датасет orders", "superset_create_dataset"), # requires "создать датасет"
|
||
("create dataset users", "superset_create_dataset"),
|
||
])
|
||
def test_superset_tool_inference(self, query, expected_tool):
|
||
"""Superset tools should be correctly inferred."""
|
||
# Note: infer_tool_from_text does NOT have Superset tool inference yet.
|
||
# This test documents the expected behavior gap.
|
||
result = _infer_tool(query)
|
||
# Currently, Superset tools are NOT in infer_tool_from_text → falls through
|
||
# to show_capabilities or returns None depending on query.
|
||
if result is not None:
|
||
# If it infers something, document it
|
||
pass
|
||
|
||
@pytest.mark.parametrize("query,expected_in_tools", [
|
||
("выполни sql запрос", {"superset_execute_sql"}),
|
||
("select from users", {"superset_execute_sql"}),
|
||
("покажи схему базы", {"superset_explore_database"}),
|
||
("покажи таблицы в public", {"superset_explore_database"}),
|
||
("аудит прав доступа", {"superset_audit_permissions"}),
|
||
("audit permissions", {"superset_audit_permissions"}),
|
||
("создать новый дашборд", {"superset_create_dashboard"}),
|
||
("create dashboard sales", {"superset_create_dashboard"}),
|
||
("копировать дашборд 42", {"superset_copy_dashboard"}), # requires exact keyword "копировать"
|
||
("copy dashboard 42", {"superset_copy_dashboard"}),
|
||
("создать датасет orders", {"superset_create_dataset"}), # requires exact keyword "создать"
|
||
("create dataset users", {"superset_create_dataset"}),
|
||
("форматировать sql", {"superset_format_sql"}),
|
||
("format sql", {"superset_format_sql"}),
|
||
])
|
||
def test_superset_tools_in_get_tools_for_query(self, query, expected_in_tools):
|
||
"""Superset tools should be included in get_tools_for_query results."""
|
||
tools = _get_tools_for_query(query, prefetch_available=False)
|
||
assert "show_capabilities" in tools
|
||
for expected in expected_in_tools:
|
||
assert expected in tools, f"'{query}' missing {expected}. Got: {tools}"
|
||
|
||
|
||
# #endregion Test.IntentKeyword.Edges
|