busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add 7 more test modules — assistant cmd parser, history, dispatch, admin routes, dataset review, maintenance + semantic resolver

Browse Source
This commit is contained in:
busya
2026-06-15 16:06:18 +03:00
parent f75c15dbc6
commit fc7e6c3c15
9 changed files with 3241 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

356
backend/tests/api/test_assistant_llm_planner_intent.py Normal file
View File

@@ -0,0 +1,356 @@
# #region Test.Assistant.LlmPlannerIntent [C:2] [TYPE Module] [SEMANTICS test,assistant,llm,planner,intent,authorization]
# @BRIEF Tests for _llm_planner_intent.py — LLM-based intent planning and authorization.
# @RELATION BINDS_TO -> [AssistantLlmPlannerIntent]
# @TEST_EDGE: plan_intent_empty_tools -> Returns None when tools list empty
# @TEST_EDGE: plan_intent_no_provider -> Returns None when no LLM provider
# @TEST_EDGE: plan_intent_llm_failure -> Returns None on LLM error
# @TEST_EDGE: plan_intent_clarify -> Returns clarify intent for ambiguous input
# @TEST_EDGE: plan_intent_valid_operation -> Valid operation mapped correctly
# @TEST_EDGE: plan_intent_production_env -> Production env sets dangerous risk
# @TEST_EDGE: authorize_intent -> Permission checks pass or raise 403
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
from fastapi import HTTPException
class TestPlanIntentWithLlm:
"""_plan_intent_with_llm — LLM-based intent planning."""
@pytest.fixture
def db(self):
return MagicMock()
@pytest.fixture
def config_manager(self):
cm = MagicMock()
cm.get_config.return_value.settings.llm = {
"assistant_planner_provider": "openai",
"assistant_planner_model": "gpt-4",
}
cm.get_environments.return_value = []
return cm
def test_empty_tools_returns_none(self, db, config_manager):
from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
result = __import__("asyncio").run(_plan_intent_with_llm("test", [], db, config_manager))
assert result is None
def test_no_provider_returns_none(self, db, config_manager):
from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
tools = [{"operation": "run_backup", "domain": "backup"}]
with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc:
mock_svc.return_value.get_all_providers.return_value = []
result = __import__("asyncio").run(
_plan_intent_with_llm("test", tools, db, config_manager)
)
assert result is None
def test_no_api_key_returns_none(self, db, config_manager):
from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
tools = [{"operation": "run_backup", "domain": "backup"}]
provider_mock = MagicMock()
provider_mock.id = "p-1"
provider_mock.provider_type = "openai"
provider_mock.base_url = None
provider_mock.default_model = "gpt-4"
with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls:
mock_svc = MagicMock()
mock_svc_cls.return_value = mock_svc
mock_svc.get_all_providers.return_value = [provider_mock]
mock_svc.get_decrypted_api_key.return_value = None
with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve:
mock_resolve.return_value = "p-1"
result = __import__("asyncio").run(
_plan_intent_with_llm("test", tools, db, config_manager)
)
assert result is None
def test_llm_failure_returns_none(self, db, config_manager):
from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
tools = [{"operation": "run_backup", "domain": "backup"}]
provider_mock = MagicMock()
provider_mock.id = "p-1"
provider_mock.provider_type = "openai"
provider_mock.base_url = None
provider_mock.default_model = "gpt-4"
with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
mock_svc = MagicMock()
mock_svc_cls.return_value = mock_svc
mock_svc.get_all_providers.return_value = [provider_mock]
mock_svc.get_decrypted_api_key.return_value = "sk-test"
mock_client = AsyncMock()
mock_client_cls.return_value = mock_client
mock_client.get_json_completion.side_effect = Exception("LLM unavailable")
with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve:
mock_resolve.return_value = "p-1"
result = __import__("asyncio").run(
_plan_intent_with_llm("test", tools, db, config_manager)
)
assert result is None
def test_clarify_response(self, db, config_manager):
from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
tools = [{"operation": "run_backup", "domain": "backup"}]
provider_mock = MagicMock()
provider_mock.id = "p-1"
provider_mock.provider_type = "openai"
provider_mock.base_url = None
provider_mock.default_model = "gpt-4"
with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
mock_svc = MagicMock()
mock_svc_cls.return_value = mock_svc
mock_svc.get_all_providers.return_value = [provider_mock]
mock_svc.get_decrypted_api_key.return_value = "sk-test"
mock_client = AsyncMock()
mock_client_cls.return_value = mock_client
mock_client.get_json_completion.return_value = {
"operation": "clarify",
"domain": "unknown",
"confidence": 0.3,
"clarification": "Что именно вы хотите сделать?",
"entities": {},
}
with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve:
mock_resolve.return_value = "p-1"
result = __import__("asyncio").run(
_plan_intent_with_llm("test", tools, db, config_manager)
)
assert result is not None
assert result["operation"] == "clarify"
assert "Что именно" in result.get("clarification", "")
def test_invalid_operation_returns_none(self, db, config_manager):
from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
tools = [{"operation": "run_backup", "domain": "backup"}]
provider_mock = MagicMock()
provider_mock.id = "p-1"
provider_mock.provider_type = "openai"
provider_mock.base_url = None
provider_mock.default_model = "gpt-4"
with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
mock_svc = MagicMock()
mock_svc_cls.return_value = mock_svc
mock_svc.get_all_providers.return_value = [provider_mock]
mock_svc.get_decrypted_api_key.return_value = "sk-test"
mock_client = AsyncMock()
mock_client_cls.return_value = mock_client
mock_client.get_json_completion.return_value = {
"operation": "nonexistent_op",
"domain": "backup",
"confidence": 0.9,
"entities": {},
}
with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve:
mock_resolve.return_value = "p-1"
result = __import__("asyncio").run(
_plan_intent_with_llm("test", tools, db, config_manager)
)
assert result is None
def test_non_dict_response_returns_none(self, db, config_manager):
from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
tools = [{"operation": "run_backup", "domain": "backup"}]
provider_mock = MagicMock()
provider_mock.id = "p-1"
provider_mock.provider_type = "openai"
provider_mock.base_url = None
provider_mock.default_model = "gpt-4"
with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
mock_svc = MagicMock()
mock_svc_cls.return_value = mock_svc
mock_svc.get_all_providers.return_value = [provider_mock]
mock_svc.get_decrypted_api_key.return_value = "sk-test"
mock_client = AsyncMock()
mock_client_cls.return_value = mock_client
mock_client.get_json_completion.return_value = "not a dict"
with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve:
mock_resolve.return_value = "p-1"
result = __import__("asyncio").run(
_plan_intent_with_llm("test", tools, db, config_manager)
)
assert result is None
def test_valid_operation_with_defaults(self, db, config_manager):
from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
tools = [{
"operation": "run_backup",
"domain": "backup",
"risk_level": "guarded",
"requires_confirmation": False,
"defaults": {"environment": "dev"},
}]
provider_mock = MagicMock()
provider_mock.id = "p-1"
provider_mock.provider_type = "openai"
provider_mock.base_url = None
provider_mock.default_model = "gpt-4"
with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
mock_svc = MagicMock()
mock_svc_cls.return_value = mock_svc
mock_svc.get_all_providers.return_value = [provider_mock]
mock_svc.get_decrypted_api_key.return_value = "sk-test"
mock_client = AsyncMock()
mock_client_cls.return_value = mock_client
mock_client.get_json_completion.return_value = {
"operation": "run_backup",
"domain": "backup",
"confidence": 0.9,
"entities": {},
"risk_level": "guarded",
"requires_confirmation": False,
}
with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve, \
patch("src.api.routes.assistant._llm_planner_intent._coerce_intent_entities") as mock_coerce, \
patch("src.api.routes.assistant._llm_planner_intent.normalize_llm_settings") as mock_norm:
mock_resolve.return_value = "p-1"
mock_coerce.return_value = {
"domain": "backup",
"operation": "run_backup",
"entities": {},
"confidence": 0.9,
"risk_level": "guarded",
"requires_confirmation": False,
}
mock_norm.return_value = {
"assistant_planner_provider": "openai",
"assistant_planner_model": "gpt-4",
}
result = __import__("asyncio").run(
_plan_intent_with_llm("test", tools, db, config_manager)
)
assert result is not None
assert result["operation"] == "run_backup"
# Default should be applied
assert result["entities"].get("environment") == "dev"
def test_production_deploy_sets_dangerous(self, db, config_manager):
from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
# Environments must be JSON-serializable dicts
config_manager.get_environments.return_value = [
type('Env', (), {'id': 'prod', 'name': 'PROD'})(),
]
tools = [{
"operation": "deploy_dashboard",
"domain": "git",
"risk_level": "guarded",
"requires_confirmation": False,
}]
provider_mock = MagicMock()
provider_mock.id = "p-1"
provider_mock.provider_type = "openai"
provider_mock.base_url = None
provider_mock.default_model = "gpt-4"
with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
mock_svc = MagicMock()
mock_svc_cls.return_value = mock_svc
mock_svc.get_all_providers.return_value = [provider_mock]
mock_svc.get_decrypted_api_key.return_value = "sk-test"
mock_client = AsyncMock()
mock_client_cls.return_value = mock_client
mock_client.get_json_completion.return_value = {
"operation": "deploy_dashboard",
"domain": "git",
"confidence": 0.9,
"entities": {"environment": "prod"},
"risk_level": "guarded",
"requires_confirmation": False,
}
with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve, \
patch("src.api.routes.assistant._llm_planner_intent._coerce_intent_entities") as mock_coerce, \
patch("src.api.routes.assistant._llm_planner_intent._is_production_env") as mock_is_prod, \
patch("src.api.routes.assistant._llm_planner_intent.normalize_llm_settings") as mock_norm:
mock_resolve.return_value = "p-1"
mock_coerce.return_value = {
"domain": "git",
"operation": "deploy_dashboard",
"entities": {"environment": "prod"},
"confidence": 0.9,
"risk_level": "guarded",
"requires_confirmation": False,
}
mock_is_prod.return_value = True
mock_norm.return_value = {
"assistant_planner_provider": "openai",
"assistant_planner_model": "gpt-4",
}
result = __import__("asyncio").run(
_plan_intent_with_llm("test", tools, db, config_manager)
)
assert result is not None
assert result["risk_level"] == "dangerous"
assert result["requires_confirmation"] is True
class TestAuthorizeIntent:
"""_authorize_intent — permission checks."""
def test_authorized(self):
from src.api.routes.assistant._llm_planner_intent import _authorize_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=[])
intent = {"operation": "run_backup"}
with patch("src.api.routes.assistant._llm_planner_intent.get_permission_checks") as mock_checks, \
patch("src.api.routes.assistant._llm_planner_intent._check_any_permission") as mock_check:
mock_checks.return_value = {"run_backup": [("backup", "WRITE")]}
_authorize_intent(intent, user)
mock_check.assert_called_once_with(user, [("backup", "WRITE")])
def test_no_checks_skips(self):
from src.api.routes.assistant._llm_planner_intent import _authorize_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=[])
intent = {"operation": "show_capabilities"}
with patch("src.api.routes.assistant._llm_planner_intent.get_permission_checks") as mock_checks:
mock_checks.return_value = {} # no checks for this op
# Should not raise
_authorize_intent(intent, user)
# #endregion Test.Assistant.LlmPlannerIntent