141 lines
5.9 KiB
Python
141 lines
5.9 KiB
Python
# #region Test.Assistant.Dispatch [C:2] [TYPE Module] [SEMANTICS test,assistant,dispatch,confirmation,clarification]
|
|
# @BRIEF Tests for _dispatch.py — clarification text, confirmation summary, and dispatch wrapper.
|
|
# @RELATION BINDS_TO -> [AssistantDispatch]
|
|
# @TEST_EDGE: clarification_text -> Missing parameter guidance per operation
|
|
# @TEST_EDGE: confirmation_summary -> Confirmation prompts for supported ops
|
|
# @TEST_EDGE: confirmation_summary_unknown -> Fallback for unknown ops
|
|
# @TEST_EDGE: confirmation_summary_migration -> Migration summary with flags and dry-run
|
|
# @TEST_EDGE: dispatch_intent -> Backward-compat wrapper calls registry
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
import pytest
|
|
|
|
|
|
class TestClarificationTextForIntent:
|
|
"""_clarification_text_for_intent — per-operation guidance texts."""
|
|
|
|
@pytest.mark.parametrize("operation,expected_keyword", [
|
|
("run_llm_validation", "LLM-валидации"),
|
|
("run_llm_documentation", "документации"),
|
|
("create_branch", "ветки"),
|
|
("commit_changes", "коммита"),
|
|
("deploy_dashboard", "дашборд"),
|
|
("execute_migration", "source_env"),
|
|
("run_backup", "окружение"),
|
|
("unknown_op", "default"),
|
|
])
|
|
def test_clarification_text(self, operation, expected_keyword):
|
|
from src.api.routes.assistant._dispatch import _clarification_text_for_intent
|
|
|
|
intent = {"operation": operation}
|
|
detail = "default clarification needed"
|
|
text = _clarification_text_for_intent(intent, detail)
|
|
if expected_keyword == "default":
|
|
assert text == detail
|
|
else:
|
|
assert expected_keyword in text
|
|
|
|
|
|
class TestAsyncConfirmationSummary:
|
|
"""_async_confirmation_summary — confirmation prompt generation."""
|
|
|
|
@pytest.fixture
|
|
def config_manager(self):
|
|
cm = MagicMock()
|
|
cm.get_environments = MagicMock(return_value=[])
|
|
return cm
|
|
|
|
def _make_intent(self, operation, entities=None):
|
|
return {"operation": operation, "entities": entities or {}}
|
|
|
|
@pytest.mark.parametrize("operation,expected_title_word", [
|
|
("create_branch", "ветки"),
|
|
("commit_changes", "Коммит"),
|
|
("deploy_dashboard", "Деплой"),
|
|
("execute_migration", "Миграция"),
|
|
("run_backup", "Бэкап"),
|
|
("run_llm_validation", "LLM-валидация"),
|
|
("run_llm_documentation", "документации"),
|
|
("start_maintenance", "Обслуживание"),
|
|
("end_maintenance", "Завершение"),
|
|
])
|
|
def test_known_operations(self, operation, expected_title_word, config_manager):
|
|
from src.api.routes.assistant._dispatch import _async_confirmation_summary
|
|
|
|
db = MagicMock()
|
|
intent = self._make_intent(operation, {
|
|
"dashboard_id": "42",
|
|
"branch_name": "feature/test",
|
|
"environment": "dev",
|
|
"source_env": "dev",
|
|
"target_env": "prod",
|
|
"tables": ["users", "orders"],
|
|
"dataset_id": "123",
|
|
})
|
|
text = __import__("asyncio").run(_async_confirmation_summary(intent, config_manager, db))
|
|
assert expected_title_word in text
|
|
|
|
def test_unknown_operation(self, config_manager):
|
|
from src.api.routes.assistant._dispatch import _async_confirmation_summary
|
|
|
|
db = MagicMock()
|
|
intent = self._make_intent("unknown_operation")
|
|
text = __import__("asyncio").run(_async_confirmation_summary(intent, config_manager, db))
|
|
assert "Подтвердите" in text
|
|
|
|
def test_migration_with_flags(self, config_manager):
|
|
from src.api.routes.assistant._dispatch import _async_confirmation_summary
|
|
|
|
db = MagicMock()
|
|
intent = self._make_intent("execute_migration", {
|
|
"dashboard_id": "42",
|
|
"source_env": "dev",
|
|
"target_env": "staging",
|
|
"replace_db_config": True,
|
|
"fix_cross_filters": True,
|
|
"dry_run": False,
|
|
})
|
|
text = __import__("asyncio").run(_async_confirmation_summary(intent, config_manager, db))
|
|
assert "маппинг" in text # should mention db mapping
|
|
assert "cross" in text or "кросс" in text
|
|
|
|
def test_migration_with_tables_list(self, config_manager):
|
|
from src.api.routes.assistant._dispatch import _async_confirmation_summary
|
|
|
|
db = MagicMock()
|
|
intent = self._make_intent("start_maintenance", {
|
|
"tables": ["users", "orders", "products", "reviews", "logs", "extra"],
|
|
})
|
|
text = __import__("asyncio").run(_async_confirmation_summary(intent, config_manager, db))
|
|
assert "users" in text
|
|
|
|
|
|
class TestDispatchIntent:
|
|
"""_dispatch_intent — backward-compat wrapper."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatches_to_registry(self):
|
|
from src.api.routes.assistant._dispatch import _dispatch_intent
|
|
from src.schemas.auth import User
|
|
from datetime import UTC, datetime
|
|
|
|
user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
task_manager = MagicMock()
|
|
config_manager = MagicMock()
|
|
db = MagicMock()
|
|
|
|
intent = {"operation": "run_backup", "entities": {"environment": "dev"}}
|
|
|
|
with patch("src.api.routes.assistant._dispatch._registry_dispatch",
|
|
new=AsyncMock(return_value=("Backup started", "task-42", []))) as mock_dispatch:
|
|
text, task_id, actions = await _dispatch_intent(intent, user, task_manager, config_manager, db)
|
|
assert text == "Backup started"
|
|
assert task_id == "task-42"
|
|
mock_dispatch.assert_called_once_with("run_backup", intent, user, task_manager, config_manager, db)
|
|
# #endregion Test.Assistant.Dispatch
|