448 lines
15 KiB
Python
448 lines
15 KiB
Python
"""Step definitions for provider_fixes.feature.
|
|
|
|
All step text is unique to this file to avoid collisions with existing
|
|
step definitions in other feature files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given — imports
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I import AutoDebugAgent for provider fix tests")
|
|
def step_import_auto_debug_for_fix(context: Context) -> None:
|
|
from cleveragents.agents.graphs.auto_debug import AutoDebugAgent
|
|
|
|
context.AutoDebugAgent = AutoDebugAgent
|
|
|
|
|
|
@given("I import ContextAnalysisAgent for provider fix tests")
|
|
def step_import_context_analysis_for_fix(context: Context) -> None:
|
|
from cleveragents.agents.graphs.context_analysis import (
|
|
ContextAnalysisAgent,
|
|
)
|
|
|
|
context.ContextAnalysisAgent = ContextAnalysisAgent
|
|
|
|
|
|
@given("I import PlanGenerationGraph for provider fix tests")
|
|
def step_import_plan_generation_for_fix(context: Context) -> None:
|
|
from cleveragents.agents.graphs.plan_generation import (
|
|
PlanGenerationGraph,
|
|
)
|
|
|
|
context.PlanGenerationGraph = PlanGenerationGraph
|
|
|
|
|
|
@given("I import the provider exception classes")
|
|
def step_import_provider_exceptions(context: Context) -> None:
|
|
from cleveragents.core.exceptions import (
|
|
ProviderError,
|
|
ProviderNotConfiguredError,
|
|
)
|
|
|
|
context.ProviderError = ProviderError
|
|
context.ProviderNotConfiguredError = ProviderNotConfiguredError
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When — agent creation without LLM (should raise)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I attempt to create an AutoDebugAgent without an LLM")
|
|
def step_attempt_auto_debug_no_llm(context: Context) -> None:
|
|
from cleveragents.core.exceptions import ProviderNotConfiguredError
|
|
|
|
context.raised_error = None
|
|
try:
|
|
context.AutoDebugAgent()
|
|
except ProviderNotConfiguredError as exc:
|
|
context.raised_error = exc
|
|
|
|
|
|
@when("I attempt to create a ContextAnalysisAgent without LLM")
|
|
def step_attempt_context_analysis_no_llm(context: Context) -> None:
|
|
from cleveragents.core.exceptions import ProviderNotConfiguredError
|
|
|
|
context.raised_error = None
|
|
try:
|
|
context.ContextAnalysisAgent()
|
|
except ProviderNotConfiguredError as exc:
|
|
context.raised_error = exc
|
|
|
|
|
|
@when("I attempt to create a PlanGenerationGraph without an LLM")
|
|
def step_attempt_plan_generation_no_llm(context: Context) -> None:
|
|
from cleveragents.core.exceptions import ProviderNotConfiguredError
|
|
|
|
context.raised_error = None
|
|
try:
|
|
context.PlanGenerationGraph()
|
|
except ProviderNotConfiguredError as exc:
|
|
context.raised_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When — agent creation with explicit LLM (should succeed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create an AutoDebugAgent with a mock FakeListLLM")
|
|
def step_create_auto_debug_with_fake(context: Context) -> None:
|
|
from langchain_community.llms import FakeListLLM
|
|
|
|
llm = FakeListLLM(responses=["mock"] * 3)
|
|
context.agent = context.AutoDebugAgent(llm=llm)
|
|
|
|
|
|
@when("I create a ContextAnalysisAgent with a mock FakeListLLM")
|
|
def step_create_context_with_fake(context: Context) -> None:
|
|
from langchain_community.llms import FakeListLLM
|
|
|
|
llm = FakeListLLM(
|
|
responses=[
|
|
"Dependencies: ['os']",
|
|
"Relevance: High",
|
|
"Summary: Test",
|
|
]
|
|
)
|
|
context.context_agent = context.ContextAnalysisAgent(llm=llm)
|
|
|
|
|
|
@when("I create a PlanGenerationGraph with a mock FakeListLLM")
|
|
def step_create_plan_gen_with_fake(context: Context) -> None:
|
|
from langchain_community.llms import FakeListLLM
|
|
|
|
llm = FakeListLLM(
|
|
responses=[
|
|
"Requirements: test",
|
|
"Generated code",
|
|
"Validation passed",
|
|
]
|
|
)
|
|
context.plan_graph = context.PlanGenerationGraph(llm=llm)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then — agent creation assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("a provider-not-configured error should be raised")
|
|
def step_assert_provider_error_raised(context: Context) -> None:
|
|
from cleveragents.core.exceptions import ProviderNotConfiguredError
|
|
|
|
assert context.raised_error is not None, (
|
|
"Expected ProviderNotConfiguredError but none was raised"
|
|
)
|
|
assert isinstance(context.raised_error, ProviderNotConfiguredError)
|
|
|
|
|
|
@then("the raised error should mention explicit LLM")
|
|
def step_assert_error_mentions_explicit_llm(context: Context) -> None:
|
|
msg = str(context.raised_error)
|
|
assert "explicit LLM" in msg or "LLM instance" in msg, (
|
|
f"Error message does not mention explicit LLM: {msg}"
|
|
)
|
|
|
|
|
|
@then("the auto-debug agent should be created successfully")
|
|
def step_assert_auto_debug_created(context: Context) -> None:
|
|
assert context.agent is not None
|
|
|
|
|
|
@then("the context analysis agent should be created successfully")
|
|
def step_assert_context_agent_created(context: Context) -> None:
|
|
assert context.context_agent is not None
|
|
|
|
|
|
@then("the plan generation graph should be created successfully")
|
|
def step_assert_plan_graph_created(context: Context) -> None:
|
|
assert context.plan_graph is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Exception hierarchy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("ProviderNotConfiguredError should be a subclass of ProviderError")
|
|
def step_assert_subclass(context: Context) -> None:
|
|
assert issubclass(context.ProviderNotConfiguredError, context.ProviderError)
|
|
|
|
|
|
@when('I create a ProviderNotConfiguredError with name "{name}"')
|
|
def step_create_error_with_name(context: Context, name: str) -> None:
|
|
context.raised_error = context.ProviderNotConfiguredError(
|
|
"test error", provider_name=name
|
|
)
|
|
|
|
|
|
@then('the error should have provider_name "{name}"')
|
|
def step_assert_provider_name(context: Context, name: str) -> None:
|
|
assert context.raised_error.provider_name == name
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry — resolve_provider_by_name
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I have a fresh ProviderRegistry for fix tests")
|
|
def step_have_fresh_registry(context: Context) -> None:
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.providers.registry import ProviderRegistry
|
|
|
|
Settings._instance = None
|
|
context.registry = ProviderRegistry()
|
|
|
|
|
|
@when("I call resolve_provider_by_name with empty string")
|
|
def step_resolve_empty(context: Context) -> None:
|
|
from cleveragents.core.exceptions import ProviderNotConfiguredError
|
|
|
|
context.raised_error = None
|
|
try:
|
|
context.registry.resolve_provider_by_name("")
|
|
except ProviderNotConfiguredError as exc:
|
|
context.raised_error = exc
|
|
|
|
|
|
@when('I call resolve_provider_by_name with "{name}"')
|
|
def step_resolve_by_name(context: Context, name: str) -> None:
|
|
from cleveragents.core.exceptions import ProviderNotConfiguredError
|
|
|
|
context.raised_error = None
|
|
try:
|
|
context.registry.resolve_provider_by_name(name)
|
|
except ProviderNotConfiguredError as exc:
|
|
context.raised_error = exc
|
|
|
|
|
|
@when('I call resolve_provider_by_name with unconfigured "{name}"')
|
|
def step_resolve_unconfigured(context: Context, name: str) -> None:
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.core.exceptions import ProviderNotConfiguredError
|
|
from cleveragents.providers.registry import ProviderRegistry
|
|
|
|
# Build a registry where no API keys are set so the provider is
|
|
# known but NOT configured.
|
|
env_keys = [
|
|
"OPENAI_API_KEY",
|
|
"ANTHROPIC_API_KEY",
|
|
"GOOGLE_API_KEY",
|
|
"AZURE_OPENAI_API_KEY",
|
|
"AZURE_API_KEY",
|
|
"OPENROUTER_API_KEY",
|
|
"GEMINI_API_KEY",
|
|
"COHERE_API_KEY",
|
|
"GROQ_API_KEY",
|
|
"TOGETHER_API_KEY",
|
|
]
|
|
saved: dict[str, str | None] = {}
|
|
for key in env_keys:
|
|
saved[key] = os.environ.pop(key, None)
|
|
|
|
context.raised_error = None
|
|
try:
|
|
Settings._instance = None
|
|
registry = ProviderRegistry()
|
|
registry.resolve_provider_by_name(name)
|
|
except ProviderNotConfiguredError as exc:
|
|
context.raised_error = exc
|
|
finally:
|
|
for key, val in saved.items():
|
|
if val is not None:
|
|
os.environ[key] = val
|
|
|
|
|
|
@then("the raised error should mention unknown provider")
|
|
def step_assert_unknown_provider(context: Context) -> None:
|
|
msg = str(context.raised_error)
|
|
assert "Unknown provider" in msg or "unknown" in msg.lower(), (
|
|
f"Error message does not mention unknown provider: {msg}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Provider selection trace logging
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I have a ProviderRegistry with openai key for fix tests")
|
|
def step_registry_with_openai_key(context: Context) -> None:
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.providers.registry import ProviderRegistry
|
|
|
|
Settings._instance = None
|
|
env_patch = {"OPENAI_API_KEY": "test-key-12345"}
|
|
context._env_patcher = patch.dict(os.environ, env_patch)
|
|
context._env_patcher.start()
|
|
context.registry = ProviderRegistry()
|
|
|
|
|
|
@when("I call get_default_provider_type for trace logging")
|
|
def step_call_default_provider_type(context: Context) -> None:
|
|
with patch("cleveragents.providers.registry.logger") as mock_logger:
|
|
context.mock_logger = mock_logger
|
|
context.default_type = context.registry.get_default_provider_type()
|
|
|
|
|
|
@then("provider selection debug messages should be logged")
|
|
def step_assert_debug_logged(context: Context) -> None:
|
|
assert context.default_type is not None
|
|
context.mock_logger.debug.assert_called()
|
|
call_args = str(context.mock_logger.debug.call_args_list)
|
|
assert "Provider selected" in call_args or "auto-detected" in call_args, (
|
|
f"Expected provider selection log, got: {call_args}"
|
|
)
|
|
# Cleanup env patcher if still active
|
|
patcher = getattr(context, "_env_patcher", None)
|
|
if patcher is not None:
|
|
patcher.stop()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Container — get_ai_provider
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("no provider API keys are set for fix tests")
|
|
def step_no_api_keys_for_fix(context: Context) -> None:
|
|
from cleveragents.config.settings import Settings
|
|
|
|
Settings._instance = None
|
|
provider_keys = [
|
|
"OPENAI_API_KEY",
|
|
"ANTHROPIC_API_KEY",
|
|
"GOOGLE_API_KEY",
|
|
"AZURE_OPENAI_API_KEY",
|
|
"AZURE_API_KEY",
|
|
"OPENROUTER_API_KEY",
|
|
"GEMINI_API_KEY",
|
|
"COHERE_API_KEY",
|
|
"GROQ_API_KEY",
|
|
"TOGETHER_API_KEY",
|
|
"CLEVERAGENTS_DEFAULT_PROVIDER",
|
|
"CLEVERAGENTS_DEFAULT_MODEL",
|
|
]
|
|
context._saved_env = {}
|
|
for key in provider_keys:
|
|
context._saved_env[key] = os.environ.pop(key, None)
|
|
|
|
|
|
@given("mock AI mode is disabled for fix tests")
|
|
def step_mock_disabled_for_fix(context: Context) -> None:
|
|
context._saved_mock_ai = os.environ.get("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "false"
|
|
|
|
|
|
@when("I call container get_ai_provider")
|
|
def step_call_container_get_ai_provider(context: Context) -> None:
|
|
from cleveragents.application.container import get_ai_provider
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.core.exceptions import ProviderNotConfiguredError
|
|
from cleveragents.providers.registry import reset_provider_registry
|
|
|
|
Settings._instance = None
|
|
reset_provider_registry()
|
|
|
|
context.raised_error = None
|
|
context.provider_result = None
|
|
try:
|
|
context.provider_result = get_ai_provider()
|
|
except ProviderNotConfiguredError as exc:
|
|
context.raised_error = exc
|
|
finally:
|
|
# Restore saved env vars
|
|
saved = getattr(context, "_saved_env", {})
|
|
for key, val in saved.items():
|
|
if val is not None:
|
|
os.environ[key] = val
|
|
else:
|
|
os.environ.pop(key, None)
|
|
saved_mock = getattr(context, "_saved_mock_ai", None)
|
|
if saved_mock is not None:
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = saved_mock
|
|
else:
|
|
os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
|
|
|
|
@given("mock AI mode is enabled for fix tests")
|
|
def step_mock_enabled_for_fix(context: Context) -> None:
|
|
context._saved_mock_ai = os.environ.get("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
|
|
|
|
|
|
@then("a mock provider or None should be returned")
|
|
def step_assert_mock_or_none_returned(context: Context) -> None:
|
|
assert context.raised_error is None, (
|
|
f"Unexpected error in mock mode: {context.raised_error}"
|
|
)
|
|
result = context.provider_result
|
|
# MockAIProvider or None (if mock module not importable)
|
|
if result is not None:
|
|
assert hasattr(result, "generate_changes")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Settings — mock_providers flag
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I create a fresh Settings for fix tests")
|
|
def step_create_fresh_settings(context: Context) -> None:
|
|
from cleveragents.config.settings import Settings
|
|
|
|
Settings._instance = None
|
|
context.settings = Settings()
|
|
|
|
|
|
@then("the mock_providers setting should be False")
|
|
def step_assert_mock_providers_false(context: Context) -> None:
|
|
assert context.settings.mock_providers is False
|
|
|
|
|
|
@given("mock_providers env is set to true")
|
|
def step_set_mock_providers_env(context: Context) -> None:
|
|
context._saved_mock_providers = os.environ.get("CLEVERAGENTS_MOCK_PROVIDERS")
|
|
os.environ["CLEVERAGENTS_MOCK_PROVIDERS"] = "true"
|
|
|
|
|
|
@given("CLEVERAGENTS_TESTING_USE_MOCK_AI is cleared")
|
|
def step_clear_mock_ai_env(context: Context) -> None:
|
|
context._saved_testing_mock = os.environ.get("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
|
|
|
|
@when("I create a Settings with mock_providers for fix tests")
|
|
def step_create_settings_mock_providers(context: Context) -> None:
|
|
from cleveragents.config.settings import Settings
|
|
|
|
Settings._instance = None
|
|
context.settings = Settings()
|
|
|
|
# Restore env vars
|
|
saved_mp = getattr(context, "_saved_mock_providers", None)
|
|
if saved_mp is not None:
|
|
os.environ["CLEVERAGENTS_MOCK_PROVIDERS"] = saved_mp
|
|
else:
|
|
os.environ.pop("CLEVERAGENTS_MOCK_PROVIDERS", None)
|
|
saved_tm = getattr(context, "_saved_testing_mock", None)
|
|
if saved_tm is not None:
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = saved_tm
|
|
|
|
|
|
@then("the mock_providers setting should be True")
|
|
def step_assert_mock_providers_true(context: Context) -> None:
|
|
assert context.settings.mock_providers is True
|