CoreRasurae
cdbe504b2c
CI / push-validation (pull_request) Successful in 36s
CI / helm (pull_request) Successful in 44s
CI / build (pull_request) Successful in 48s
CI / lint (pull_request) Successful in 1m10s
CI / typecheck (pull_request) Successful in 1m21s
CI / security (pull_request) Successful in 1m20s
CI / quality (pull_request) Successful in 1m26s
CI / integration_tests (pull_request) Successful in 7m53s
CI / unit_tests (pull_request) Failing after 9m56s
CI / coverage (pull_request) Has been skipped
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 4s
Replace sqlite:///:memory: with sqlite:///file::memory:?cache=shared to ensure all SQLAlchemy connections within the same process share a single in-memory database. This prevents isolation issues where SQLChatMessageHistory and EntityStore would each create separate in-memory databases, avoiding unexpected errors in CI parallel execution.
3546 lines
126 KiB
Python
3546 lines
126 KiB
Python
"""Step definitions for plan service coverage tests."""
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import os
|
|
import tempfile
|
|
from collections.abc import Callable, Iterator
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any, cast
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from sqlalchemy import create_engine
|
|
|
|
from cleveragents.application.services.memory_service import (
|
|
ConversationBufferMemoryAdapter,
|
|
MemoryService,
|
|
)
|
|
from cleveragents.application.services.plan_service import PlanService
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.core.exceptions import PlanError, ValidationError
|
|
from cleveragents.domain.models.core import (
|
|
Actor,
|
|
Change,
|
|
OperationType,
|
|
Plan,
|
|
PlanStatus,
|
|
Project,
|
|
ProjectSettings,
|
|
)
|
|
from cleveragents.domain.models.core import (
|
|
Context as PlanContext,
|
|
)
|
|
from cleveragents.domain.providers.ai_provider import (
|
|
ActorInvocationContext,
|
|
ProviderResponse,
|
|
)
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
from cleveragents.providers.registry import ProviderType
|
|
|
|
|
|
class _NoProviderRegistry:
|
|
"""Stub provider registry that always raises."""
|
|
|
|
def create_ai_provider(self, *args: Any, **kwargs: Any) -> None:
|
|
raise ValueError("No AI provider configured")
|
|
|
|
|
|
@given("I have a temporary test directory for plan service")
|
|
def step_create_temp_dir_plan_service(context: Context) -> None:
|
|
"""Create a temporary directory for testing."""
|
|
context.temp_dir = Path(tempfile.mkdtemp())
|
|
|
|
|
|
@given("I have a Unit of Work instance for plan testing")
|
|
def step_create_unit_of_work_plan(context: Context) -> None:
|
|
"""Create a Unit of Work instance for testing."""
|
|
from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
|
|
from cleveragents.infrastructure.database.models import Base
|
|
|
|
# Create a unique in-memory database URL for this test
|
|
database_url = "sqlite:///:memory:"
|
|
|
|
# Create a shared engine for in-memory database
|
|
if database_url not in MEMORY_ENGINES:
|
|
engine = create_engine(database_url, connect_args={"check_same_thread": False})
|
|
MEMORY_ENGINES[database_url] = engine
|
|
else:
|
|
engine = MEMORY_ENGINES[database_url]
|
|
|
|
# Create schema directly — much faster than 25 Alembic migrations
|
|
Base.metadata.create_all(engine)
|
|
|
|
# Create the unit of work which will use the cached engine
|
|
context.unit_of_work = UnitOfWork(database_url=database_url)
|
|
|
|
|
|
@given("I have a PlanService instance")
|
|
def step_create_plan_service(context: Context) -> None:
|
|
"""Create a PlanService instance."""
|
|
os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
settings = Settings()
|
|
context.plan_service = PlanService(
|
|
settings=settings, unit_of_work=context.unit_of_work, ai_provider=None
|
|
)
|
|
|
|
|
|
@given("I have a lightweight plan service for actor testing")
|
|
def step_create_lightweight_plan_service(context: Context) -> None:
|
|
"""Create a lightweight PlanService using in-memory DB for actor resolution tests.
|
|
|
|
This avoids the overhead of file-based DB creation, temp directories, and
|
|
project initialization — none of which are needed for testing actor/provider
|
|
resolution logic.
|
|
"""
|
|
if not hasattr(context, "unit_of_work"):
|
|
step_create_unit_of_work_plan(context)
|
|
if not hasattr(context, "temp_dir"):
|
|
context.temp_dir = Path(tempfile.mkdtemp(prefix="plan_actor_"))
|
|
settings = Settings()
|
|
context.plan_service = PlanService(
|
|
settings=settings, unit_of_work=context.unit_of_work, ai_provider=None
|
|
)
|
|
|
|
|
|
@given("I have a plan service with stub provider registry")
|
|
def step_plan_service_with_stub_registry(context: Context) -> None:
|
|
"""Create a PlanService that uses a stub provider registry for overrides."""
|
|
|
|
os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
|
|
if not hasattr(context, "unit_of_work"):
|
|
step_create_unit_of_work_plan(context)
|
|
if not hasattr(context, "temp_dir"):
|
|
context.temp_dir = Path(tempfile.mkdtemp(prefix="plan_service_stub_"))
|
|
|
|
settings = Settings()
|
|
context.stub_provider_requests = cast(list[dict[str, Any]], [])
|
|
|
|
class _StubProvider:
|
|
def __init__(self, provider_name: str, model_name: str) -> None:
|
|
self.name = provider_name
|
|
self.model_id = model_name
|
|
|
|
def generate_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> ProviderResponse:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=plan.id or 0,
|
|
file_path="stub/generated_file.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# generated by stub\n",
|
|
new_path=None,
|
|
applied=False,
|
|
created_at=datetime.now(),
|
|
applied_at=None,
|
|
)
|
|
return ProviderResponse(
|
|
changes=[change],
|
|
model_used=self.model_id,
|
|
token_count=42,
|
|
error_message=None,
|
|
)
|
|
|
|
class _StubRegistry:
|
|
def create_ai_provider(
|
|
self,
|
|
provider_type: str | ProviderType | None = None,
|
|
model_id: str | None = None,
|
|
max_retries: int = 3,
|
|
) -> _StubProvider:
|
|
if isinstance(provider_type, ProviderType):
|
|
provider_name = provider_type.value
|
|
else:
|
|
provider_name = (provider_type or "openai").lower()
|
|
model_name = model_id or "gpt-4o"
|
|
context.stub_provider_requests.append(
|
|
{"provider": provider_name, "model": model_name}
|
|
)
|
|
return _StubProvider(provider_name, model_name)
|
|
|
|
context.stub_registry = _StubRegistry()
|
|
context.plan_service = PlanService(
|
|
settings=settings,
|
|
unit_of_work=context.unit_of_work,
|
|
ai_provider=None,
|
|
provider_registry=context.stub_registry,
|
|
)
|
|
|
|
project = Project(
|
|
id=None,
|
|
name="stub-project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="stub-gpt",
|
|
),
|
|
)
|
|
with context.unit_of_work.transaction() as ctx:
|
|
context.project = ctx.projects.create(project)
|
|
context.test_project = context.project
|
|
|
|
|
|
@given("I have a plan service with a streaming stub provider")
|
|
def step_plan_service_streaming_stub(context: Context) -> None:
|
|
"""Create a PlanService that streams provider events for testing."""
|
|
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
|
|
|
|
if not hasattr(context, "unit_of_work"):
|
|
step_create_unit_of_work_plan(context)
|
|
if not hasattr(context, "temp_dir"):
|
|
context.temp_dir = Path(tempfile.mkdtemp(prefix="plan_service_streaming_"))
|
|
|
|
settings = Settings()
|
|
|
|
class _StreamingStubProvider:
|
|
def __init__(self) -> None:
|
|
self.name = "stub-stream"
|
|
self.model_id = "stub-model"
|
|
self.token_count = 321
|
|
|
|
def _build_change(self, plan: Plan) -> Change:
|
|
return Change(
|
|
id=None,
|
|
plan_id=plan.id or 0,
|
|
file_path="stream/generated_file.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# streamed change\n",
|
|
new_path=None,
|
|
applied=False,
|
|
created_at=datetime.now(),
|
|
applied_at=None,
|
|
)
|
|
|
|
def generate_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> ProviderResponse:
|
|
change = self._build_change(plan)
|
|
return ProviderResponse(
|
|
changes=[change],
|
|
model_used=self.model_id,
|
|
token_count=self.token_count,
|
|
error_message=None,
|
|
)
|
|
|
|
def stream_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> Iterator[dict[str, object]]:
|
|
response = self.generate_changes(
|
|
project,
|
|
plan,
|
|
contexts,
|
|
actor_context=actor_context,
|
|
progress_callback=progress_callback,
|
|
)
|
|
for node in (
|
|
"load_context",
|
|
"analyze_requirements",
|
|
"generate_plan",
|
|
"validate",
|
|
):
|
|
payload: dict[str, object] = {}
|
|
if node == "generate_plan":
|
|
payload = {
|
|
"status": "completed",
|
|
"change_count": len(response.changes),
|
|
}
|
|
elif node == "validate":
|
|
payload = {"status": "completed"}
|
|
yield {node: payload}
|
|
yield {"__end__": {"response": response}}
|
|
|
|
context.streaming_stub_provider = _StreamingStubProvider()
|
|
context.plan_service = PlanService(
|
|
settings=settings,
|
|
unit_of_work=context.unit_of_work,
|
|
ai_provider=context.streaming_stub_provider,
|
|
)
|
|
|
|
project = Project(
|
|
id=None,
|
|
name="streaming-project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="stub-stream",
|
|
),
|
|
)
|
|
with context.unit_of_work.transaction() as ctx:
|
|
context.test_project = ctx.projects.create(project)
|
|
|
|
|
|
def _setup_custom_streaming_plan_service(
|
|
context: Context,
|
|
provider_factory: Callable[[], object],
|
|
*,
|
|
project_name: str,
|
|
) -> None:
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
|
|
if not hasattr(context, "unit_of_work"):
|
|
step_create_unit_of_work_plan(context)
|
|
if not hasattr(context, "temp_dir"):
|
|
context.temp_dir = Path(
|
|
tempfile.mkdtemp(prefix="plan_service_streaming_custom_")
|
|
)
|
|
settings = Settings()
|
|
provider = provider_factory()
|
|
context.plan_service = PlanService(
|
|
settings=settings,
|
|
unit_of_work=context.unit_of_work,
|
|
ai_provider=provider,
|
|
)
|
|
project = Project(
|
|
id=None,
|
|
name=project_name,
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model=getattr(provider, "model_id", "custom-stream"),
|
|
),
|
|
)
|
|
with context.unit_of_work.transaction() as ctx:
|
|
context.test_project = ctx.projects.create(project)
|
|
|
|
|
|
@given("I have a plan service with an incomplete streaming provider")
|
|
def step_streaming_provider_incomplete(context: Context) -> None:
|
|
class _IncompleteStreamProvider:
|
|
def __init__(self) -> None:
|
|
self.name = "incomplete-stream"
|
|
self.model_id = "stub-incomplete"
|
|
|
|
def generate_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> ProviderResponse:
|
|
return ProviderResponse(
|
|
changes=[],
|
|
model_used=self.model_id,
|
|
token_count=0,
|
|
error_message=None,
|
|
)
|
|
|
|
def stream_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> Iterator[object]:
|
|
yield "corrupted-event"
|
|
|
|
_setup_custom_streaming_plan_service(
|
|
context,
|
|
lambda: _IncompleteStreamProvider(),
|
|
project_name="incomplete-streaming-project",
|
|
)
|
|
|
|
|
|
@given("I have a plan service with a dict streaming provider")
|
|
def step_streaming_provider_dict(context: Context) -> None:
|
|
class _DictPayloadStreamProvider:
|
|
def __init__(self) -> None:
|
|
self.name = "dict-stream"
|
|
self.model_id = "dict-model"
|
|
|
|
def stream_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> Iterator[dict[str, object]]:
|
|
yield {"generate_plan": {"status": "running"}}
|
|
payload = {
|
|
"changes": [
|
|
{
|
|
"plan_id": plan.id or 0,
|
|
"file_path": "dict_payload.py",
|
|
"operation": "create",
|
|
"original_content": None,
|
|
"new_content": "print('dict payload')",
|
|
"applied": False,
|
|
}
|
|
],
|
|
"model_used": self.model_id,
|
|
"token_count": "not-a-number",
|
|
}
|
|
yield {"__end__": payload}
|
|
|
|
_setup_custom_streaming_plan_service(
|
|
context,
|
|
lambda: _DictPayloadStreamProvider(),
|
|
project_name="dict-streaming-project",
|
|
)
|
|
|
|
|
|
@given("I have a plan service with a string error streaming provider")
|
|
def step_streaming_provider_string_error(context: Context) -> None:
|
|
class _StringErrorStreamProvider:
|
|
def __init__(self) -> None:
|
|
self.name = "string-error-stream"
|
|
self.model_id = "string-model"
|
|
|
|
def stream_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> Iterator[dict[str, object]]:
|
|
yield {"__end__": "stub failure"}
|
|
|
|
_setup_custom_streaming_plan_service(
|
|
context,
|
|
lambda: _StringErrorStreamProvider(),
|
|
project_name="string-error-project",
|
|
)
|
|
|
|
|
|
@given("I have a plan service with a non-list change streaming provider")
|
|
def step_streaming_provider_non_list(context: Context) -> None:
|
|
class _NonListChangeStreamProvider:
|
|
def __init__(self) -> None:
|
|
self.name = "non-list-stream"
|
|
self.model_id = "non-list-model"
|
|
|
|
def stream_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> Iterator[dict[str, object]]:
|
|
payload = {
|
|
"changes": {"file_path": "bogus.py"},
|
|
"model_used": self.model_id,
|
|
"token_count": 1,
|
|
}
|
|
yield {"__end__": payload}
|
|
|
|
_setup_custom_streaming_plan_service(
|
|
context,
|
|
lambda: _NonListChangeStreamProvider(),
|
|
project_name="non-list-stream-project",
|
|
)
|
|
|
|
|
|
@given("I have a plan service with an invalid change entry streaming provider")
|
|
def step_streaming_provider_invalid_entry(context: Context) -> None:
|
|
class _InvalidEntryStreamProvider:
|
|
def __init__(self) -> None:
|
|
self.name = "invalid-entry-stream"
|
|
self.model_id = "invalid-entry-model"
|
|
|
|
def stream_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> Iterator[dict[str, object]]:
|
|
payload = {
|
|
"changes": [123],
|
|
"model_used": self.model_id,
|
|
"token_count": 1,
|
|
}
|
|
yield {"__end__": payload}
|
|
|
|
_setup_custom_streaming_plan_service(
|
|
context,
|
|
lambda: _InvalidEntryStreamProvider(),
|
|
project_name="invalid-entry-stream-project",
|
|
)
|
|
|
|
|
|
@given("I have a plan service with a non-python streaming provider")
|
|
def step_streaming_provider_non_python(context: Context) -> None:
|
|
class _NonPythonStreamProvider:
|
|
def __init__(self) -> None:
|
|
self.name = "non-python-stream"
|
|
self.model_id = "non-python-model"
|
|
|
|
def stream_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> Iterator[dict[str, object]]:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=plan.id or 0,
|
|
file_path="notes.txt",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="plain text change",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
response = ProviderResponse(
|
|
changes=[change],
|
|
model_used=self.model_id,
|
|
token_count=5,
|
|
error_message=None,
|
|
)
|
|
yield {"__end__": response}
|
|
|
|
_setup_custom_streaming_plan_service(
|
|
context,
|
|
lambda: _NonPythonStreamProvider(),
|
|
project_name="non-python-stream-project",
|
|
)
|
|
|
|
|
|
@given("I have a plan service with a syntax error streaming provider")
|
|
def step_streaming_provider_syntax_error(context: Context) -> None:
|
|
class _SyntaxErrorStreamProvider:
|
|
def __init__(self) -> None:
|
|
self.name = "syntax-error-stream"
|
|
self.model_id = "syntax-model"
|
|
|
|
def stream_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> Iterator[dict[str, object]]:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=plan.id or 0,
|
|
file_path="bad.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="def broken(",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
response = ProviderResponse(
|
|
changes=[change],
|
|
model_used=self.model_id,
|
|
token_count=0,
|
|
error_message=None,
|
|
)
|
|
yield {"__end__": response}
|
|
|
|
_setup_custom_streaming_plan_service(
|
|
context,
|
|
lambda: _SyntaxErrorStreamProvider(),
|
|
project_name="syntax-error-stream-project",
|
|
)
|
|
|
|
|
|
@given("mock provider mode is forced")
|
|
def step_force_mock_provider_mode(context: Context) -> None:
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
|
|
|
|
if not hasattr(context, "env_vars_to_clean"):
|
|
context.env_vars_to_clean = []
|
|
if "CLEVERAGENTS_TESTING_USE_MOCK_AI" not in context.env_vars_to_clean:
|
|
context.env_vars_to_clean.append("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
if hasattr(context, "plan_service"):
|
|
context.plan_service.ai_provider = None
|
|
context.provider_resolution = None
|
|
|
|
|
|
@given("the plan service uses a nameless AI provider")
|
|
def step_set_nameless_ai_provider(context: Context) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
class _NamelessProvider:
|
|
pass
|
|
|
|
context.plan_service.ai_provider = _NamelessProvider()
|
|
|
|
|
|
@given("I stub the provider registry factory for lazy initialization")
|
|
def step_stub_provider_registry_lazy(context: Context) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
# Ensure mock-provider forcing doesn't short-circuit registry creation
|
|
original_mock_flag = os.environ.get("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "false"
|
|
|
|
# Force registry path by clearing any injected provider
|
|
context.plan_service.ai_provider = None
|
|
|
|
class _LazyProvider:
|
|
def __init__(self, provider_name: str, model_name: str) -> None:
|
|
self.name = provider_name
|
|
self.model_id = model_name
|
|
|
|
class _LazyRegistry:
|
|
def __init__(self) -> None:
|
|
self.requests: list[dict[str, str]] = []
|
|
|
|
def create_ai_provider(
|
|
self,
|
|
provider_type: str | ProviderType | None = None,
|
|
model_id: str | None = None,
|
|
max_retries: int = 3,
|
|
) -> _LazyProvider:
|
|
if isinstance(provider_type, ProviderType):
|
|
provider_name = provider_type.value
|
|
else:
|
|
provider_name = (provider_type or "openai").lower()
|
|
model_name = (model_id or "gpt-4o").lower()
|
|
self.requests.append({"provider": provider_name, "model": model_name})
|
|
return _LazyProvider(provider_name, model_name)
|
|
|
|
lazy_registry = _LazyRegistry()
|
|
patcher = patch(
|
|
"cleveragents.application.services.plan_service.get_provider_registry",
|
|
autospec=True,
|
|
return_value=lazy_registry,
|
|
)
|
|
patcher.start()
|
|
cleanup_handlers = getattr(context, "_cleanup_handlers", None)
|
|
if cleanup_handlers is None:
|
|
cleanup_handlers = []
|
|
context._cleanup_handlers = cleanup_handlers
|
|
cleanup_handlers.append(patcher.stop)
|
|
|
|
# Restore mock flag after scenario
|
|
cleanup_handlers.append(
|
|
lambda: (
|
|
os.environ.__setitem__(
|
|
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
|
|
)
|
|
if original_mock_flag is not None
|
|
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
)
|
|
)
|
|
|
|
context.lazy_registry = lazy_registry
|
|
context.plan_service._provider_registry = None
|
|
|
|
|
|
@then('the lazy registry should memoize provider "{provider}" and model "{model}"')
|
|
def step_assert_lazy_registry(context: Context, provider: str, model: str) -> None:
|
|
registry = getattr(context, "lazy_registry", None)
|
|
assert registry is not None, "Lazy registry stub was not configured"
|
|
expected_provider = provider.lower()
|
|
expected_model = model.lower()
|
|
assert registry.requests == [
|
|
{"provider": expected_provider, "model": expected_model}
|
|
], "Lazy registry did not capture the override request"
|
|
assert context.plan_service._provider_registry is registry
|
|
resolution = getattr(context, "provider_resolution", None)
|
|
assert resolution is not None, "Provider resolution result missing"
|
|
provider_instance, resolved_name, resolved_model = resolution[:3]
|
|
assert resolved_name == expected_provider
|
|
assert resolved_model == model
|
|
assert provider_instance.name == expected_provider
|
|
assert provider_instance.model_id == expected_model
|
|
|
|
|
|
@given("the plan service has a cached provider registry instance")
|
|
def step_cached_provider_registry(context: Context) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
cached_registry = SimpleNamespace(name="cached-registry")
|
|
context.plan_service._provider_registry = cached_registry
|
|
context.cached_registry = cached_registry
|
|
|
|
|
|
@when("I fetch the provider registry from the plan service")
|
|
def step_fetch_cached_provider_registry(context: Context) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
try:
|
|
context.registry_result = context.plan_service._get_provider_registry()
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.registry_result = None
|
|
context.exception = exc
|
|
|
|
|
|
@then("the cached provider registry should be reused")
|
|
def step_assert_cached_provider_registry(context: Context) -> None:
|
|
cached = getattr(context, "cached_registry", None)
|
|
result = getattr(context, "registry_result", None)
|
|
assert cached is not None, "Cached registry was not configured"
|
|
assert result is cached
|
|
assert getattr(context, "exception", None) is None
|
|
|
|
|
|
@given('the plan service actor lookup raises ValidationError "{message}"')
|
|
def step_actor_lookup_validation_error(context: Context, message: str) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
class _ActorServiceStub:
|
|
def get_actor(self, actor_name: str) -> None:
|
|
raise ValidationError(message)
|
|
|
|
def get_default_actor(self) -> None:
|
|
raise ValidationError(message)
|
|
|
|
def ensure_default_mock_actor(self) -> None:
|
|
raise ValidationError(message)
|
|
|
|
context.plan_service.actor_service = _ActorServiceStub()
|
|
|
|
|
|
@given("the plan service actor lookup returns no actor and mock provisioning fails")
|
|
def step_actor_lookup_returns_none(context: Context) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
class _ActorServiceStub:
|
|
def get_actor(self, actor_name: str | None) -> None:
|
|
return None
|
|
|
|
def get_default_actor(self) -> None:
|
|
return None
|
|
|
|
def ensure_default_mock_actor(self) -> None:
|
|
raise RuntimeError("no mock actor available")
|
|
|
|
context.plan_service.actor_service = _ActorServiceStub()
|
|
|
|
|
|
@given(
|
|
'the plan service actor lookup returns actor "{actor_name}" with provider "{provider}" and model "{model}"'
|
|
)
|
|
def step_actor_lookup_returns_actor(
|
|
context: Context, actor_name: str, provider: str, model: str
|
|
) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
actor = Actor(
|
|
id=None,
|
|
name=actor_name,
|
|
provider=provider,
|
|
model=model,
|
|
config_blob={},
|
|
config_hash=Actor.compute_hash({}),
|
|
graph_descriptor=None,
|
|
unsafe=False,
|
|
is_built_in=False,
|
|
is_default=False,
|
|
)
|
|
|
|
class _ActorServiceStub:
|
|
def get_actor(self, name: str) -> Actor:
|
|
return actor
|
|
|
|
def get_default_actor(self) -> Actor:
|
|
return actor
|
|
|
|
def ensure_default_mock_actor(self) -> Actor:
|
|
return actor
|
|
|
|
context.plan_service.actor_service = _ActorServiceStub()
|
|
|
|
|
|
@given(
|
|
'the plan service actor lookup returns actor "{actor_name}" with provider "{provider}" and no model'
|
|
)
|
|
def step_actor_lookup_returns_actor_without_model(
|
|
context: Context, actor_name: str, provider: str
|
|
) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
actor = Actor(
|
|
id=None,
|
|
name=actor_name,
|
|
provider=provider,
|
|
model="placeholder",
|
|
config_blob={},
|
|
config_hash=Actor.compute_hash({}),
|
|
graph_descriptor=None,
|
|
unsafe=False,
|
|
is_built_in=False,
|
|
is_default=False,
|
|
)
|
|
object.__setattr__(actor, "model", None)
|
|
|
|
class _ActorServiceStub:
|
|
def get_actor(self, name: str) -> Actor:
|
|
return actor
|
|
|
|
def get_default_actor(self) -> Actor:
|
|
return actor
|
|
|
|
def ensure_default_mock_actor(self) -> Actor:
|
|
return actor
|
|
|
|
context.plan_service.actor_service = _ActorServiceStub()
|
|
|
|
|
|
@when('I try to resolve actor "{actor_name}"')
|
|
def step_try_resolve_actor(context: Context, actor_name: str) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
if actor_name == "ghost/actor":
|
|
context.plan_service.actor_service = None
|
|
try:
|
|
context.actor_resolution = context.plan_service._resolve_actor(actor_name)
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.actor_resolution = None
|
|
context.exception = exc
|
|
|
|
|
|
@then('the actor resolution should return actor "{actor_name}" from source "{source}"')
|
|
def step_assert_actor_resolution(
|
|
context: Context, actor_name: str, source: str
|
|
) -> None:
|
|
resolution = getattr(context, "actor_resolution", None)
|
|
assert resolution is not None, "Actor resolution result missing"
|
|
actor, actor_source = resolution
|
|
assert actor.name == actor_name
|
|
assert actor_source == source
|
|
|
|
|
|
@then('the PlanError should include actor detail "{actor_name}"')
|
|
def step_assert_plan_error_actor_detail(context: Context, actor_name: str) -> None:
|
|
exc = getattr(context, "exception", None)
|
|
assert isinstance(exc, PlanError), "Expected PlanError to be raised"
|
|
details = getattr(exc, "details", {}) or {}
|
|
assert details.get("actor", "") == actor_name
|
|
|
|
|
|
@given('the plan service AI provider is "{provider_name}" with model "{model_name}"')
|
|
def step_set_plan_service_ai_provider(
|
|
context: Context, provider_name: str, model_name: str
|
|
) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
class _Provider:
|
|
def __init__(self, name: str, model_id: str) -> None:
|
|
self.name = name
|
|
self.model_id = model_id
|
|
|
|
context.plan_service.ai_provider = _Provider(provider_name, model_name)
|
|
|
|
|
|
@given('the plan service AI provider has name "{provider_name}" without model id')
|
|
def step_set_plan_service_ai_provider_name_only(
|
|
context: Context, provider_name: str
|
|
) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
class _ProviderNameOnly:
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
|
|
context.plan_service.ai_provider = _ProviderNameOnly(provider_name)
|
|
|
|
|
|
@given("the plan service AI provider has no identifier attributes")
|
|
def step_set_anonymous_ai_provider(context: Context) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
|
|
if not hasattr(context, "env_vars_to_clean"):
|
|
context.env_vars_to_clean = []
|
|
if "CLEVERAGENTS_TESTING_USE_MOCK_AI" not in context.env_vars_to_clean:
|
|
context.env_vars_to_clean.append("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
|
|
class _AnonymousProvider:
|
|
"""Provider stub without name/model_id attributes."""
|
|
|
|
pass
|
|
|
|
context.plan_service.ai_provider = _AnonymousProvider()
|
|
context.provider_resolution = None
|
|
|
|
|
|
@given("the plan service uses a delayed mock provider")
|
|
def step_set_delayed_mock_provider(context: Context) -> None:
|
|
if not hasattr(context, "unit_of_work"):
|
|
step_create_unit_of_work_plan(context)
|
|
|
|
settings = Settings()
|
|
|
|
class _DelayedProvider:
|
|
def __init__(self) -> None:
|
|
self.name = "delayed-provider"
|
|
self.model_id = "delayed-model"
|
|
|
|
class _DelayedPlanService(PlanService):
|
|
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
|
super().__init__(*args, **kwargs)
|
|
self._provider_toggle = False
|
|
self._late_provider = kwargs.get("ai_provider") or _DelayedProvider()
|
|
|
|
@property
|
|
def ai_provider(self): # type: ignore[override]
|
|
if not getattr(self, "_provider_toggle", False):
|
|
self._provider_toggle = True
|
|
return None
|
|
return self._late_provider
|
|
|
|
@ai_provider.setter
|
|
def ai_provider(self, value): # type: ignore[override]
|
|
self._late_provider = value
|
|
|
|
def _use_mock_provider(self) -> bool:
|
|
return True
|
|
|
|
context.plan_service = _DelayedPlanService(
|
|
settings=settings, unit_of_work=context.unit_of_work, ai_provider=None
|
|
)
|
|
|
|
|
|
@given("the plan service actor provider resolution will fail")
|
|
def step_force_actor_provider_resolution_failure(context: Context) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
context.plan_service.actor_service = SimpleNamespace(name="stub-actor-service")
|
|
patcher = patch.object(
|
|
context.plan_service,
|
|
"_resolve_ai_provider_for_actor",
|
|
side_effect=PlanError(message="actor resolution failed"),
|
|
)
|
|
patcher.start()
|
|
cleanup_handlers = getattr(context, "_cleanup_handlers", None)
|
|
if cleanup_handlers is None:
|
|
cleanup_handlers = []
|
|
context._cleanup_handlers = cleanup_handlers
|
|
cleanup_handlers.append(patcher.stop)
|
|
|
|
|
|
@when('I resolve the AI provider for actor "{actor_name}"')
|
|
def step_resolve_ai_provider_for_actor(context: Context, actor_name: str) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
try:
|
|
context.provider_resolution = (
|
|
context.plan_service._resolve_ai_provider_for_actor(actor_name)
|
|
)
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.provider_resolution = None
|
|
context.exception = exc
|
|
|
|
|
|
@then("the provider resolution should use fallback names")
|
|
def step_assert_provider_resolution_fallback(context: Context) -> None:
|
|
resolution = getattr(context, "provider_resolution", None)
|
|
assert resolution is not None, "Provider resolution result missing"
|
|
provider_instance, provider_name, model_name, actor, *_ = resolution
|
|
assert provider_name == "mock-provider"
|
|
|
|
actor_model = getattr(actor, "model", None)
|
|
expected_model = actor_model or provider_name
|
|
assert model_name == expected_model
|
|
|
|
assert not hasattr(provider_instance, "name")
|
|
assert not hasattr(provider_instance, "model_id")
|
|
|
|
|
|
@then(
|
|
'the nameless provider resolution should return provider "{provider}" and model "{model}"'
|
|
)
|
|
def step_assert_nameless_provider_resolution(
|
|
context: Context, provider: str, model: str
|
|
) -> None:
|
|
resolution = getattr(context, "provider_resolution", None)
|
|
assert resolution is not None, "Provider resolution result missing"
|
|
provider_instance, provider_name, model_name = resolution[:3]
|
|
assert provider_name == provider
|
|
assert model_name == model
|
|
assert not hasattr(provider_instance, "name")
|
|
assert not hasattr(provider_instance, "model_id")
|
|
|
|
|
|
@when(
|
|
'I resolve provider for actor "{actor_name}" with model override "{model_override}"'
|
|
)
|
|
def step_resolve_provider_with_override(
|
|
context: Context, actor_name: str, model_override: str
|
|
) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
try:
|
|
context.provider_resolution = (
|
|
context.plan_service._resolve_ai_provider_for_actor(
|
|
actor_name, model=model_override
|
|
)
|
|
)
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.provider_resolution = None
|
|
context.exception = exc
|
|
|
|
|
|
@when('I resolve provider for actor "{actor_name}"')
|
|
def step_resolve_provider_for_actor(context: Context, actor_name: str) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
try:
|
|
context.provider_resolution = (
|
|
context.plan_service._resolve_ai_provider_for_actor(actor_name)
|
|
)
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.provider_resolution = None
|
|
context.exception = exc
|
|
|
|
|
|
@when("I try to resolve a provider for the plan service")
|
|
def step_resolve_provider_for_plan_service(context: Context) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
try:
|
|
context.provider_resolution = (
|
|
context.plan_service._resolve_ai_provider_for_actor(None)
|
|
)
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.provider_resolution = None
|
|
context.exception = exc
|
|
|
|
|
|
@then('the provider resolution should return provider "{provider}" and model "{model}"')
|
|
def step_assert_provider_resolution(
|
|
context: Context, provider: str, model: str
|
|
) -> None:
|
|
resolution = getattr(context, "provider_resolution", None)
|
|
assert resolution is not None, "Provider resolution result missing"
|
|
provider_instance, provider_name, model_name = resolution[:3]
|
|
assert provider_name == provider
|
|
assert model_name == model
|
|
assert getattr(provider_instance, "name", provider_name) == provider_name
|
|
assert getattr(provider_instance, "model_id", model_name) == model_name
|
|
|
|
|
|
@then('the lazy registry should record actor provider "{provider}" and model "{model}"')
|
|
def step_assert_lazy_registry_actor(
|
|
context: Context, provider: str, model: str
|
|
) -> None:
|
|
registry = getattr(context, "lazy_registry", None)
|
|
assert registry is not None, "Lazy registry stub was not configured"
|
|
expected = {"provider": provider.lower(), "model": model.lower()}
|
|
assert registry.requests == [expected], (
|
|
"Lazy registry did not capture the actor request"
|
|
)
|
|
|
|
resolution = getattr(context, "provider_resolution", None)
|
|
assert resolution is not None, "Provider resolution result missing"
|
|
provider_instance, provider_name, model_name, *_ = resolution
|
|
assert provider_name == expected["provider"]
|
|
assert model_name == model
|
|
assert getattr(provider_instance, "name", provider_name) == expected["provider"]
|
|
assert getattr(provider_instance, "model_id", model_name) == model
|
|
|
|
|
|
@given('the provider registry raises ValueError "{message}" for actor providers')
|
|
def step_provider_registry_raises_for_actor(context: Context, message: str) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
class _Registry:
|
|
def create_ai_provider(self, *args: Any, **kwargs: Any) -> None:
|
|
raise ValueError(message)
|
|
|
|
original_mock_flag = os.environ.get("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "false"
|
|
|
|
cleanup_handlers = getattr(context, "_cleanup_handlers", None)
|
|
if cleanup_handlers is None:
|
|
cleanup_handlers = []
|
|
context._cleanup_handlers = cleanup_handlers
|
|
cleanup_handlers.append(
|
|
lambda: (
|
|
os.environ.__setitem__(
|
|
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
|
|
)
|
|
if original_mock_flag is not None
|
|
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
)
|
|
)
|
|
|
|
context.plan_service._provider_registry = _Registry()
|
|
context.plan_service.ai_provider = None
|
|
|
|
|
|
@given(
|
|
'the provider registry factory raises ValueError "{message}" for actor providers'
|
|
)
|
|
def step_provider_registry_factory_raises_for_actor(
|
|
context: Context, message: str
|
|
) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
class _Registry:
|
|
def create_ai_provider(self, *args: Any, **kwargs: Any) -> None:
|
|
raise ValueError(message)
|
|
|
|
original_mock_flag = os.environ.get("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "false"
|
|
|
|
patcher = patch(
|
|
"cleveragents.application.services.plan_service.get_provider_registry",
|
|
autospec=True,
|
|
return_value=_Registry(),
|
|
)
|
|
patcher.start()
|
|
|
|
cleanup_handlers = getattr(context, "_cleanup_handlers", None)
|
|
if cleanup_handlers is None:
|
|
cleanup_handlers = []
|
|
context._cleanup_handlers = cleanup_handlers
|
|
cleanup_handlers.append(patcher.stop)
|
|
cleanup_handlers.append(
|
|
lambda: (
|
|
os.environ.__setitem__(
|
|
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
|
|
)
|
|
if original_mock_flag is not None
|
|
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
)
|
|
)
|
|
|
|
context.plan_service.ai_provider = None
|
|
context.plan_service._provider_registry = None
|
|
|
|
|
|
@given(
|
|
'the provider registry returns provider "{provider_name}" with model "{model_name}"'
|
|
)
|
|
def step_provider_registry_returns_provider(
|
|
context: Context, provider_name: str, model_name: str
|
|
) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
|
|
original_mock_flag = os.environ.get("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "false"
|
|
|
|
cleanup_handlers = getattr(context, "_cleanup_handlers", None)
|
|
if cleanup_handlers is None:
|
|
cleanup_handlers = []
|
|
context._cleanup_handlers = cleanup_handlers
|
|
cleanup_handlers.append(
|
|
lambda: (
|
|
os.environ.__setitem__(
|
|
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
|
|
)
|
|
if original_mock_flag is not None
|
|
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
)
|
|
)
|
|
|
|
class _Provider:
|
|
def __init__(self, name: str, model_id: str) -> None:
|
|
self.name = name
|
|
self.model_id = model_id
|
|
|
|
class _Registry:
|
|
def __init__(self) -> None:
|
|
self.requests: list[dict[str, Any]] = []
|
|
|
|
def create_ai_provider(
|
|
self,
|
|
provider_type: str | ProviderType | None = None,
|
|
model_id: str | None = None,
|
|
max_retries: int = 3,
|
|
) -> _Provider:
|
|
self.requests.append({"provider": provider_type, "model": model_id})
|
|
return _Provider(provider_name, model_name)
|
|
|
|
registry = _Registry()
|
|
context.plan_service.ai_provider = None
|
|
context.plan_service._provider_registry = registry
|
|
context.provider_registry_requests = registry.requests
|
|
|
|
|
|
@given("I have a plan service with an unrecoverable streaming provider")
|
|
def step_plan_service_unrecoverable_stream(context: Context) -> None:
|
|
class _UnrecoverableStreamProvider:
|
|
def __init__(self) -> None:
|
|
self.name = "fatal-stream"
|
|
self.model_id = "fatal-model"
|
|
|
|
def generate_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> ProviderResponse:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=plan.id or 0,
|
|
file_path="fatal.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="```python\nif True print('oops')\n```",
|
|
new_path=None,
|
|
applied=False,
|
|
created_at=datetime.now(),
|
|
applied_at=None,
|
|
)
|
|
return ProviderResponse(
|
|
changes=[change],
|
|
model_used=self.model_id,
|
|
token_count=0,
|
|
error_message=None,
|
|
)
|
|
|
|
def stream_changes(
|
|
self,
|
|
project: Project,
|
|
plan: Plan,
|
|
contexts: list[PlanContext],
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> Iterator[object]:
|
|
response = self.generate_changes(
|
|
project,
|
|
plan,
|
|
contexts,
|
|
actor_context=actor_context,
|
|
progress_callback=progress_callback,
|
|
)
|
|
yield {"generate_plan": {"status": "completed", "change_count": 1}}
|
|
yield {"__end__": {"response": response}}
|
|
|
|
_setup_custom_streaming_plan_service(
|
|
context,
|
|
lambda: _UnrecoverableStreamProvider(),
|
|
project_name="unrecoverable-stream-project",
|
|
)
|
|
|
|
patcher = patch(
|
|
"cleveragents.application.services.plan_service.ast.parse",
|
|
autospec=True,
|
|
side_effect=SyntaxError("invalid python"),
|
|
)
|
|
patcher.start()
|
|
cleanup_handlers = getattr(context, "_cleanup_handlers", None)
|
|
if cleanup_handlers is None:
|
|
cleanup_handlers = []
|
|
context._cleanup_handlers = cleanup_handlers
|
|
cleanup_handlers.append(patcher.stop)
|
|
|
|
|
|
@when('I stream plan generation with prompt "{prompt}"')
|
|
def step_stream_plan_generation(context: Context, prompt: str) -> None:
|
|
"""Run the streaming plan generation helper and capture events."""
|
|
|
|
async def _collect() -> list[dict[str, Any]]:
|
|
events: list[dict[str, Any]] = []
|
|
async for event in context.plan_service.generate_plan_streaming(
|
|
project=context.test_project,
|
|
description=prompt,
|
|
):
|
|
events.append(event)
|
|
return events
|
|
|
|
context.stream_events = asyncio.run(_collect())
|
|
context.partial_stream_events = context.stream_events
|
|
context.exception = None
|
|
|
|
|
|
@when('I try to stream plan generation with prompt "{prompt}"')
|
|
def step_try_stream_plan_generation(context: Context, prompt: str) -> None:
|
|
"""Capture streaming events even when the generator raises an error."""
|
|
|
|
async def _collect() -> list[dict[str, Any]]:
|
|
events: list[dict[str, Any]] = []
|
|
stream = context.plan_service.generate_plan_streaming(
|
|
project=context.test_project,
|
|
description=prompt,
|
|
)
|
|
try:
|
|
async for event in stream:
|
|
events.append(event)
|
|
except Exception:
|
|
context.partial_stream_events = events
|
|
raise
|
|
return events
|
|
|
|
try:
|
|
context.stream_events = asyncio.run(_collect())
|
|
context.partial_stream_events = context.stream_events
|
|
context.exception = None
|
|
except Exception as exc:
|
|
if not hasattr(context, "partial_stream_events"):
|
|
context.partial_stream_events = []
|
|
context.stream_events = context.partial_stream_events
|
|
context.exception = exc
|
|
|
|
|
|
@then('the streaming events should include nodes "{node_list}"')
|
|
def step_assert_stream_nodes(context: Context, node_list: str) -> None:
|
|
"""Verify that the streamed events contained the expected nodes."""
|
|
|
|
assert hasattr(context, "stream_events"), "Streaming events were not captured"
|
|
expected = [node.strip() for node in node_list.split(",") if node.strip()]
|
|
actual = [next(iter(event.keys())) for event in context.stream_events if event]
|
|
for node in expected:
|
|
assert node in actual, f"Expected streaming event '{node}' not found"
|
|
|
|
|
|
@then("the streamed plan should persist token count {expected:d}")
|
|
def step_assert_stream_token_count(context: Context, expected: int) -> None:
|
|
"""Ensure token counts from streaming providers are stored on the plan."""
|
|
|
|
with context.unit_of_work.transaction() as ctx:
|
|
plan = ctx.plans.get_current_for_project(context.test_project.id)
|
|
assert plan is not None, "Plan was not persisted"
|
|
assert plan.token_count == expected, (
|
|
f"Expected token count {expected}, got {plan.token_count}"
|
|
)
|
|
|
|
|
|
@then('the streaming failure events should include an error with message "{message}"')
|
|
def step_assert_stream_failure_event(context: Context, message: str) -> None:
|
|
"""Verify that streaming failures emit a descriptive error event."""
|
|
|
|
events = getattr(context, "partial_stream_events", [])
|
|
failure_payload: dict[str, Any] | None = None
|
|
for event in events:
|
|
if not isinstance(event, dict):
|
|
continue
|
|
payload = event.get("generate_plan")
|
|
if isinstance(payload, dict) and payload.get("status") == "failed":
|
|
failure_payload = payload
|
|
break
|
|
assert failure_payload is not None, "Expected a failure event in the stream"
|
|
assert failure_payload.get("error") == message, (
|
|
f"Expected failure message '{message}', got '{failure_payload.get('error')}'"
|
|
)
|
|
|
|
|
|
@then(
|
|
"the streaming result should include the non-python change without validation errors"
|
|
)
|
|
def step_assert_non_python_stream(context: Context) -> None:
|
|
"""Ensure non-Python changes stream without triggering validation errors."""
|
|
|
|
events = getattr(context, "stream_events", [])
|
|
assert events, "Streaming events were not captured"
|
|
end_event = events[-1]
|
|
assert isinstance(end_event, dict) and "__end__" in end_event, (
|
|
"Final streaming event did not include end payload"
|
|
)
|
|
payload = end_event["__end__"]
|
|
changes = payload.get("changes") or []
|
|
assert len(changes) == 1, "Expected exactly one streamed change"
|
|
change = changes[0]
|
|
assert change.file_path.endswith("notes.txt")
|
|
assert change.new_content == "plain text change"
|
|
|
|
|
|
@when('I build the plan with actor "{actor_name}"')
|
|
def step_build_plan_with_actor(context: Context, actor_name: str) -> None:
|
|
"""Build the current plan using the specified actor selection."""
|
|
if "/" in actor_name:
|
|
provider, model = actor_name.split("/", 1)
|
|
else:
|
|
provider, model = actor_name, "default-model"
|
|
|
|
actor_service = context.plan_service.actor_service
|
|
# Actors used in plan service tests must be local/ namespace since
|
|
# is_built_in was removed from upsert_actor (issue #10923).
|
|
actor_id = f"local/{actor_name.replace('/', '-')}"
|
|
actor_service.upsert_actor(
|
|
name=actor_id,
|
|
provider=provider,
|
|
model=model,
|
|
config_blob={"provider": provider, "model": model},
|
|
graph_descriptor=None,
|
|
unsafe=False,
|
|
set_default=True,
|
|
)
|
|
|
|
context.changes = context.plan_service.build_plan(
|
|
project=context.test_project,
|
|
actor=actor_id,
|
|
)
|
|
|
|
|
|
@then(
|
|
'the stub provider registry should record provider "{provider}" and model "{model}"'
|
|
)
|
|
def step_assert_stub_provider_requests(
|
|
context: Context, provider: str, model: str
|
|
) -> None:
|
|
"""Verify the stub registry saw the requested provider/model overrides."""
|
|
requests = getattr(context, "stub_provider_requests", None)
|
|
assert requests, "No provider override requests were recorded"
|
|
last_request = requests[-1]
|
|
assert last_request["provider"] == provider, (
|
|
f"Expected provider '{provider}', got '{last_request['provider']}'"
|
|
)
|
|
assert last_request["model"] == model, (
|
|
f"Expected model '{model}', got '{last_request['model']}'"
|
|
)
|
|
|
|
|
|
@given("I have a plan service without providers configured")
|
|
def step_plan_service_without_providers(context: Context) -> None:
|
|
"""Create a PlanService that will raise when resolving providers."""
|
|
|
|
original_mock_flag = os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
cleanup_handlers = getattr(context, "_cleanup_handlers", None)
|
|
if cleanup_handlers is None:
|
|
cleanup_handlers = []
|
|
context._cleanup_handlers = cleanup_handlers
|
|
cleanup_handlers.append(
|
|
lambda: (
|
|
os.environ.__setitem__(
|
|
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
|
|
)
|
|
if original_mock_flag is not None
|
|
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
)
|
|
)
|
|
|
|
if not hasattr(context, "unit_of_work"):
|
|
step_create_unit_of_work_plan(context)
|
|
if not hasattr(context, "temp_dir"):
|
|
context.temp_dir = Path(tempfile.mkdtemp(prefix="plan_service_no_provider_"))
|
|
|
|
settings = Settings()
|
|
context.plan_service = PlanService(
|
|
settings=settings,
|
|
unit_of_work=context.unit_of_work,
|
|
ai_provider=None,
|
|
provider_registry=_NoProviderRegistry(),
|
|
)
|
|
|
|
# Provision a default actor so provider resolution reaches the registry path
|
|
context.plan_service.actor_service.ensure_default_mock_actor(force=True)
|
|
|
|
project = Project(
|
|
id=None,
|
|
name="no-provider-project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="noop",
|
|
),
|
|
)
|
|
with context.unit_of_work.transaction() as ctx:
|
|
context.project = ctx.projects.create(project)
|
|
context.test_project = context.project
|
|
|
|
|
|
@given("I have an unsaved project without ID")
|
|
def step_create_unsaved_project(context: Context) -> None:
|
|
"""Create an unsaved project without ID."""
|
|
context.project = Project(
|
|
id=None,
|
|
name="test_project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="mock-gpt",
|
|
),
|
|
)
|
|
|
|
|
|
@when("I try to create a plan for the unsaved project")
|
|
def step_try_create_plan_unsaved(context: Context) -> None:
|
|
"""Try to create a plan for unsaved project."""
|
|
try:
|
|
context.plan_service.create_plan(context.project, "Test prompt", "test-plan")
|
|
context.exception = None
|
|
except Exception as e:
|
|
context.exception = e
|
|
|
|
|
|
@then('a ValidationError should be raised with message "{message}"')
|
|
def step_check_validation_error(context: Context, message: str) -> None:
|
|
"""Check that a ValidationError was raised with specific message."""
|
|
assert context.exception is not None
|
|
assert isinstance(context.exception, ValidationError)
|
|
assert message in str(context.exception.message)
|
|
|
|
|
|
@when("I get the current plan for the project")
|
|
def step_get_current_plan(context: Context) -> None:
|
|
"""Fetch the current plan for the active project."""
|
|
context.current_plan_result = context.plan_service.get_current_plan(context.project)
|
|
context.exception = None
|
|
|
|
|
|
@then("the current plan result should be None")
|
|
def step_assert_current_plan_none(context: Context) -> None:
|
|
"""Assert that no current plan was found."""
|
|
assert getattr(context, "current_plan_result", "__missing__") is None
|
|
|
|
|
|
@when("I list the plans for the project")
|
|
def step_list_plans_for_project(context: Context) -> None:
|
|
"""Retrieve all plans for the active project."""
|
|
context.plans_result = context.plan_service.list_plans(context.project)
|
|
context.exception = None
|
|
|
|
|
|
@then("the plans result should be an empty list")
|
|
def step_assert_plans_empty(context: Context) -> None:
|
|
"""Verify no plans were returned."""
|
|
assert getattr(context, "plans_result", None) == []
|
|
|
|
|
|
@when('I attempt to switch to plan "{name}"')
|
|
def step_attempt_switch_plan(context: Context, name: str) -> None:
|
|
"""Attempt to switch plans while capturing errors."""
|
|
try:
|
|
context.switch_result = context.plan_service.switch_to_plan(
|
|
context.project, name
|
|
)
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.switch_result = None
|
|
context.exception = exc
|
|
|
|
|
|
@then("the validation error details should hint to initialize the project")
|
|
def step_assert_validation_hint(context: Context) -> None:
|
|
"""Validate the hint provided with the ValidationError."""
|
|
assert isinstance(context.exception, ValidationError)
|
|
details = getattr(context.exception, "details", {}) or {}
|
|
assert details.get("hint") == "Initialize project first with 'agents init'"
|
|
|
|
|
|
@given('the project has plans named "{first}" and "{second}"')
|
|
def step_project_with_named_plans(context: Context, first: str, second: str) -> None:
|
|
"""Ensure the project has specific plan names configured."""
|
|
assert context.project.id is not None, (
|
|
"Project must be persisted before adding plans"
|
|
)
|
|
with context.unit_of_work.transaction() as ctx:
|
|
current_plan = ctx.plans.get_current_for_project(context.project.id)
|
|
assert current_plan is not None, "Current plan is required for this step"
|
|
current_plan.name = first
|
|
ctx.plans.update(current_plan)
|
|
|
|
second_plan = Plan(
|
|
id=None,
|
|
project_id=context.project.id,
|
|
name=second,
|
|
prompt="Secondary plan",
|
|
status=PlanStatus.PENDING,
|
|
current=False,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
build=None,
|
|
build_started_at=None,
|
|
build_completed_at=None,
|
|
model_used=None,
|
|
token_count=None,
|
|
result=None,
|
|
applied_at=None,
|
|
files_created=None,
|
|
files_modified=None,
|
|
files_deleted=None,
|
|
)
|
|
ctx.plans.create(second_plan)
|
|
|
|
context.available_plan_names = [first, second]
|
|
|
|
|
|
@then(
|
|
'the validation error details should list available plans "{first}" and "{second}"'
|
|
)
|
|
def step_assert_available_plans(context: Context, first: str, second: str) -> None:
|
|
"""Check that available plan names are surfaced in the error."""
|
|
assert isinstance(context.exception, ValidationError)
|
|
details = getattr(context.exception, "details", {}) or {}
|
|
assert details.get("available_plans") == [first, second]
|
|
|
|
|
|
@when('I attempt to add prompt "{prompt}" to the current plan')
|
|
def step_attempt_add_prompt(context: Context, prompt: str) -> None:
|
|
"""Attempt to add instructions to the current plan, capturing errors."""
|
|
try:
|
|
context.plan_service.add_to_plan(context.project, prompt)
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.exception = exc
|
|
|
|
|
|
@then("the plan error details should suggest creating a plan first")
|
|
def step_assert_plan_hint(context: Context) -> None:
|
|
"""Confirm the PlanError hint references plan creation."""
|
|
assert isinstance(context.exception, PlanError)
|
|
details = getattr(context.exception, "details", {}) or {}
|
|
assert details.get("hint") == "Create a plan first with 'agents new'"
|
|
|
|
|
|
@given("the current plan prompt is set to None")
|
|
def step_set_current_plan_prompt_none(context: Context) -> None:
|
|
"""Record the current plan for later patching to simulate a missing prompt."""
|
|
assert context.project.id is not None, "Project must be saved"
|
|
with context.unit_of_work.transaction() as ctx:
|
|
plan = ctx.plans.get_current_for_project(context.project.id)
|
|
assert plan is not None, "Expected a current plan to modify"
|
|
context.current_plan = plan
|
|
context.plan_previous_updated_at = plan.updated_at
|
|
context.plan_id_for_none_prompt = plan.id
|
|
|
|
|
|
@when('I add prompt "{prompt}" to the current plan')
|
|
def step_add_prompt_to_current_plan(context: Context, prompt: str) -> None:
|
|
"""Add instructions to the current plan and persist the result."""
|
|
assert context.project.id is not None, "Project must be saved"
|
|
|
|
with context.unit_of_work.transaction() as ctx:
|
|
plan_before = ctx.plans.get_current_for_project(context.project.id)
|
|
assert plan_before is not None, "Expected a current plan before adding prompt"
|
|
context.plan_previous_updated_at = plan_before.updated_at
|
|
|
|
from cleveragents.infrastructure.database.repositories import PlanRepository
|
|
|
|
original_get_current = PlanRepository.get_current_for_project
|
|
|
|
def get_current_with_none_prompt(self, project_id: int): # type: ignore[override]
|
|
plan = original_get_current(self, project_id)
|
|
target_plan_id = getattr(context, "plan_id_for_none_prompt", None)
|
|
if plan and plan.id == target_plan_id:
|
|
plan_dict = plan.model_dump()
|
|
plan_with_none = Plan.model_construct(**plan_dict)
|
|
object.__setattr__(plan_with_none, "prompt", None)
|
|
return plan_with_none
|
|
return plan
|
|
|
|
with patch.object(
|
|
PlanRepository, "get_current_for_project", get_current_with_none_prompt
|
|
):
|
|
context.plan_service.add_to_plan(context.project, prompt)
|
|
|
|
with context.unit_of_work.transaction() as ctx:
|
|
context.current_plan_after_add = ctx.plans.get_current_for_project(
|
|
context.project.id
|
|
)
|
|
context.added_prompt = prompt
|
|
|
|
|
|
@then('the current plan prompt should be "{expected}"')
|
|
def step_assert_current_plan_prompt(context: Context, expected: str) -> None:
|
|
"""Verify the current plan prompt matches expectations."""
|
|
plan = getattr(context, "current_plan_after_add", None)
|
|
assert plan is not None, "No plan result found after adding prompt"
|
|
assert plan.prompt == expected
|
|
|
|
|
|
@then("the current plan status should be PENDING")
|
|
def step_assert_current_plan_status_pending(context: Context) -> None:
|
|
"""Ensure the current plan status is set to PENDING."""
|
|
plan = getattr(context, "current_plan_after_add", None)
|
|
assert plan is not None, "No plan result found after adding prompt"
|
|
assert plan.status == PlanStatus.PENDING
|
|
|
|
|
|
@then("the current plan updated_at should be refreshed")
|
|
def step_assert_current_plan_updated(context: Context) -> None:
|
|
"""Confirm the plan updated_at timestamp increased after adding prompt."""
|
|
plan = getattr(context, "current_plan_after_add", None)
|
|
assert plan is not None, "No plan result found after adding prompt"
|
|
previous = getattr(context, "plan_previous_updated_at", None)
|
|
assert previous is not None, "No previous updated_at stored"
|
|
assert plan.updated_at > previous
|
|
|
|
|
|
@given("I have a saved project with no current plan")
|
|
def step_create_saved_project_no_plan(context: Context) -> None:
|
|
"""Create a saved project with no current plan."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = Project(
|
|
id=None,
|
|
name="test_project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="mock-gpt",
|
|
),
|
|
)
|
|
context.project = ctx.projects.create(project)
|
|
|
|
|
|
@when("I try to build the plan for the project")
|
|
def step_try_build_plan_no_current(context: Context) -> None:
|
|
"""Try to build plan when no current plan exists."""
|
|
try:
|
|
context.plan_service.build_plan(context.project)
|
|
context.exception = None
|
|
except Exception as e:
|
|
context.exception = e
|
|
|
|
|
|
@then('a PlanError should be raised with message "{message}"')
|
|
def step_check_plan_error(context: Context, message: str) -> None:
|
|
"""Check that a PlanError was raised with specific message."""
|
|
assert context.exception is not None, "No exception captured"
|
|
assert isinstance(context.exception, PlanError), (
|
|
f"Unexpected exception type {type(context.exception).__name__}"
|
|
)
|
|
actual_message = str(getattr(context.exception, "message", "")) or str(
|
|
context.exception
|
|
)
|
|
assert message in actual_message, f"Expected '{message}' in '{actual_message}'"
|
|
|
|
|
|
@then('a PlanError should be raised containing "{message}"')
|
|
def step_check_plan_error_contains(context: Context, message: str) -> None:
|
|
"""Check that a PlanError message contains specific text."""
|
|
assert context.exception is not None
|
|
assert isinstance(context.exception, PlanError)
|
|
assert message in str(context.exception.message)
|
|
|
|
|
|
@then("the PlanError details should include provider diagnostics")
|
|
def step_plan_error_details_include_provider_diagnostics(context: Context) -> None:
|
|
"""Ensure provider diagnostics are present in the captured PlanError details."""
|
|
assert context.exception is not None, "No exception captured"
|
|
assert isinstance(context.exception, PlanError), (
|
|
f"Expected PlanError but got {type(context.exception).__name__}"
|
|
)
|
|
details = getattr(context.exception, "details", {}) or {}
|
|
required_keys = {
|
|
"configured_providers",
|
|
"expected_env_vars",
|
|
"requested_provider",
|
|
"requested_model",
|
|
}
|
|
missing = sorted(key for key in required_keys if key not in details)
|
|
assert not missing, f"Missing provider diagnostics keys: {', '.join(missing)}"
|
|
|
|
|
|
@given("I configured an auto-debug plan with a disappearing plan ID")
|
|
def step_configure_auto_debug_disappearing_plan(context: Context) -> None:
|
|
"""Set up a plan service whose current plan loses its ID mid-run."""
|
|
assert hasattr(context, "temp_dir"), (
|
|
"Temporary directory setup is required before configuring auto-debug"
|
|
)
|
|
|
|
class FlickeringPlan:
|
|
def __init__(self, plan_id: int) -> None:
|
|
self._plan_id = plan_id
|
|
self._first_access = True
|
|
self.prompt = "Investigate flaky build"
|
|
self.name = "auto-debug-plan"
|
|
self.status = PlanStatus.PENDING
|
|
self.current = True
|
|
self.updated_at = datetime.now()
|
|
self.result = None
|
|
|
|
@property
|
|
def id(self) -> int | None:
|
|
if self._first_access:
|
|
self._first_access = False
|
|
return self._plan_id
|
|
return None
|
|
|
|
class FakePlansRepository:
|
|
def __init__(self, plan):
|
|
self.plan = plan
|
|
|
|
def get_current_for_project(self, project_id: int):
|
|
return self.plan
|
|
|
|
class FakeContextsRepository:
|
|
def __init__(self) -> None:
|
|
self.called = False
|
|
|
|
def get_for_plan(self, plan_id: int):
|
|
self.called = True
|
|
return []
|
|
|
|
class FakeDebugAttemptsRepository:
|
|
def __init__(self) -> None:
|
|
self.records: list = []
|
|
|
|
def add(self, attempt):
|
|
attempt.id = len(self.records) + 1
|
|
self.records.append(attempt)
|
|
return attempt
|
|
|
|
def get(self, attempt_id: int):
|
|
return None
|
|
|
|
def update(self, attempt):
|
|
self.updated_attempt = attempt
|
|
|
|
class FakeUnitOfWork:
|
|
def __init__(self, contexts: list[SimpleNamespace]) -> None:
|
|
self._contexts = contexts
|
|
|
|
def transaction(self):
|
|
outer = self
|
|
|
|
class _Transaction:
|
|
def __enter__(self_inner):
|
|
if not outer._contexts:
|
|
raise AssertionError("No fake transaction contexts remaining")
|
|
return outer._contexts.pop(0)
|
|
|
|
def __exit__(self_inner, exc_type, exc_val, exc_tb):
|
|
return False
|
|
|
|
return _Transaction()
|
|
|
|
flicker_plan = FlickeringPlan(plan_id=101)
|
|
contexts_repo = FakeContextsRepository()
|
|
debug_repo = FakeDebugAttemptsRepository()
|
|
fake_contexts = [
|
|
SimpleNamespace(plans=FakePlansRepository(flicker_plan)),
|
|
SimpleNamespace(contexts=contexts_repo),
|
|
SimpleNamespace(debug_attempts=debug_repo),
|
|
]
|
|
|
|
context.auto_debug_context_repo = contexts_repo
|
|
context.auto_debug_debug_repo = debug_repo
|
|
context.plan_service = PlanService(
|
|
settings=Settings(),
|
|
unit_of_work=FakeUnitOfWork(fake_contexts),
|
|
ai_provider=None,
|
|
)
|
|
context.project = Project(
|
|
id=1,
|
|
name="auto-debug-project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=1,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="mock-gpt",
|
|
),
|
|
)
|
|
|
|
|
|
@when("I run auto-debug build with missing plan context")
|
|
def step_run_auto_debug_missing_context(context: Context) -> None:
|
|
"""Trigger auto-debug build to exercise empty context handling."""
|
|
context.plan_service.build_plan = MagicMock(
|
|
side_effect=RuntimeError("Plan build failed")
|
|
)
|
|
captured_state: dict[str, object] = {}
|
|
|
|
class DummyAutoDebugAgent:
|
|
def __init__(self, **_kwargs):
|
|
pass
|
|
|
|
def invoke(self, debug_state, config=None):
|
|
captured_state["state"] = debug_state
|
|
captured_state["config"] = config
|
|
return {"result": {"success": False, "fix": None}}
|
|
|
|
with (
|
|
patch.object(
|
|
context.plan_service, "_prepare_langsmith_config", return_value={}
|
|
),
|
|
patch(
|
|
"cleveragents.agents.AutoDebugAgent",
|
|
DummyAutoDebugAgent,
|
|
),
|
|
):
|
|
context.auto_debug_result = context.plan_service.auto_debug_build(
|
|
context.project, max_attempts=1
|
|
)
|
|
|
|
context.auto_debug_captured_state = captured_state.get("state")
|
|
|
|
|
|
@then("the auto-debug build should fall back to an empty context")
|
|
def step_assert_auto_debug_empty_context(context: Context) -> None:
|
|
"""Verify empty code context and debug attempt tracking."""
|
|
result = getattr(context, "auto_debug_result", None)
|
|
assert result is not None, "Auto-debug result was not captured"
|
|
success, changes, error = result
|
|
assert success is False
|
|
assert changes == []
|
|
assert error == "Plan build failed"
|
|
|
|
contexts_repo = getattr(context, "auto_debug_context_repo", None)
|
|
assert contexts_repo is not None, "Contexts repository tracking missing"
|
|
assert contexts_repo.called is False
|
|
|
|
debug_state = getattr(context, "auto_debug_captured_state", None)
|
|
assert debug_state is not None, "Debug state was not recorded"
|
|
assert debug_state.get("code_context") == ""
|
|
|
|
|
|
@given("I have a saved project with current plan missing ID")
|
|
def step_create_project_plan_no_id(context: Context) -> None:
|
|
"""Create a project with current plan that has no ID."""
|
|
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = Project(
|
|
id=None,
|
|
name="test_project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="mock-gpt",
|
|
),
|
|
)
|
|
context.project = ctx.projects.create(project)
|
|
|
|
# Create a plan that will be saved
|
|
plan = Plan(
|
|
id=None,
|
|
project_id=context.project.id,
|
|
name="test-plan",
|
|
prompt="Test prompt",
|
|
status=PlanStatus.PENDING,
|
|
current=True,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
build=None,
|
|
build_started_at=None,
|
|
build_completed_at=None,
|
|
model_used=None,
|
|
token_count=None,
|
|
result=None,
|
|
applied_at=None,
|
|
files_created=None,
|
|
files_modified=None,
|
|
files_deleted=None,
|
|
)
|
|
created_plan = ctx.plans.create(plan)
|
|
|
|
# Update project to have current plan ID
|
|
context.project.current_plan_id = created_plan.id
|
|
ctx.projects.update(context.project)
|
|
|
|
# Store the created plan for mocking
|
|
context.created_plan_with_id = created_plan
|
|
|
|
|
|
@when("I try to build the plan with invalid ID")
|
|
def step_try_build_plan_invalid_id(context: Context) -> None:
|
|
"""Try to build plan with invalid ID."""
|
|
from unittest.mock import patch
|
|
|
|
# Mock get_current_for_project to return a plan without ID
|
|
def mock_get_current(project_id):
|
|
# Return the plan but with ID set to None
|
|
plan = context.created_plan_with_id
|
|
plan.id = None
|
|
return plan
|
|
|
|
try:
|
|
# Patch the repository method at the class level
|
|
with patch.object(
|
|
context.unit_of_work.transaction().__enter__().plans.__class__,
|
|
"get_current_for_project",
|
|
side_effect=mock_get_current,
|
|
):
|
|
context.plan_service.build_plan(context.project)
|
|
context.exception = None
|
|
except Exception as e:
|
|
context.exception = e
|
|
|
|
|
|
@given("I have a saved project with valid current plan")
|
|
def step_create_project_with_valid_plan(context: Context) -> None:
|
|
"""Create a project with valid current plan."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = Project(
|
|
id=None,
|
|
name="test_project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="mock-gpt",
|
|
),
|
|
)
|
|
context.project = ctx.projects.create(project)
|
|
|
|
plan = Plan(
|
|
id=None,
|
|
project_id=context.project.id,
|
|
name="test-plan",
|
|
prompt="Test prompt",
|
|
status=PlanStatus.PENDING,
|
|
current=True,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
build=None,
|
|
build_started_at=None,
|
|
build_completed_at=None,
|
|
model_used=None,
|
|
token_count=None,
|
|
result=None,
|
|
applied_at=None,
|
|
files_created=None,
|
|
files_modified=None,
|
|
files_deleted=None,
|
|
)
|
|
context.current_plan = ctx.plans.create(plan)
|
|
if context.project.id and context.current_plan.id:
|
|
ctx.plans.set_current(context.project.id, context.current_plan.id)
|
|
|
|
|
|
@given("I have a plan service without AI provider")
|
|
def step_create_plan_service_no_provider(context: Context) -> None:
|
|
"""Create a plan service without AI provider."""
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
|
|
if not hasattr(context, "env_vars_to_clean"):
|
|
context.env_vars_to_clean = []
|
|
if "CLEVERAGENTS_TESTING_USE_MOCK_AI" not in context.env_vars_to_clean:
|
|
context.env_vars_to_clean.append("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
|
|
settings = Settings()
|
|
context.plan_service = PlanService(
|
|
settings=settings,
|
|
unit_of_work=context.unit_of_work,
|
|
ai_provider=None,
|
|
provider_registry=_NoProviderRegistry(),
|
|
)
|
|
|
|
|
|
@when("I try to build the plan without AI provider")
|
|
def step_try_build_without_provider(context: Context) -> None:
|
|
"""Try to build plan without AI provider."""
|
|
try:
|
|
context.plan_service.build_plan(context.project)
|
|
context.exception = None
|
|
except Exception as e:
|
|
context.exception = e
|
|
|
|
|
|
@when("I get pending changes for the project")
|
|
def step_get_pending_changes(context: Context) -> None:
|
|
"""Get pending changes for the project."""
|
|
context.pending_changes = context.plan_service.get_pending_changes(context.project)
|
|
|
|
|
|
@then("the pending changes list should be empty")
|
|
def step_check_pending_changes_empty(context: Context) -> None:
|
|
"""Check that pending changes list is empty."""
|
|
assert context.pending_changes == []
|
|
|
|
|
|
@given("I have a saved project with current plan")
|
|
def step_create_project_with_current_plan(context: Context) -> None:
|
|
"""Create a project with current plan."""
|
|
step_create_project_with_valid_plan(context)
|
|
|
|
|
|
@given("the current plan has applied and unapplied changes")
|
|
def step_add_mixed_changes(context: Context) -> None:
|
|
"""Add both applied and unapplied changes to the plan."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
# Add applied change
|
|
applied_change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path="applied.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="applied content",
|
|
new_path=None,
|
|
applied=True,
|
|
applied_at=datetime.now(),
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(applied_change)
|
|
|
|
# Add unapplied changes
|
|
for i in range(2):
|
|
unapplied_change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path=f"unapplied_{i}.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content=f"unapplied content {i}",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(unapplied_change)
|
|
|
|
|
|
@then("I should get only the unapplied changes")
|
|
def step_check_only_unapplied(context: Context) -> None:
|
|
"""Check that only unapplied changes are returned."""
|
|
assert len(context.pending_changes) == 2
|
|
for change in context.pending_changes:
|
|
assert not change.applied
|
|
assert "unapplied" in change.file_path
|
|
|
|
|
|
@when("I try to apply changes without current plan")
|
|
def step_try_apply_without_plan(context: Context) -> None:
|
|
"""Try to apply changes without current plan."""
|
|
try:
|
|
context.plan_service.apply_changes(context.project)
|
|
context.exception = None
|
|
except Exception as e:
|
|
context.exception = e
|
|
|
|
|
|
@given("the plan has a pending CREATE change")
|
|
def step_add_create_change(context: Context) -> None:
|
|
"""Add a pending CREATE change to the plan."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path="new_file.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# New file content\nprint('Hello')",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
context.create_change = ctx.changes.add(change)
|
|
|
|
|
|
@when("I apply the plan changes")
|
|
def step_apply_changes(context: Context) -> None:
|
|
"""Apply the changes."""
|
|
context.applied_count = context.plan_service.apply_changes(context.project)
|
|
|
|
|
|
@then("the file should be created with correct content")
|
|
def step_check_file_created(context: Context) -> None:
|
|
"""Check that file was created with correct content."""
|
|
file_path = context.project.path / "new_file.py"
|
|
assert file_path.exists()
|
|
assert file_path.read_text() == "# New file content\nprint('Hello')"
|
|
|
|
|
|
@then("the change should be marked as applied in plan service")
|
|
def step_check_change_applied(context: Context) -> None:
|
|
"""Check that change was marked as applied."""
|
|
assert context.applied_count == 1
|
|
|
|
|
|
@given("the plan has a pending MODIFY change for existing file")
|
|
def step_add_modify_change(context: Context) -> None:
|
|
"""Add a pending MODIFY change for existing file."""
|
|
# Create the file first
|
|
file_path = context.project.path / "existing.py"
|
|
file_path.write_text("# Original content")
|
|
|
|
with context.unit_of_work.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path="existing.py",
|
|
operation=OperationType.MODIFY,
|
|
original_content="# Original content",
|
|
new_content="# Modified content\nprint('Modified')",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
context.modify_change = ctx.changes.add(change)
|
|
|
|
|
|
@then("the file should be modified with new content")
|
|
def step_check_file_modified(context: Context) -> None:
|
|
"""Check that file was modified with new content."""
|
|
file_path = context.project.path / "existing.py"
|
|
assert file_path.exists()
|
|
assert file_path.read_text() == "# Modified content\nprint('Modified')"
|
|
|
|
|
|
@given("the plan has a pending DELETE change for existing file")
|
|
def step_add_delete_change(context: Context) -> None:
|
|
"""Add a pending DELETE change for existing file."""
|
|
# Create the file first
|
|
file_path = context.project.path / "to_delete.py"
|
|
file_path.write_text("# To be deleted")
|
|
|
|
with context.unit_of_work.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path="to_delete.py",
|
|
operation=OperationType.DELETE,
|
|
original_content="# To be deleted",
|
|
new_content=None,
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
context.delete_change = ctx.changes.add(change)
|
|
|
|
|
|
@then("the file should be deleted")
|
|
def step_check_file_deleted(context: Context) -> None:
|
|
"""Check that file was deleted."""
|
|
file_path = context.project.path / "to_delete.py"
|
|
assert not file_path.exists()
|
|
|
|
|
|
@given("the plan has a pending MOVE change for existing file")
|
|
def step_add_move_change(context: Context) -> None:
|
|
"""Add a pending MOVE change for existing file."""
|
|
# Create the file first
|
|
file_path = context.project.path / "old_location.py"
|
|
file_path.write_text("# File to move")
|
|
|
|
with context.unit_of_work.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path="old_location.py",
|
|
operation=OperationType.MOVE,
|
|
original_content="# File to move",
|
|
new_content="# File to move",
|
|
new_path="new_location.py",
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
context.move_change = ctx.changes.add(change)
|
|
|
|
|
|
@then("the file should be moved to new location")
|
|
def step_check_file_moved(context: Context) -> None:
|
|
"""Check that file was moved to new location."""
|
|
old_path = context.project.path / "old_location.py"
|
|
new_path = context.project.path / "new_location.py"
|
|
assert not old_path.exists()
|
|
assert new_path.exists()
|
|
assert new_path.read_text() == "# File to move"
|
|
|
|
|
|
@given("the plan has a pending MOVE change to a nested relative path")
|
|
def step_add_nested_move_change(context: Context) -> None:
|
|
"""Add a MOVE change that targets a nested relative destination."""
|
|
source_relative = Path("nested") / "source.py"
|
|
source_path = context.project.path / source_relative
|
|
source_path.parent.mkdir(parents=True, exist_ok=True)
|
|
source_path.write_text("# Nested file to move")
|
|
|
|
destination_relative = Path("dest") / "nested" / "target.py"
|
|
with context.unit_of_work.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path=str(source_relative),
|
|
operation=OperationType.MOVE,
|
|
original_content="# Nested file to move",
|
|
new_content="# Nested file to move",
|
|
new_path=str(destination_relative),
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
context.nested_move_paths = (
|
|
source_path,
|
|
context.project.path / destination_relative,
|
|
)
|
|
|
|
|
|
@then("the file should be moved to the nested relative location")
|
|
def step_check_nested_move(context: Context) -> None:
|
|
"""Verify nested MOVE change relocated the file correctly."""
|
|
assert hasattr(context, "nested_move_paths"), "Nested move paths were not recorded"
|
|
old_path, new_path = context.nested_move_paths
|
|
assert not old_path.exists(), "Original nested file still exists"
|
|
assert new_path.exists(), "Nested destination file was not created"
|
|
assert new_path.read_text() == "# Nested file to move", (
|
|
"Nested destination content mismatch"
|
|
)
|
|
|
|
|
|
@given("the plan has a pending MOVE change to an absolute path")
|
|
def step_add_absolute_move_change(context: Context) -> None:
|
|
"""Add a MOVE change whose destination is an absolute path."""
|
|
assert getattr(context, "project", None) is not None, "Project is required"
|
|
assert getattr(context, "current_plan", None) is not None, (
|
|
"Current plan is required"
|
|
)
|
|
|
|
source_path = context.project.path / "absolute_source.py"
|
|
source_path.write_text("# Absolute move source")
|
|
absolute_target = (context.project.path / "absolute_target.py").resolve()
|
|
|
|
with context.unit_of_work.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path="absolute_source.py",
|
|
operation=OperationType.MOVE,
|
|
original_content="# Absolute move source",
|
|
new_content="# Absolute move source",
|
|
new_path=str(absolute_target),
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
context.absolute_move_paths = (source_path, absolute_target)
|
|
|
|
|
|
@then("the absolute destination should exist and the source should be removed")
|
|
def step_check_absolute_move_destination(context: Context) -> None:
|
|
"""Ensure MOVE changes create files at absolute targets."""
|
|
paths = getattr(context, "absolute_move_paths", None)
|
|
assert paths is not None, "Absolute move paths were not recorded"
|
|
source_path, absolute_target = paths
|
|
assert not source_path.exists(), "Source file still exists after MOVE"
|
|
destination_path = Path(absolute_target)
|
|
assert destination_path.exists(), "Absolute destination file missing"
|
|
assert destination_path.read_text() == "# Absolute move source"
|
|
assert getattr(context, "applied_count", 0) == 1
|
|
|
|
|
|
@given("the plan has changes with absolute file paths")
|
|
def step_add_absolute_path_changes(context: Context) -> None:
|
|
"""Add changes with absolute file paths."""
|
|
abs_path = context.temp_dir / "absolute_file.py"
|
|
|
|
with context.unit_of_work.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path=str(abs_path), # Absolute path
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# Absolute path content",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@then("the files should be created at absolute paths")
|
|
def step_check_absolute_paths(context: Context) -> None:
|
|
"""Check that files were created at absolute paths."""
|
|
abs_path = context.temp_dir / "absolute_file.py"
|
|
assert abs_path.exists()
|
|
assert abs_path.read_text() == "# Absolute path content"
|
|
|
|
|
|
@then("the plan changes should be marked as applied")
|
|
def step_check_changes_marked_applied(context: Context) -> None:
|
|
"""Check that changes were marked as applied."""
|
|
assert context.applied_count > 0
|
|
|
|
|
|
@then("the changes should be marked as applied")
|
|
def step_check_changes_applied(context: Context) -> None:
|
|
"""Check that changes were marked as applied."""
|
|
assert context.applied_count > 0
|
|
|
|
|
|
@given("the plan has a change that will fail on apply")
|
|
def step_add_failing_change(context: Context) -> None:
|
|
"""Add a change that will fail when applied."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
# Create a file then simulate permission denial via mock
|
|
test_file = context.temp_dir / "readonly_file.py"
|
|
test_file.write_text("original content")
|
|
|
|
# Try to modify a read-only file - this should fail
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path=str(test_file), # Existing readonly file
|
|
operation=OperationType.MODIFY, # Try to modify it
|
|
original_content="original content",
|
|
new_content="# New content that will fail to write",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
# Store path for the mock in the apply step
|
|
context.readonly_file = str(test_file)
|
|
|
|
|
|
@when("I try to apply the changes")
|
|
def step_try_apply_changes(context: Context) -> None:
|
|
"""Try to apply changes, simulating PermissionError for readonly targets."""
|
|
target = getattr(context, "readonly_file", None)
|
|
|
|
original_write_text = Path.write_text
|
|
|
|
def guarded_write_text(self, *args, **kwargs):
|
|
if target and str(self) == target:
|
|
raise PermissionError(f"[Errno 13] Permission denied: '{self}'")
|
|
return original_write_text(self, *args, **kwargs)
|
|
|
|
try:
|
|
with (
|
|
patch.object(Path, "write_text", guarded_write_text)
|
|
if target
|
|
else contextlib.nullcontext()
|
|
):
|
|
result = context.plan_service.apply_changes(context.project)
|
|
context.exception = None
|
|
context.apply_result = result
|
|
except Exception as e:
|
|
context.exception = e
|
|
|
|
|
|
@then("a PlanError should be raised with file operation details")
|
|
def step_check_plan_error_with_details(context: Context) -> None:
|
|
"""Check that PlanError was raised with file operation details."""
|
|
assert context.exception is not None, (
|
|
f"Expected an exception but got result: {getattr(context, 'apply_result', 'none')}"
|
|
)
|
|
assert isinstance(context.exception, PlanError), (
|
|
f"Expected PlanError but got {type(context.exception).__name__}: {context.exception}"
|
|
)
|
|
assert "Failed to apply change" in str(context.exception.message)
|
|
|
|
|
|
@given("I have a saved project")
|
|
def step_create_saved_project(context: Context) -> None:
|
|
"""Create a saved project."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = Project(
|
|
id=None,
|
|
name="test_project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="mock-gpt",
|
|
),
|
|
)
|
|
context.project = ctx.projects.create(project)
|
|
|
|
|
|
@when("I create a plan with minimal prompt")
|
|
def step_create_plan_minimal_prompt(context: Context) -> None:
|
|
"""Create a plan with minimal prompt."""
|
|
# Use a single character prompt - the service should generate a name from it
|
|
context.created_plan = context.plan_service.create_plan(context.project, "x", None)
|
|
|
|
|
|
@then("the plan name should be generated from prompt")
|
|
def step_check_generated_name(context: Context) -> None:
|
|
"""Check that plan name was generated from the prompt."""
|
|
# With prompt "x", the name should be "x"
|
|
assert context.created_plan.name == "x"
|
|
|
|
|
|
@when("I create a plan with very long prompt")
|
|
def step_create_plan_long_prompt(context: Context) -> None:
|
|
"""Create a plan with very long prompt."""
|
|
long_prompt = "This is a very long prompt with many words that should be truncated"
|
|
context.created_plan = context.plan_service.create_plan(
|
|
context.project, long_prompt, None
|
|
)
|
|
|
|
|
|
@then("the plan name should be truncated to first three words")
|
|
def step_check_truncated_name(context: Context) -> None:
|
|
"""Check that plan name was truncated."""
|
|
assert context.created_plan.name == "this_is_a"
|
|
|
|
|
|
@given("the plan has a MOVE change for non-existent file")
|
|
def step_add_move_nonexistent(context: Context) -> None:
|
|
"""Add a MOVE change for non-existent file."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path="nonexistent.py",
|
|
operation=OperationType.MOVE,
|
|
original_content=None,
|
|
new_content=None,
|
|
new_path="new_location.py",
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@then("the MOVE operation should handle missing source gracefully")
|
|
def step_check_move_handles_missing(context: Context) -> None:
|
|
"""Check that MOVE handles missing source file."""
|
|
# The operation should complete without error even if source doesn't exist
|
|
assert context.applied_count == 1
|
|
# New file should not exist since source didn't exist
|
|
new_path = context.project.path / "new_location.py"
|
|
assert not new_path.exists()
|
|
|
|
|
|
@given("the plan has a DELETE change for non-existent file")
|
|
def step_add_delete_nonexistent(context: Context) -> None:
|
|
"""Add a DELETE change for non-existent file."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path="already_deleted.py",
|
|
operation=OperationType.DELETE,
|
|
original_content=None,
|
|
new_content=None,
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@then("the DELETE operation should handle missing file gracefully")
|
|
def step_check_delete_handles_missing(context: Context) -> None:
|
|
"""Check that DELETE handles missing file."""
|
|
# The operation should complete without error even if file doesn't exist
|
|
assert context.applied_count == 1
|
|
|
|
|
|
@given("I have a plan service with mock AI provider")
|
|
def step_create_plan_service_with_mock_provider(context: Context) -> None:
|
|
"""Create a plan service with mock AI provider."""
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
|
|
if not hasattr(context, "env_vars_to_clean"):
|
|
context.env_vars_to_clean = []
|
|
if "CLEVERAGENTS_TESTING_USE_MOCK_AI" not in context.env_vars_to_clean:
|
|
context.env_vars_to_clean.append("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
|
|
mock_provider = MagicMock()
|
|
|
|
# Mock the generate_changes method
|
|
def generate_changes_with_callback(
|
|
project,
|
|
plan,
|
|
contexts,
|
|
actor_context: ActorInvocationContext | None = None,
|
|
progress_callback=None,
|
|
):
|
|
if progress_callback:
|
|
# Simulate progress updates
|
|
for i in range(0, 101, 20):
|
|
progress_callback(i)
|
|
|
|
# Return mock response
|
|
return ProviderResponse(
|
|
changes=[
|
|
Change(
|
|
id=None,
|
|
plan_id=plan.id,
|
|
file_path="generated.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# Generated content",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
],
|
|
model_used="mock-gpt",
|
|
token_count=100,
|
|
)
|
|
|
|
mock_provider.generate_changes = generate_changes_with_callback
|
|
|
|
settings = Settings()
|
|
context.plan_service = PlanService(
|
|
settings=settings, unit_of_work=context.unit_of_work, ai_provider=mock_provider
|
|
)
|
|
context.mock_provider = mock_provider
|
|
|
|
|
|
@when("I build the plan with progress callback")
|
|
def step_build_with_progress_callback(context: Context) -> None:
|
|
"""Build plan with progress callback."""
|
|
context.progress_values = []
|
|
|
|
def progress_callback(value):
|
|
context.progress_values.append(value)
|
|
|
|
context.changes = context.plan_service.build_plan(
|
|
context.project, progress_callback=progress_callback
|
|
)
|
|
|
|
|
|
@then("the progress callback should be called with values from 0 to 100")
|
|
def step_check_progress_callback(context: Context) -> None:
|
|
"""Check that progress callback was called correctly."""
|
|
assert len(context.progress_values) > 0
|
|
assert 0 in context.progress_values
|
|
assert 100 in context.progress_values
|
|
# Check that values are in ascending order
|
|
for i in range(1, len(context.progress_values)):
|
|
assert context.progress_values[i] >= context.progress_values[i - 1]
|
|
|
|
|
|
@given("the plan has 3 applied and 2 unapplied changes")
|
|
def step_add_specific_mixed_changes(context: Context) -> None:
|
|
"""Add specific number of applied and unapplied changes."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
# Add 3 applied changes
|
|
for i in range(3):
|
|
applied_change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path=f"applied_{i}.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content=f"applied content {i}",
|
|
new_path=None,
|
|
applied=True,
|
|
applied_at=datetime.now(),
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(applied_change)
|
|
|
|
# Add 2 unapplied changes
|
|
for i in range(2):
|
|
unapplied_change = Change(
|
|
id=None,
|
|
plan_id=context.current_plan.id,
|
|
file_path=f"pending_{i}.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content=f"pending content {i}",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(unapplied_change)
|
|
|
|
|
|
@then("I should get exactly 2 pending changes")
|
|
def step_check_exactly_two_pending(context: Context) -> None:
|
|
"""Check that exactly 2 pending changes are returned."""
|
|
assert len(context.pending_changes) == 2
|
|
|
|
|
|
@then("the pending changes should not include applied ones")
|
|
def step_check_no_applied_in_pending(context: Context) -> None:
|
|
"""Check that pending changes don't include applied ones."""
|
|
for change in context.pending_changes:
|
|
assert not change.applied
|
|
assert "pending" in change.file_path
|
|
assert "applied" not in change.file_path
|
|
|
|
|
|
def _ensure_memory_database_url(context: Context) -> None:
|
|
"""Ensure plan service uses an in-memory database for memory tests."""
|
|
try:
|
|
if (
|
|
context.plan_service.settings.database_url
|
|
!= "sqlite:///file::memory:?cache=shared"
|
|
):
|
|
context.plan_service.settings.database_url = (
|
|
"sqlite:///file::memory:?cache=shared"
|
|
)
|
|
except AttributeError:
|
|
context.plan_service.settings = Settings(
|
|
database_url="sqlite:///file::memory:?cache=shared"
|
|
)
|
|
|
|
|
|
def _request_memory_service_for_session(
|
|
context: Context,
|
|
session_id: str,
|
|
persistent: bool,
|
|
max_messages: int | None,
|
|
) -> MemoryService:
|
|
if persistent:
|
|
_ensure_memory_database_url(context)
|
|
service = context.plan_service.get_memory_service(
|
|
session_id,
|
|
persistent=persistent,
|
|
max_messages=max_messages,
|
|
)
|
|
context.last_memory_service = service
|
|
context.last_memory_session = session_id
|
|
return service
|
|
|
|
|
|
@when(
|
|
'I request an in-memory memory service for session "{session_id}" with max messages {max_messages:d}'
|
|
)
|
|
def step_request_in_memory_memory_service(
|
|
context: Context, session_id: str, max_messages: int
|
|
) -> None:
|
|
service = _request_memory_service_for_session(
|
|
context, session_id, persistent=False, max_messages=max_messages
|
|
)
|
|
assert isinstance(service, MemoryService)
|
|
|
|
|
|
@when(
|
|
'I request a persistent memory service for session "{session_id}" with max messages {max_messages:d}'
|
|
)
|
|
def step_request_persistent_memory_service(
|
|
context: Context, session_id: str, max_messages: int
|
|
) -> None:
|
|
service = _request_memory_service_for_session(
|
|
context, session_id, persistent=True, max_messages=max_messages
|
|
)
|
|
assert isinstance(service, MemoryService)
|
|
|
|
|
|
@when(
|
|
'I request an in-memory memory service for session "{session_id}" without changing message limit'
|
|
)
|
|
def step_request_in_memory_without_limit(context: Context, session_id: str) -> None:
|
|
service = _request_memory_service_for_session(
|
|
context, session_id, persistent=False, max_messages=None
|
|
)
|
|
assert isinstance(service, MemoryService)
|
|
|
|
|
|
@then(
|
|
'the session "{session_id}" memory should be in-memory with max messages {max_messages:d}'
|
|
)
|
|
def step_verify_in_memory_with_max(
|
|
context: Context, session_id: str, max_messages: int
|
|
) -> None:
|
|
service = context.plan_service._memory_services.get(session_id)
|
|
assert service is not None
|
|
assert isinstance(service, MemoryService)
|
|
assert service.connection_string is None
|
|
assert service.max_messages == max_messages
|
|
context.last_memory_service = service
|
|
|
|
|
|
@then(
|
|
'the session "{session_id}" memory should be persistent with max messages {max_messages:d}'
|
|
)
|
|
def step_verify_persistent_with_max(
|
|
context: Context, session_id: str, max_messages: int
|
|
) -> None:
|
|
service = context.plan_service._memory_services.get(session_id)
|
|
assert service is not None
|
|
assert isinstance(service, MemoryService)
|
|
assert service.connection_string == context.plan_service.settings.database_url
|
|
assert service.max_messages == max_messages
|
|
context.last_memory_service = service
|
|
|
|
|
|
@then(
|
|
'the session "{session_id}" memory should be in-memory without a connection string'
|
|
)
|
|
def step_verify_in_memory_without_connection(context: Context, session_id: str) -> None:
|
|
service = context.plan_service._memory_services.get(session_id)
|
|
assert service is not None
|
|
assert isinstance(service, MemoryService)
|
|
assert service.connection_string is None
|
|
|
|
|
|
@when('I request conversation memory for session "{session_id}" with custom keys')
|
|
def step_request_conversation_memory_custom(context: Context, session_id: str) -> None:
|
|
adapter = context.plan_service.get_conversation_memory(
|
|
session_id,
|
|
persistent=False,
|
|
memory_key="chat_history",
|
|
input_key="prompt",
|
|
output_key="response",
|
|
return_messages=False,
|
|
max_messages=4,
|
|
)
|
|
context.conversation_adapter = adapter
|
|
context.last_memory_session = session_id
|
|
|
|
|
|
@then("the conversation adapter should expose the custom keys")
|
|
def step_verify_conversation_adapter_keys(context: Context) -> None:
|
|
adapter = getattr(context, "conversation_adapter", None)
|
|
assert isinstance(adapter, ConversationBufferMemoryAdapter)
|
|
assert adapter.memory_variables == ["chat_history"]
|
|
assert adapter.input_key == "prompt"
|
|
assert adapter.output_key == "response"
|
|
assert adapter.return_messages is False
|
|
|
|
|
|
@then("the conversation adapter should share the existing memory service")
|
|
def step_verify_conversation_adapter_service(context: Context) -> None:
|
|
session_id = getattr(context, "last_memory_session", None)
|
|
assert session_id is not None
|
|
service = context.plan_service._memory_services.get(session_id)
|
|
assert service is not None
|
|
adapter = context.conversation_adapter
|
|
assert adapter._message_history is service.message_history # type: ignore[attr-defined]
|
|
|
|
|
|
@given('I stored a chat message in session "{session_id}"')
|
|
def step_store_chat_message(context: Context, session_id: str) -> None:
|
|
service = _request_memory_service_for_session(
|
|
context, session_id, persistent=False, max_messages=None
|
|
)
|
|
service.add_user_message("hello memory branch")
|
|
assert len(service.message_history.messages) == 1
|
|
context.stored_memory_session = session_id
|
|
context.stored_memory_service = service
|
|
|
|
|
|
@when('I clear the session "{session_id}" memory with forget history')
|
|
def step_clear_memory_with_forget(context: Context, session_id: str) -> None:
|
|
assert getattr(context, "stored_memory_session", None) == session_id
|
|
context.plan_service.clear_memory(session_id, forget_history=True)
|
|
context.cleared_memory_service = context.stored_memory_service
|
|
|
|
|
|
@when('I clear the session "{session_id}" memory without forgetting history')
|
|
def step_clear_memory_without_forget(context: Context, session_id: str) -> None:
|
|
assert getattr(context, "stored_memory_session", None) == session_id
|
|
context.plan_service.clear_memory(session_id)
|
|
context.cleared_memory_service = context.stored_memory_service
|
|
|
|
|
|
@then('the session "{session_id}" memory should be removed and emptied')
|
|
def step_verify_memory_cleared(context: Context, session_id: str) -> None:
|
|
assert session_id not in context.plan_service._memory_services
|
|
service = getattr(context, "cleared_memory_service", None)
|
|
assert isinstance(service, MemoryService)
|
|
assert service.message_history.messages == []
|
|
|
|
|
|
@then('the session "{session_id}" memory should retain its stored messages')
|
|
def step_verify_memory_retained(context: Context, session_id: str) -> None:
|
|
service = getattr(context, "cleared_memory_service", None)
|
|
assert isinstance(service, MemoryService), "Memory service reference missing"
|
|
assert service is getattr(context, "stored_memory_service", None)
|
|
assert session_id not in context.plan_service._memory_services
|
|
messages = service.message_history.messages
|
|
assert messages, "Message history should remain intact"
|
|
|
|
|
|
class _StubPlansRepository:
|
|
"""Minimal stub for plan repository interactions."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
create_return_plan: Plan | None = None,
|
|
current_plan: Plan | None = None,
|
|
) -> None:
|
|
self._create_return_plan = create_return_plan
|
|
self._current_plan = current_plan
|
|
self.created_plans: list[Plan] = []
|
|
self.update_calls: list[Plan] = []
|
|
self.set_current_calls: list[tuple[int | None, int | None]] = []
|
|
|
|
def create(self, plan: Plan) -> Plan:
|
|
self.created_plans.append(plan)
|
|
if self._create_return_plan is not None:
|
|
return self._create_return_plan
|
|
return plan
|
|
|
|
def get_current_for_project(self, project_id: int | None) -> Plan | None:
|
|
return self._current_plan
|
|
|
|
def update(self, plan: Plan) -> None:
|
|
self.update_calls.append(plan)
|
|
|
|
def set_current(self, project_id: int | None, plan_id: int | None) -> None:
|
|
self.set_current_calls.append((project_id, plan_id))
|
|
|
|
|
|
class _StaticContext:
|
|
"""Context object supplying repositories for stubbed transactions."""
|
|
|
|
def __init__(
|
|
self,
|
|
plans_repo: _StubPlansRepository,
|
|
contexts_repo: MagicMock | None = None,
|
|
changes_repo: MagicMock | None = None,
|
|
) -> None:
|
|
self.plans = plans_repo
|
|
self.contexts = contexts_repo or MagicMock()
|
|
self.changes = changes_repo or MagicMock()
|
|
|
|
|
|
class _StubTransaction:
|
|
"""Simple context manager that can invoke an exit callback."""
|
|
|
|
def __init__(
|
|
self,
|
|
context_obj: _StaticContext,
|
|
exit_callback: Callable[[], None] | None = None,
|
|
) -> None:
|
|
self._context = context_obj
|
|
self._exit_callback = exit_callback
|
|
|
|
def __enter__(self) -> _StaticContext:
|
|
return self._context
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
|
|
if self._exit_callback is not None:
|
|
self._exit_callback()
|
|
return False
|
|
|
|
|
|
class _SequencedUnitOfWork(UnitOfWork):
|
|
"""Unit of work that plays back a sequence of stubbed transactions."""
|
|
|
|
def __init__(
|
|
self, transactions: list[tuple[_StaticContext, Callable[[], None] | None]]
|
|
):
|
|
self._transactions = transactions
|
|
|
|
def transaction(self) -> _StubTransaction: # type: ignore[override]
|
|
if not self._transactions:
|
|
raise RuntimeError("No stub transactions configured")
|
|
if len(self._transactions) > 1:
|
|
context_obj, exit_callback = self._transactions.pop(0)
|
|
else:
|
|
context_obj, exit_callback = self._transactions[0]
|
|
return _StubTransaction(context_obj, exit_callback)
|
|
|
|
|
|
@given(
|
|
"I configure a stub plan service whose current plan loses its ID after the transaction"
|
|
)
|
|
def step_configure_plan_service_losing_id(context: Context) -> None:
|
|
plan = Plan(
|
|
id=123,
|
|
project_id=1,
|
|
name="stub-plan",
|
|
prompt="stub",
|
|
status=PlanStatus.PENDING,
|
|
current=True,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
build=None,
|
|
build_started_at=None,
|
|
build_completed_at=None,
|
|
model_used=None,
|
|
token_count=None,
|
|
result=None,
|
|
applied_at=None,
|
|
files_created=None,
|
|
files_modified=None,
|
|
files_deleted=None,
|
|
)
|
|
|
|
plans_repo = _StubPlansRepository(current_plan=plan)
|
|
context.stub_plans_repo = plans_repo
|
|
static_context = _StaticContext(plans_repo)
|
|
|
|
def invalidate_plan_id() -> None:
|
|
plan.id = None
|
|
|
|
stub_uow = _SequencedUnitOfWork([(static_context, invalidate_plan_id)])
|
|
context.unit_of_work = stub_uow
|
|
settings = Settings()
|
|
context.plan_service = PlanService(
|
|
settings=settings,
|
|
unit_of_work=stub_uow,
|
|
ai_provider=None,
|
|
)
|
|
context.project = Project(
|
|
id=1,
|
|
name="stub-project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=plan.id,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="mock-gpt",
|
|
),
|
|
)
|
|
|
|
|
|
@when("I try to build the plan with the stubbed service")
|
|
def step_try_build_plan_stub(context: Context) -> None:
|
|
try:
|
|
context.plan_service.build_plan(context.project)
|
|
context.exception = None
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
context.exception = exc
|
|
|
|
|
|
@given("I replace the plan service repository with a stub that never assigns plan IDs")
|
|
def step_replace_plan_service_with_stub(context: Context) -> None:
|
|
plans_repo = _StubPlansRepository()
|
|
context.stub_plans_repo = plans_repo
|
|
static_context = _StaticContext(plans_repo)
|
|
stub_uow = _SequencedUnitOfWork([(static_context, None)])
|
|
context.unit_of_work = stub_uow
|
|
settings = Settings()
|
|
context.plan_service = PlanService(
|
|
settings=settings,
|
|
unit_of_work=stub_uow,
|
|
ai_provider=None,
|
|
)
|
|
context.project = Project(
|
|
id=7,
|
|
name="stub-project",
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="stub-model",
|
|
),
|
|
)
|
|
|
|
|
|
@when('I create a plan with stub prompt "{prompt}"')
|
|
def step_create_plan_with_stub_prompt(context: Context, prompt: str) -> None:
|
|
context.stub_created_plan = context.plan_service.create_plan(
|
|
context.project,
|
|
prompt,
|
|
None,
|
|
)
|
|
|
|
|
|
@then("the stub-created plan should remain non-current")
|
|
def step_verify_stub_plan_not_current(context: Context) -> None:
|
|
plan = getattr(context, "stub_created_plan", None)
|
|
assert isinstance(plan, Plan)
|
|
assert plan.current is False
|
|
assert plan.id is None
|
|
|
|
|
|
@then("the stub repository should not record a current plan")
|
|
def step_verify_stub_repository_not_called(context: Context) -> None:
|
|
repo = getattr(context, "stub_plans_repo", None)
|
|
assert isinstance(repo, _StubPlansRepository)
|
|
assert repo.set_current_calls == []
|
|
|
|
|
|
@when('I monitor max message updates for session "{session_id}"')
|
|
def step_monitor_max_messages(context: Context, session_id: str) -> None:
|
|
service = context.plan_service._memory_services.get(session_id)
|
|
assert service is not None, "Memory service must exist before monitoring"
|
|
original_setter = service.set_max_messages
|
|
monitored_setter = MagicMock(wraps=original_setter)
|
|
service.set_max_messages = monitored_setter # type: ignore[assignment]
|
|
context.monitored_service = service
|
|
context.monitored_setter = monitored_setter
|
|
|
|
|
|
@when(
|
|
'I request the same persistent memory service for session "{session_id}" with max messages {max_messages:d}'
|
|
)
|
|
def step_request_same_persistent_memory(
|
|
context: Context, session_id: str, max_messages: int
|
|
) -> None:
|
|
service = _request_memory_service_for_session(
|
|
context,
|
|
session_id,
|
|
persistent=True,
|
|
max_messages=max_messages,
|
|
)
|
|
context.reused_memory_service = service
|
|
|
|
|
|
@then(
|
|
'the session "{session_id}" memory should reuse the existing service without changing limits'
|
|
)
|
|
def step_verify_memory_service_reused(context: Context, session_id: str) -> None:
|
|
service = context.plan_service._memory_services.get(session_id)
|
|
monitored_service = getattr(context, "monitored_service", None)
|
|
monitored_setter = getattr(context, "monitored_setter", None)
|
|
assert service is monitored_service
|
|
assert monitored_service is not None
|
|
assert monitored_setter is not None
|
|
assert monitored_setter.call_count == 0
|
|
assert monitored_service.max_messages is not None
|
|
|
|
|
|
@given("LangSmith integration is enabled for plan service")
|
|
def step_enable_langsmith_integration(context: Context) -> None:
|
|
"""Reconfigure the plan service with LangSmith support enabled."""
|
|
|
|
if not hasattr(context, "unit_of_work"):
|
|
step_create_unit_of_work_plan(context)
|
|
|
|
settings = Settings()
|
|
settings.langsmith_enabled = True
|
|
settings.langsmith_project = "plan-service-tests"
|
|
settings.langsmith_api_key = "plan-service-api-key"
|
|
settings.langsmith_tags = ["global-langsmith"]
|
|
settings.langsmith_user_id = "plan-service-user"
|
|
|
|
original_builder = Settings.build_langsmith_config
|
|
|
|
def _forward_build_config(
|
|
self: Settings, *, tags=None, metadata=None, run_name=None
|
|
):
|
|
return original_builder(
|
|
self,
|
|
tags=tags,
|
|
metadata=metadata,
|
|
run_name=run_name,
|
|
)
|
|
|
|
builder_patch = patch.object(
|
|
Settings,
|
|
"build_langsmith_config",
|
|
autospec=True,
|
|
side_effect=_forward_build_config,
|
|
)
|
|
builder_mock = builder_patch.start()
|
|
cleanup = getattr(context, "add_cleanup", None)
|
|
if callable(cleanup):
|
|
cleanup(builder_patch.stop)
|
|
else: # pragma: no cover - behave contexts should supply add_cleanup
|
|
context._langsmith_builder_patch = builder_patch # type: ignore[attr-defined]
|
|
context.langsmith_builder_mock = builder_mock
|
|
|
|
context.plan_service = PlanService(
|
|
settings=settings,
|
|
unit_of_work=context.unit_of_work,
|
|
ai_provider=None,
|
|
)
|
|
|
|
|
|
@when(
|
|
'I prepare a LangSmith config for project "{project_name}" and plan "{plan_name}"'
|
|
)
|
|
def step_prepare_langsmith_config(
|
|
context: Context, project_name: str, plan_name: str
|
|
) -> None:
|
|
"""Build and prepare a LangSmith config to capture metadata and tags."""
|
|
|
|
project = Project(
|
|
id=101,
|
|
name=project_name,
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="langsmith",
|
|
),
|
|
)
|
|
plan = Plan(
|
|
id=202,
|
|
project_id=project.id,
|
|
name=plan_name,
|
|
prompt="LangSmith coverage",
|
|
status=PlanStatus.PENDING,
|
|
current=True,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
build=None,
|
|
build_started_at=None,
|
|
build_completed_at=None,
|
|
model_used=None,
|
|
token_count=None,
|
|
result=None,
|
|
applied_at=None,
|
|
files_created=None,
|
|
files_modified=None,
|
|
files_deleted=None,
|
|
)
|
|
config = context.plan_service._prepare_langsmith_config(
|
|
project,
|
|
plan,
|
|
run_name="coverage",
|
|
tags=["custom-tag"],
|
|
metadata={"env": "test"},
|
|
)
|
|
context.langsmith_project = project
|
|
context.langsmith_plan = plan
|
|
context.langsmith_prepared_config = config
|
|
builder_mock = getattr(context, "langsmith_builder_mock", None)
|
|
if builder_mock is not None:
|
|
context.langsmith_builder_call = builder_mock.call_args
|
|
|
|
|
|
@when('I prepare a LangSmith config without plan metadata for project "{project_name}"')
|
|
def step_prepare_langsmith_config_without_plan(
|
|
context: Context, project_name: str
|
|
) -> None:
|
|
"""Prepare a LangSmith config when plan metadata is unavailable."""
|
|
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
settings = context.plan_service.settings
|
|
settings.langsmith_tags = []
|
|
settings.langsmith_user_id = None
|
|
|
|
project = Project(
|
|
id=None,
|
|
name=project_name,
|
|
path=context.temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="langsmith",
|
|
),
|
|
)
|
|
config = context.plan_service._prepare_langsmith_config(
|
|
project,
|
|
None,
|
|
run_name="coverage-minimal",
|
|
tags=None,
|
|
metadata=None,
|
|
)
|
|
context.langsmith_project = project
|
|
context.langsmith_plan = None
|
|
context.langsmith_prepared_config = config
|
|
builder_mock = getattr(context, "langsmith_builder_mock", None)
|
|
if builder_mock is not None:
|
|
context.langsmith_builder_call = builder_mock.call_args
|
|
|
|
|
|
@then(
|
|
'the LangSmith builder should receive metadata for project "{project_name}" and plan "{plan_name}"'
|
|
)
|
|
def step_assert_langsmith_builder_metadata(
|
|
context: Context, project_name: str, plan_name: str
|
|
) -> None:
|
|
"""Verify the builder received merged metadata and tags."""
|
|
|
|
call = getattr(context, "langsmith_builder_call", None)
|
|
assert call is not None, "LangSmith builder was not invoked"
|
|
_, kwargs = call
|
|
metadata = kwargs.get("metadata", {})
|
|
tags = kwargs.get("tags", [])
|
|
project = getattr(context, "langsmith_project", None)
|
|
plan = getattr(context, "langsmith_plan", None)
|
|
assert metadata.get("project_name") == project_name
|
|
assert metadata.get("plan_name") == plan_name
|
|
assert metadata.get("env") == "test"
|
|
assert metadata.get("project_id") == getattr(project, "id", None)
|
|
assert metadata.get("plan_id") == getattr(plan, "id", None)
|
|
assert metadata.get("user_id") == "plan-service-user"
|
|
assert "service:plan" in tags
|
|
ctr_project = getattr(project, "id", None)
|
|
assert f"project:{ctr_project}" in tags
|
|
assert f"plan:{getattr(plan, 'id', None)}" in tags
|
|
assert "global-langsmith" in tags
|
|
assert "custom-tag" in tags
|
|
|
|
|
|
@then(
|
|
'the LangSmith builder should receive only base metadata for project "{project_name}"'
|
|
)
|
|
def step_assert_langsmith_builder_base_metadata(
|
|
context: Context, project_name: str
|
|
) -> None:
|
|
"""Ensure only project metadata is provided when no plan is available."""
|
|
|
|
call = getattr(context, "langsmith_builder_call", None)
|
|
assert call is not None, "LangSmith builder was not invoked"
|
|
_, kwargs = call
|
|
metadata = kwargs.get("metadata", {})
|
|
tags = kwargs.get("tags", [])
|
|
assert metadata.get("project_name") == project_name
|
|
assert metadata.get("project_id") is None
|
|
extra_metadata_keys = set(metadata.keys()) - {"project_id", "project_name"}
|
|
assert not extra_metadata_keys, (
|
|
f"Unexpected metadata keys present: {sorted(extra_metadata_keys)}"
|
|
)
|
|
assert tags == ["service:plan"], f"Expected only base service tag, got {tags}"
|
|
|
|
|
|
@then("the prepared LangSmith config should include a generated thread id")
|
|
def step_assert_langsmith_thread_id(context: Context) -> None:
|
|
config = getattr(context, "langsmith_prepared_config", None)
|
|
assert config, "LangSmith config was not prepared"
|
|
configurable = config.get("configurable")
|
|
assert isinstance(configurable, dict)
|
|
thread_id = configurable.get("thread_id")
|
|
assert isinstance(thread_id, str)
|
|
assert thread_id.startswith("plan-service-")
|
|
|
|
|
|
@given("I have a plan service with a failing AI provider")
|
|
def step_plan_service_with_failing_provider(context: Context) -> None:
|
|
"""Configure the plan service to surface provider failures."""
|
|
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
|
|
if not hasattr(context, "env_vars_to_clean"):
|
|
context.env_vars_to_clean = []
|
|
if "CLEVERAGENTS_TESTING_USE_MOCK_AI" not in context.env_vars_to_clean:
|
|
context.env_vars_to_clean.append("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
|
|
failing_provider = MagicMock()
|
|
failing_provider.generate_changes.return_value = ProviderResponse(
|
|
changes=[],
|
|
model_used="mock-gpt",
|
|
token_count=0,
|
|
error_message="mock provider failure",
|
|
)
|
|
context.plan_service = PlanService(
|
|
settings=Settings(),
|
|
unit_of_work=context.unit_of_work,
|
|
ai_provider=failing_provider,
|
|
)
|
|
context.mock_provider = failing_provider
|
|
|
|
|
|
@when("I attempt to build the plan and the provider returns an error")
|
|
def step_build_plan_with_provider_error(context: Context) -> None:
|
|
"""Invoke build_plan and capture the provider error."""
|
|
|
|
try:
|
|
context.plan_service.build_plan(context.project)
|
|
context.exception = None
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
context.exception = exc
|
|
|
|
|
|
@when("I run the auto debug build for the project")
|
|
def step_run_auto_debug_without_plan(context: Context) -> None:
|
|
"""Run auto_debug_build while capturing exceptions."""
|
|
|
|
try:
|
|
context.auto_debug_result = context.plan_service.auto_debug_build(
|
|
context.project
|
|
)
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.auto_debug_result = None
|
|
context.exception = exc
|
|
|
|
|
|
@given("auto debug retries should run with a failing build pipeline")
|
|
def step_force_auto_debug_failure(context: Context) -> None:
|
|
"""Mark that auto_debug_build should simulate failing build attempts."""
|
|
|
|
context.force_auto_debug_failure = True
|
|
|
|
|
|
@when("I run the auto debug build with max attempts {attempts:d}")
|
|
def step_run_auto_debug_with_attempts(context: Context, attempts: int) -> None:
|
|
"""Execute auto_debug_build, optionally forcing failures for coverage."""
|
|
|
|
def failing_build(
|
|
self: PlanService,
|
|
project: Project,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> list[Change]:
|
|
raise RuntimeError("Simulated auto debug failure")
|
|
|
|
agent_path = "cleveragents.agents.AutoDebugAgent"
|
|
|
|
try:
|
|
if getattr(context, "force_auto_debug_failure", False):
|
|
with (
|
|
patch.object(PlanService, "build_plan", side_effect=failing_build),
|
|
patch(agent_path) as agent_cls,
|
|
):
|
|
agent_instance = MagicMock()
|
|
agent_instance.invoke.return_value = {
|
|
"result": {"success": False, "fix": {}},
|
|
}
|
|
agent_cls.return_value = agent_instance
|
|
context.auto_debug_result = context.plan_service.auto_debug_build(
|
|
context.project, max_attempts=attempts
|
|
)
|
|
else:
|
|
context.auto_debug_result = context.plan_service.auto_debug_build(
|
|
context.project, max_attempts=attempts
|
|
)
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.auto_debug_result = None
|
|
context.exception = exc
|
|
|
|
|
|
@when("I run the auto debug build with a retrying success")
|
|
def step_run_auto_debug_retry_success(context: Context) -> None:
|
|
assert hasattr(context, "plan_service"), "PlanService is not configured"
|
|
current_plan = getattr(context, "current_plan", None)
|
|
assert current_plan is not None, "Current plan is not configured"
|
|
|
|
call_count = {"count": 0}
|
|
|
|
def build_plan_side_effect(
|
|
self: PlanService,
|
|
project: Project,
|
|
progress_callback: Callable[[int], None] | None = None,
|
|
) -> list[Change]:
|
|
call_count["count"] += 1
|
|
if call_count["count"] == 1:
|
|
raise RuntimeError("first failure")
|
|
return [
|
|
Change(
|
|
id=None,
|
|
plan_id=current_plan.id or 0,
|
|
file_path="after_retry.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# retry success",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
]
|
|
|
|
agent_path = "cleveragents.agents.AutoDebugAgent"
|
|
with (
|
|
patch.object(PlanService, "build_plan", side_effect=build_plan_side_effect),
|
|
patch(agent_path) as agent_cls,
|
|
):
|
|
agent_instance = MagicMock()
|
|
agent_instance.invoke.return_value = {"result": {"success": False, "fix": {}}}
|
|
agent_cls.return_value = agent_instance
|
|
context.auto_debug_result = context.plan_service.auto_debug_build(
|
|
context.project, max_attempts=2
|
|
)
|
|
|
|
with context.unit_of_work.transaction() as ctx:
|
|
context.latest_debug_attempts = ctx.debug_attempts.get_for_plan(current_plan.id)
|
|
|
|
|
|
@then('the auto debug result should indicate failure with error message "{message}"')
|
|
def step_assert_auto_debug_failure(context: Context, message: str) -> None:
|
|
"""Validate that auto_debug_build returned a failure tuple."""
|
|
|
|
result = getattr(context, "auto_debug_result", None)
|
|
assert result is not None, "Auto debug result was not captured"
|
|
success, changes, error_message = result
|
|
assert success is False
|
|
assert isinstance(changes, list)
|
|
assert isinstance(error_message, str)
|
|
assert message in error_message
|
|
|
|
|
|
@then("the auto debug attempt should be marked successful after retry")
|
|
def step_assert_auto_debug_retry_success(context: Context) -> None:
|
|
result = getattr(context, "auto_debug_result", None)
|
|
assert result is not None, "Auto debug result was not captured"
|
|
success, changes, error_message = result
|
|
assert success is True
|
|
assert isinstance(changes, list)
|
|
assert error_message is None
|
|
|
|
attempts = getattr(context, "latest_debug_attempts", []) or []
|
|
assert attempts, "No debug attempts were recorded"
|
|
assert any(attempt.success for attempt in attempts), (
|
|
"Debug attempt was not marked successful"
|
|
)
|
|
|
|
|
|
@when("I try to stream plan generation without an AI provider")
|
|
def step_stream_generate_without_provider(context: Context) -> None:
|
|
"""Attempt to consume the streaming generator without configuring an AI provider."""
|
|
|
|
import asyncio
|
|
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
|
|
if not hasattr(context, "env_vars_to_clean"):
|
|
context.env_vars_to_clean = []
|
|
if "CLEVERAGENTS_TESTING_USE_MOCK_AI" not in context.env_vars_to_clean:
|
|
context.env_vars_to_clean.append("CLEVERAGENTS_TESTING_USE_MOCK_AI")
|
|
|
|
# Ensure the plan service cannot resolve a provider regardless of environment
|
|
context.plan_service.ai_provider = None
|
|
context.plan_service._provider_registry = _NoProviderRegistry()
|
|
|
|
async def _consume_stream() -> None:
|
|
generator = context.plan_service.generate_plan_streaming(
|
|
context.project,
|
|
description="Stream coverage scenario",
|
|
)
|
|
await generator.__anext__()
|
|
|
|
try:
|
|
asyncio.run(_consume_stream())
|
|
context.exception = None
|
|
except Exception as exc:
|
|
context.exception = exc
|