1434 lines
53 KiB
Python
1434 lines
53 KiB
Python
"""Step definitions for service-level tests."""
|
|
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.application.container import get_container, override_providers
|
|
from cleveragents.application.services.context_service import ContextService
|
|
from cleveragents.application.services.plan_service import PlanService
|
|
from cleveragents.application.services.project_service import ProjectService
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.core.exceptions import PlanError, ValidationError
|
|
|
|
|
|
def add_cleanup(context: Context, handler):
|
|
"""Add a cleanup handler to the context."""
|
|
if not hasattr(context, "_cleanup_handlers"):
|
|
context._cleanup_handlers = []
|
|
context._cleanup_handlers.append(handler)
|
|
|
|
|
|
@given("I have a project service")
|
|
def step_have_project_service(context: Context) -> None:
|
|
"""Create a project service instance."""
|
|
import uuid
|
|
|
|
settings = Settings()
|
|
|
|
# Create a unique temp dir for this test
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_project_")
|
|
|
|
# Create a unique database file
|
|
db_file = Path(context.test_dir) / f"test_{uuid.uuid4().hex}.db"
|
|
db_url = f"sqlite:///{db_file}"
|
|
|
|
# Create unit of work instead of passing db_url directly
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
|
|
unit_of_work = UnitOfWork(db_url)
|
|
|
|
context.project_service = ProjectService(settings, unit_of_work)
|
|
context.project_service.search_root = Path(context.test_dir)
|
|
|
|
def cleanup():
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
|
|
@when('I initialize a project "{name}" at "{path}"')
|
|
def step_initialize_project(context: Context, name: str, path: str) -> None:
|
|
"""Initialize a project."""
|
|
if path == "/tmp/test":
|
|
path = context.test_dir
|
|
try:
|
|
context.project = context.project_service.initialize_project(
|
|
name=name, path=Path(path), force=False
|
|
)
|
|
context.error = None
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@when('I initialize a project "{name}" at "{path}" with force')
|
|
def step_initialize_project_with_force(context: Context, name: str, path: str) -> None:
|
|
"""Initialize a project with force flag."""
|
|
if path == "/tmp/test":
|
|
path = context.test_dir
|
|
context.project = context.project_service.initialize_project(
|
|
name=name, path=Path(path), force=True
|
|
)
|
|
context.error = None
|
|
|
|
|
|
@when('I try to initialize a project "{name}" at "{path}" without force')
|
|
def step_try_initialize_without_force(context: Context, name: str, path: str) -> None:
|
|
"""Try to initialize a project without force."""
|
|
if path == "/tmp/test":
|
|
path = context.test_dir
|
|
try:
|
|
context.project = context.project_service.initialize_project(
|
|
name=name, path=Path(path), force=False
|
|
)
|
|
context.error = None
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@given('I have initialized a service project "{name}" at "{path}"')
|
|
def step_have_initialized_service_project(
|
|
context: Context, name: str, path: str
|
|
) -> None:
|
|
"""Initialize a project as a given for service tests."""
|
|
if path == "/tmp/test":
|
|
path = context.test_dir
|
|
context.project = context.project_service.initialize_project(
|
|
name=name, path=Path(path), force=False
|
|
)
|
|
|
|
|
|
@then("the project should be created successfully")
|
|
def step_project_created_successfully(context: Context) -> None:
|
|
"""Verify project was created."""
|
|
assert context.project is not None, "Project was not created"
|
|
assert context.error is None, f"Got error: {context.error}"
|
|
|
|
|
|
@then('the project name should be "{name}"')
|
|
def step_project_name_should_be(context: Context, name: str) -> None:
|
|
"""Verify project name."""
|
|
assert context.project.name == name, (
|
|
f"Expected name {name}, got {context.project.name}"
|
|
)
|
|
|
|
|
|
@then('the project path should be "{path}"')
|
|
def step_project_path_should_be(context: Context, path: str) -> None:
|
|
"""Verify project path."""
|
|
if path == "/tmp/test":
|
|
path = context.test_dir
|
|
expected_path = Path(path)
|
|
assert context.project.path == expected_path, (
|
|
f"Expected path {expected_path}, got {context.project.path}"
|
|
)
|
|
|
|
|
|
@then('the .cleveragents directory should exist at "{path}"')
|
|
def step_cleveragents_dir_should_exist(context: Context, path: str) -> None:
|
|
"""Verify .cleveragents directory exists."""
|
|
if path.startswith("/tmp/test"):
|
|
path = path.replace("/tmp/test", context.test_dir)
|
|
assert Path(path).exists(), f"Directory {path} does not exist"
|
|
|
|
|
|
@then("I should get a ValidationError")
|
|
def step_should_get_validation_error(context: Context) -> None:
|
|
"""Verify we got a ValidationError."""
|
|
assert isinstance(context.error, ValidationError), (
|
|
f"Expected ValidationError, got {type(context.error)}"
|
|
)
|
|
|
|
|
|
@then('the error message should contain "{text}"')
|
|
def step_error_should_contain(context: Context, text: str) -> None:
|
|
"""Verify error message contains text."""
|
|
assert text in str(context.error), (
|
|
f"Error message doesn't contain '{text}': {context.error}"
|
|
)
|
|
|
|
|
|
# Context Service steps
|
|
@given("I have a context service")
|
|
def step_have_context_service(context: Context) -> None:
|
|
"""Create a context service instance."""
|
|
import os
|
|
import uuid
|
|
|
|
settings = Settings()
|
|
|
|
# Create a unique temp dir for this test
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_context_")
|
|
|
|
# Create .cleveragents directory for context storage
|
|
cleveragents_dir = Path(context.test_dir) / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
# Create a unique database file
|
|
db_file = Path(context.test_dir) / f"test_{uuid.uuid4().hex}.db"
|
|
db_url = f"sqlite:///{db_file}"
|
|
|
|
# Change to test directory so service finds the .cleveragents dir
|
|
context.original_cwd = os.getcwd()
|
|
os.chdir(context.test_dir)
|
|
|
|
# Create unit of work instead of passing db_url directly
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
|
|
unit_of_work = UnitOfWork(db_url)
|
|
|
|
# Create services we need
|
|
context.context_service = ContextService(settings, unit_of_work)
|
|
|
|
# Also create a project service to initialize a test project
|
|
from cleveragents.application.services.project_service import ProjectService
|
|
|
|
project_service = ProjectService(settings, unit_of_work)
|
|
|
|
# Initialize a test project
|
|
context.test_project = project_service.initialize_project(
|
|
name="test-project", path=Path(context.test_dir), force=True
|
|
)
|
|
|
|
# Also create a plan service and create a plan
|
|
from cleveragents.application.services.plan_service import PlanService
|
|
|
|
plan_service = PlanService(settings, unit_of_work)
|
|
context.test_plan = plan_service.create_plan(
|
|
project=context.test_project, prompt="Test plan for context"
|
|
)
|
|
|
|
# Return to original directory
|
|
os.chdir(context.original_cwd)
|
|
|
|
def cleanup():
|
|
if hasattr(context, "original_cwd"):
|
|
os.chdir(context.original_cwd)
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
|
|
@given("I have a context service without a plan")
|
|
def step_have_context_service_without_plan(context: Context) -> None:
|
|
"""Create a context service without creating a plan."""
|
|
import os
|
|
import uuid
|
|
|
|
settings = Settings()
|
|
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_context_")
|
|
|
|
cleveragents_dir = Path(context.test_dir) / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
db_file = Path(context.test_dir) / f"test_{uuid.uuid4().hex}.db"
|
|
db_url = f"sqlite:///{db_file}"
|
|
|
|
context.original_cwd = os.getcwd()
|
|
os.chdir(context.test_dir)
|
|
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
|
|
unit_of_work = UnitOfWork(db_url)
|
|
|
|
context.context_service = ContextService(settings, unit_of_work)
|
|
|
|
from cleveragents.application.services.project_service import ProjectService
|
|
|
|
project_service = ProjectService(settings, unit_of_work)
|
|
context.test_project = project_service.initialize_project(
|
|
name="test-project", path=Path(context.test_dir), force=True
|
|
)
|
|
|
|
if context.test_project.id is not None:
|
|
with context.context_service.unit_of_work.transaction() as txn:
|
|
plans = txn.plans.get_all_for_project(context.test_project.id)
|
|
for plan in plans:
|
|
plan.current = False
|
|
txn.plans.update(plan)
|
|
context.test_project.current_plan_id = None
|
|
txn.projects.update(context.test_project)
|
|
|
|
context.test_project.current_plan_id = None
|
|
|
|
os.chdir(context.original_cwd)
|
|
|
|
def cleanup():
|
|
if hasattr(context, "original_cwd"):
|
|
os.chdir(context.original_cwd)
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
|
|
@when('I try to add "{filename}" to context')
|
|
def step_try_add_file_to_context(context: Context, filename: str) -> None:
|
|
"""Attempt to add a file to context and capture errors."""
|
|
test_file = Path(context.test_dir) / filename
|
|
if not test_file.exists():
|
|
test_file.write_text("test content")
|
|
try:
|
|
context.context_service.add_to_context(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
context.error = None
|
|
except Exception as exc:
|
|
context.error = exc
|
|
|
|
|
|
@then("a plan error should be raised for missing plan")
|
|
def step_plan_error_missing_plan(context: Context) -> None:
|
|
"""Ensure a PlanError was raised when no plan exists."""
|
|
assert isinstance(context.error, PlanError), (
|
|
f"Expected PlanError, got {type(context.error)}"
|
|
)
|
|
assert "No current plan" in str(context.error), str(context.error)
|
|
|
|
|
|
@when('I add file "{filename}" to context again')
|
|
def step_add_file_to_context_again(context: Context, filename: str) -> None:
|
|
"""Add a file to context that already exists."""
|
|
test_file = Path(context.test_dir) / filename
|
|
added_files, already_in_context = context.context_service.add_to_context(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
context.added_files = added_files
|
|
context.already_in_context = already_in_context
|
|
context.add_result = (added_files, already_in_context)
|
|
context.last_added_filename = filename
|
|
|
|
|
|
@then("the file should be reported as already in context")
|
|
def step_file_reported_already(context: Context) -> None:
|
|
"""Verify duplicate additions land in the already-in-context list."""
|
|
assert hasattr(context, "already_in_context"), "No duplicate tracking recorded"
|
|
filenames = {Path(p).name for p in context.already_in_context}
|
|
expected = getattr(context, "last_added_filename", None)
|
|
assert expected in filenames, (
|
|
f"Expected {expected} in already_in_context, found {filenames}"
|
|
)
|
|
|
|
|
|
@when("I add the directory to context without recursion")
|
|
def step_add_directory_without_recursion(context: Context) -> None:
|
|
"""Add a directory to context without descending into subdirectories."""
|
|
target_dir = getattr(context, "dir_under_test", None)
|
|
if target_dir is None:
|
|
target_dir = context.test_subdir
|
|
added_files, already_in_context = context.context_service.add_to_context(
|
|
project=context.test_project, path=target_dir, recursive=False
|
|
)
|
|
context.added_files = added_files
|
|
context.already_in_context = already_in_context
|
|
context.add_result = (added_files, already_in_context)
|
|
|
|
|
|
@then("{count:d} top-level file should be added from the directory")
|
|
def step_top_level_file_count(context: Context, count: int) -> None:
|
|
"""Ensure the expected number of files were added without recursion."""
|
|
assert len(context.added_files) == count, (
|
|
f"Expected {count} files, got {len(context.added_files)}"
|
|
)
|
|
|
|
|
|
@then('the nested file "{filename}" should not be added to context')
|
|
def step_nested_file_not_added(context: Context, filename: str) -> None:
|
|
"""Verify nested files are excluded when not recursing."""
|
|
added_names = {Path(p).name for p in context.added_files}
|
|
assert filename not in added_names, (
|
|
f"Nested file {filename} should not be added: {added_names}"
|
|
)
|
|
|
|
|
|
@then('the ignored file "{filename}" should not be added to context')
|
|
def step_ignored_file_not_added(context: Context, filename: str) -> None:
|
|
"""Verify ignored patterns are skipped when scanning directories."""
|
|
added_names = {Path(p).name for p in context.added_files}
|
|
assert filename not in added_names, (
|
|
f"Ignored file {filename} should not be added: {added_names}"
|
|
)
|
|
|
|
|
|
@when('I add file "{filename}" to context')
|
|
def step_add_file_to_context(context: Context, filename: str) -> None:
|
|
"""Add a file to context."""
|
|
test_file = Path(context.test_dir) / filename
|
|
if not test_file.exists():
|
|
test_file.write_text("test content")
|
|
added_files, already_in_context = context.context_service.add_to_context(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
context.added_files = added_files
|
|
context.already_in_context = already_in_context
|
|
context.add_result = (added_files, already_in_context)
|
|
|
|
|
|
@given("the context service max file size is set to {size:d} bytes")
|
|
def step_set_max_file_size(context: Context, size: int) -> None:
|
|
"""Override the per-file size guardrail."""
|
|
context.context_service.max_file_size = size
|
|
|
|
|
|
@given("the context service max context size is set to {size:d} bytes")
|
|
def step_set_max_context_size(context: Context, size: int) -> None:
|
|
"""Override the aggregate context size guardrail."""
|
|
context.context_service.max_context_size = size
|
|
|
|
|
|
@given('I have a test file "{filename}" of size {size:d} bytes')
|
|
def step_have_test_file_of_size(context: Context, filename: str, size: int) -> None:
|
|
"""Create a test file with a specific size."""
|
|
if not hasattr(context, "test_dir"):
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_context_")
|
|
|
|
def cleanup():
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
test_file = Path(context.test_dir) / filename
|
|
test_file.write_text("a" * size)
|
|
|
|
|
|
@when('I add existing file "{filename}" to context')
|
|
def step_add_existing_file(context: Context, filename: str) -> None:
|
|
"""Add a previously created file without rewriting it."""
|
|
test_file = Path(context.test_dir) / filename
|
|
added_files, already_in_context = context.context_service.add_to_context(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
context.added_files = added_files
|
|
context.already_in_context = already_in_context
|
|
context.add_result = (added_files, already_in_context)
|
|
context.last_added_filename = filename
|
|
|
|
|
|
@given("file reads will fail during addition")
|
|
def step_patch_read_text(context: Context) -> None:
|
|
"""Patch Path.read_text to simulate unreadable files."""
|
|
original_read_text = Path.read_text
|
|
|
|
def failing_read_text(self: Path, *args, **kwargs):
|
|
if hasattr(context, "test_dir") and str(self).startswith(context.test_dir):
|
|
raise OSError("Simulated read failure")
|
|
return original_read_text(self, *args, **kwargs)
|
|
|
|
patcher = patch("pathlib.Path.read_text", new=failing_read_text)
|
|
patcher.start()
|
|
context.read_text_patcher = patcher
|
|
|
|
def cleanup():
|
|
patcher.stop()
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
|
|
@then("no files should be added by the last operation")
|
|
def step_no_files_added(context: Context) -> None:
|
|
"""Ensure the most recent add_to_context call added nothing."""
|
|
assert len(getattr(context, "added_files", [])) == 0, (
|
|
f"Expected no files added, got {context.added_files}"
|
|
)
|
|
|
|
|
|
@then('only "{filename}" should remain in the context')
|
|
def step_only_specific_file_in_context(context: Context, filename: str) -> None:
|
|
"""Ensure only the provided file remains tracked."""
|
|
files = context.context_service.list_context(project=context.test_project)
|
|
names = {Path(f.path).name for f in files}
|
|
assert names == {filename}, f"Expected only {filename}, found {names}"
|
|
|
|
|
|
@then("the file should be added to service context")
|
|
def step_file_should_be_added(context: Context) -> None:
|
|
"""Verify file was added."""
|
|
assert len(context.added_files) > 0, "No files were added"
|
|
|
|
|
|
@given('I have a special filesystem entry "{name}"')
|
|
def step_have_special_filesystem_entry(context: Context, name: str) -> None:
|
|
"""Create a special filesystem entry such as a FIFO."""
|
|
if not hasattr(context, "test_dir"):
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_context_")
|
|
|
|
def cleanup_dir():
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup_dir)
|
|
|
|
special_path = Path(context.test_dir) / name
|
|
if special_path.exists():
|
|
special_path.unlink()
|
|
os.mkfifo(special_path)
|
|
context.special_path = special_path
|
|
|
|
def cleanup_entry() -> None:
|
|
if special_path.exists():
|
|
special_path.unlink()
|
|
|
|
add_cleanup(context, cleanup_entry)
|
|
|
|
|
|
@when('I add special entry "{name}" to context')
|
|
def step_add_special_entry(context: Context, name: str) -> None:
|
|
"""Attempt to add a special filesystem entry to context."""
|
|
special_path = getattr(context, "special_path", Path(context.test_dir) / name)
|
|
added_files, already_in_context = context.context_service.add_to_context(
|
|
project=context.test_project, path=special_path
|
|
)
|
|
context.added_files = added_files
|
|
context.already_in_context = already_in_context
|
|
context.add_result = (added_files, already_in_context)
|
|
|
|
|
|
@when("I add the directory to context recursively again")
|
|
def step_add_directory_recursively_again(context: Context) -> None:
|
|
"""Re-run recursive directory addition to surface duplicates."""
|
|
step_add_directory_recursively(context)
|
|
|
|
|
|
@when("I add the directory to context without recursion again")
|
|
def step_add_directory_without_recursion_again(context: Context) -> None:
|
|
"""Re-run non-recursive directory addition to surface duplicates."""
|
|
step_add_directory_without_recursion(context)
|
|
|
|
|
|
@then("the last addition should report {count:d} duplicates")
|
|
def step_last_addition_should_report_duplicates(context: Context, count: int) -> None:
|
|
"""Verify the most recent add_to_context reported duplicates."""
|
|
duplicates = getattr(context, "already_in_context", [])
|
|
assert len(duplicates) == count, (
|
|
f"Expected {count} duplicates, found {len(duplicates)}: {duplicates}"
|
|
)
|
|
|
|
|
|
@when('I add "{filename}" with an unsaved plan')
|
|
def step_add_with_unsaved_plan(context: Context, filename: str) -> None:
|
|
"""Try to add a file when the plan lacks an identifier."""
|
|
assert hasattr(context, "test_plan"), "Test plan not initialized"
|
|
test_file = Path(context.test_dir) / filename
|
|
if not test_file.exists():
|
|
test_file.write_text("test content")
|
|
|
|
with context.context_service.unit_of_work.transaction() as txn:
|
|
plan = context.test_plan.model_copy(deep=True)
|
|
plan.id = None
|
|
result = context.context_service._add_file_to_context(txn, plan, test_file)
|
|
|
|
context.last_add_status = result
|
|
context.added_files = []
|
|
context.already_in_context = []
|
|
context.add_result = ([], [])
|
|
|
|
|
|
@then('the context listings should not contain "{filename}"')
|
|
def step_list_should_not_contain(context: Context, filename: str) -> None:
|
|
"""Ensure a given filename is absent from collected listings."""
|
|
file_names: set[str] = set()
|
|
if getattr(context, "context_files", None):
|
|
files = context.context_files
|
|
if files:
|
|
if hasattr(files[0], "path"):
|
|
file_names = {Path(str(f.path)).name for f in files}
|
|
else:
|
|
file_names = {Path(str(f["path"])).name for f in files}
|
|
elif getattr(context, "listed_files", None):
|
|
file_names = {Path(path).name for path in context.listed_files}
|
|
|
|
assert filename not in file_names, (
|
|
f'Filename "{filename}" unexpectedly present in {sorted(file_names)}'
|
|
)
|
|
|
|
|
|
@then("the removed count should be 0")
|
|
def step_removed_count_zero(context: Context) -> None:
|
|
"""Verify that no context entries were removed."""
|
|
assert getattr(context, "removed_count", 0) == 0, (
|
|
f"Expected 0 removals, got {getattr(context, 'removed_count', None)}"
|
|
)
|
|
|
|
|
|
@when("I clear the context and record the removed count")
|
|
def step_clear_context_and_record(context: Context) -> None:
|
|
"""Clear context and store the removed item count."""
|
|
context.cleared_count = context.context_service.clear_context(
|
|
project=context.test_project
|
|
)
|
|
|
|
|
|
@then("the recorded removal count should be 0")
|
|
def step_recorded_removal_zero(context: Context) -> None:
|
|
"""Verify that a recorded clear operation removed nothing."""
|
|
assert getattr(context, "cleared_count", 0) == 0, (
|
|
f"Expected 0 removals, got {getattr(context, 'cleared_count', None)}"
|
|
)
|
|
|
|
|
|
@given("the test project loses its identifier")
|
|
def step_project_loses_identifier(context: Context) -> None:
|
|
"""Remove the project identifier to exercise guard clauses."""
|
|
assert hasattr(context, "test_project"), "Test project not initialized"
|
|
context.test_project.id = None
|
|
|
|
|
|
@when("I show context content")
|
|
def step_show_context_content(context: Context) -> None:
|
|
"""Capture the current context content mapping."""
|
|
context.shown_content = context.context_service.show_context_content(
|
|
project=context.test_project
|
|
)
|
|
|
|
|
|
@then('the shown context content should include "{filename}"')
|
|
def step_shown_content_includes_filename(context: Context, filename: str) -> None:
|
|
"""Ensure the shown context includes the requested filename."""
|
|
shown = getattr(context, "shown_content", {})
|
|
file_names = {Path(path).name for path in shown}
|
|
assert filename in file_names, (
|
|
f'Expected "{filename}" in {sorted(file_names)} from shown content'
|
|
)
|
|
|
|
|
|
@then('fetching context content for "{filename}" should return stored content')
|
|
def step_fetch_context_content_matches(context: Context, filename: str) -> None:
|
|
"""Verify content lookup returns the stored file text."""
|
|
test_file = Path(context.test_dir) / filename
|
|
expected = test_file.read_text()
|
|
content = context.context_service.get_context_content(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
assert content == expected, (
|
|
f"Expected stored content for {filename}, got {content!r}"
|
|
)
|
|
|
|
|
|
@then('fetching context content for "{filename}" should return nothing')
|
|
def step_fetch_context_content_none(context: Context, filename: str) -> None:
|
|
"""Ensure content lookup returns None when not present."""
|
|
test_file = Path(context.test_dir) / filename
|
|
content = context.context_service.get_context_content(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
assert content is None, f"Expected no content for {filename}, got {content!r}"
|
|
|
|
|
|
@when("I list context files for the current project")
|
|
def step_list_files_for_current_project(context: Context) -> None:
|
|
"""List context files using an explicit project argument."""
|
|
context.listed_files = context.context_service.list_files(
|
|
project=context.test_project
|
|
)
|
|
|
|
|
|
@given("the container project service returns no project")
|
|
def step_container_project_service_returns_none(context: Context) -> None:
|
|
"""Override the container project service to return None."""
|
|
|
|
class NullProjectService:
|
|
def get_current_project(self):
|
|
return None
|
|
|
|
override = NullProjectService()
|
|
override_providers(project_service=override)
|
|
container = get_container()
|
|
|
|
def cleanup() -> None:
|
|
if hasattr(container, "project_service"):
|
|
container.project_service.reset_override()
|
|
|
|
add_cleanup(context, cleanup)
|
|
context.project_service_override = override
|
|
|
|
|
|
# Plan Service steps
|
|
@given("I have a plan service")
|
|
def step_have_plan_service(context: Context) -> None:
|
|
"""Create a plan service instance."""
|
|
import os
|
|
import uuid
|
|
|
|
settings = Settings()
|
|
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
|
|
|
|
# Always create a fresh temp dir for isolation
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_plan_")
|
|
|
|
# Create .cleveragents directory for plan storage
|
|
cleveragents_dir = Path(context.test_dir) / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
# Create a unique database file
|
|
db_file = Path(context.test_dir) / f"test_{uuid.uuid4().hex}.db"
|
|
db_url = f"sqlite:///{db_file}"
|
|
|
|
# Store original directory
|
|
context.original_cwd = os.getcwd()
|
|
|
|
# Change to test directory and stay there for the service to work properly
|
|
os.chdir(context.test_dir)
|
|
|
|
# Create unit of work instead of passing db_url directly
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
|
|
unit_of_work = UnitOfWork(db_url)
|
|
|
|
# Get the AI provider from container (which has been set up by environment.py)
|
|
from cleveragents.application.container import get_container
|
|
|
|
container = get_container()
|
|
ai_provider = container.ai_provider()
|
|
|
|
context.plan_service = PlanService(settings, unit_of_work, ai_provider)
|
|
context.plan_service.actor_service.ensure_default_mock_actor(force=True)
|
|
|
|
# Also create a project service to initialize a test project
|
|
from cleveragents.application.services.project_service import ProjectService
|
|
|
|
project_service = ProjectService(settings, unit_of_work)
|
|
|
|
# Initialize a test project
|
|
context.test_project = project_service.initialize_project(
|
|
name="test-project", path=Path(context.test_dir), force=True
|
|
)
|
|
|
|
# Don't change back - stay in test directory for operations
|
|
|
|
def cleanup():
|
|
if hasattr(context, "original_cwd"):
|
|
os.chdir(context.original_cwd)
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
|
|
@given('I have created a plan with "{prompt}"')
|
|
def step_given_created_plan_with_prompt(context: Context, prompt: str) -> None:
|
|
"""Create a plan with given prompt as a Given step."""
|
|
# Check if we're in a service test context by checking for plan_service
|
|
if hasattr(context, "plan_service"):
|
|
# Use service for service tests
|
|
# We're already in the test directory from step_have_plan_service
|
|
context.plan = context.plan_service.create_plan(
|
|
project=context.test_project, prompt=prompt
|
|
)
|
|
elif hasattr(context, "test_dir"):
|
|
# Use in-process CLI for core_cli_commands tests (avoids subprocess overhead)
|
|
import os
|
|
|
|
from cleveragents.cli.commands.plan import tell as plan_tell
|
|
|
|
prev_cwd = os.getcwd()
|
|
os.chdir(context.test_dir)
|
|
try:
|
|
plan_tell(prompt=prompt)
|
|
context.has_plan = True
|
|
except Exception as exc:
|
|
raise AssertionError(f"Failed to create plan: {exc}") from exc
|
|
finally:
|
|
os.chdir(prev_cwd)
|
|
else:
|
|
# No context set up - need to create plan service
|
|
step_have_plan_service(context)
|
|
context.plan = context.plan_service.create_plan(
|
|
project=context.test_project, prompt=prompt
|
|
)
|
|
|
|
|
|
@when('I create a plan with prompt "{prompt}"')
|
|
def step_create_plan_with_prompt(context: Context, prompt: str) -> None:
|
|
"""Create a plan with given prompt."""
|
|
context.plan = context.plan_service.create_plan(
|
|
project=context.test_project, prompt=prompt
|
|
)
|
|
|
|
|
|
@then("the plan should be created")
|
|
def step_plan_should_be_created(context: Context) -> None:
|
|
"""Verify plan was created."""
|
|
assert context.plan is not None, "Plan was not created"
|
|
|
|
|
|
@then('the plan prompt should be "{prompt}"')
|
|
def step_plan_prompt_should_be(context: Context, prompt: str) -> None:
|
|
"""Verify plan prompt."""
|
|
assert context.plan.prompt == prompt, (
|
|
f"Expected prompt '{prompt}', got '{context.plan.prompt}'"
|
|
)
|
|
|
|
|
|
@when("I build the plan")
|
|
def step_build_plan(context: Context) -> None:
|
|
"""Build the current plan."""
|
|
# We're already in the test directory from plan service creation
|
|
context.changes = context.plan_service.build_plan(project=context.test_project)
|
|
|
|
|
|
@when("I try to build the plan")
|
|
def step_try_build_plan(context: Context) -> None:
|
|
"""Attempt to build the plan while capturing exceptions."""
|
|
try:
|
|
context.plan_service.build_plan(project=context.test_project)
|
|
context.exception = None
|
|
except Exception as exc: # pragma: no cover - exercised via Behave tests
|
|
context.exception = exc
|
|
|
|
|
|
@given("I have created and built a plan")
|
|
def step_given_created_and_built_plan(context: Context) -> None:
|
|
"""Create and build a plan for testing."""
|
|
# Check if we're in a service test context by checking for plan_service
|
|
if hasattr(context, "plan_service"):
|
|
# Use service for service tests
|
|
# Create a plan
|
|
context.plan = context.plan_service.create_plan(
|
|
project=context.test_project, prompt="Test plan for apply"
|
|
)
|
|
# Build it to generate changes
|
|
context.changes = context.plan_service.build_plan(project=context.test_project)
|
|
elif hasattr(context, "test_dir"):
|
|
# Use in-process CLI for core_cli_commands tests (avoids subprocess overhead)
|
|
import os
|
|
|
|
from cleveragents.cli.commands.plan import build as plan_build
|
|
from cleveragents.cli.commands.plan import tell as plan_tell
|
|
|
|
prev_cwd = os.getcwd()
|
|
os.chdir(context.test_dir)
|
|
try:
|
|
plan_tell(prompt="Add example code")
|
|
plan_build()
|
|
except Exception as exc:
|
|
err_msg = str(exc)
|
|
if "No AI provider configured" in err_msg:
|
|
context.scenario.skip(
|
|
"Skipping test - no AI provider configured (expected in test env)"
|
|
)
|
|
return
|
|
raise AssertionError(f"Failed to create/build plan: {exc}") from exc
|
|
finally:
|
|
os.chdir(prev_cwd)
|
|
else:
|
|
# No context set up - need to create plan service
|
|
step_have_plan_service(context)
|
|
context.plan = context.plan_service.create_plan(
|
|
project=context.test_project, prompt="Test plan for apply"
|
|
)
|
|
context.changes = context.plan_service.build_plan(project=context.test_project)
|
|
|
|
|
|
@then("changes should be generated")
|
|
def step_changes_should_be_generated(context: Context) -> None:
|
|
"""Verify changes were generated."""
|
|
assert len(context.changes) > 0, "No changes were generated"
|
|
|
|
|
|
@when("I apply the changes")
|
|
def step_apply_changes(context: Context) -> None:
|
|
"""Apply the changes."""
|
|
context.applied_count = context.plan_service.apply_changes(
|
|
project=context.test_project
|
|
)
|
|
|
|
|
|
@then("the changes should be applied")
|
|
def step_changes_should_be_applied(context: Context) -> None:
|
|
"""Verify changes were applied."""
|
|
assert context.applied_count > 0, "No changes were applied"
|
|
|
|
|
|
@then("files should be created or modified")
|
|
def step_files_should_be_created(context: Context) -> None:
|
|
"""Verify files were created or modified."""
|
|
from pathlib import Path
|
|
|
|
# Check that at least one file from the changes exists
|
|
for change in context.changes:
|
|
file_path = (
|
|
Path(change.file_path)
|
|
if isinstance(change.file_path, str)
|
|
else change.file_path
|
|
)
|
|
if file_path.exists():
|
|
return # At least one file was created
|
|
# For test purposes, if no actual files were created, check that we have changes
|
|
assert len(context.changes) > 0, "No changes were generated"
|
|
|
|
|
|
@given('I have a test file "{filename}"')
|
|
def step_have_test_file(context: Context, filename: str) -> None:
|
|
"""Create a test file."""
|
|
if not hasattr(context, "test_dir"):
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_context_")
|
|
|
|
def cleanup():
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
test_file = Path(context.test_dir) / filename
|
|
test_file.write_text("test content")
|
|
|
|
|
|
@then("the context should contain {count:d} file")
|
|
@then("the context should contain {count:d} files")
|
|
def step_context_should_contain_count(context: Context, count: int) -> None:
|
|
"""Verify context contains specific number of files."""
|
|
files = context.context_service.list_context(project=context.test_project)
|
|
assert len(files) == count, f"Expected {count} files, got {len(files)}"
|
|
|
|
|
|
@given("I have a directory with multiple files")
|
|
def step_have_directory_with_files(context: Context) -> None:
|
|
"""Create a directory with multiple test files."""
|
|
if not hasattr(context, "test_dir"):
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_context_")
|
|
|
|
def cleanup():
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
# Create a subdirectory with files
|
|
subdir = Path(context.test_dir) / "subdir"
|
|
subdir.mkdir()
|
|
(subdir / "file1.py").write_text("content1")
|
|
(subdir / "file2.py").write_text("content2")
|
|
context.test_subdir = subdir
|
|
|
|
|
|
@given("I have a directory with nested ignored files")
|
|
def step_have_directory_with_nested_ignored(context: Context) -> None:
|
|
"""Create a directory containing nested files and ignored patterns."""
|
|
if not hasattr(context, "test_dir"):
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_context_")
|
|
|
|
def cleanup():
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
base_dir = Path(context.test_dir) / "nested_ignored"
|
|
base_dir.mkdir(exist_ok=True)
|
|
(base_dir / "root.py").write_text("root")
|
|
(base_dir / "ignored.pyc").write_text("cache")
|
|
|
|
nested_dir = base_dir / "child"
|
|
nested_dir.mkdir(exist_ok=True)
|
|
(nested_dir / "nested.py").write_text("nested")
|
|
|
|
git_dir = base_dir / ".git"
|
|
git_dir.mkdir(exist_ok=True)
|
|
(git_dir / "config").write_text("config")
|
|
|
|
context.dir_under_test = base_dir
|
|
|
|
|
|
@given("I have a directory with ignored context files")
|
|
def step_have_directory_with_ignored_context_files(context: Context) -> None:
|
|
"""Create a directory with files that should be ignored."""
|
|
if not hasattr(context, "test_dir"):
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_context_")
|
|
|
|
def cleanup():
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
target_dir = Path(context.test_dir) / "ignored_context_files"
|
|
target_dir.mkdir(exist_ok=True)
|
|
(target_dir / "keep.txt").write_text("keep")
|
|
(target_dir / "hidden.txt").write_text("hidden")
|
|
(target_dir / "ignored.pyc").write_text("ignored cache")
|
|
|
|
context.context_service.extra_ignore_patterns = ["hidden.txt"]
|
|
context.dir_under_test = target_dir
|
|
|
|
|
|
@when("I add the directory to context recursively")
|
|
def step_add_directory_recursively(context: Context) -> None:
|
|
"""Add a directory to context recursively."""
|
|
target_dir = getattr(context, "dir_under_test", None)
|
|
if target_dir is None:
|
|
target_dir = context.test_subdir
|
|
|
|
added_files, already_in_context = context.context_service.add_to_context(
|
|
project=context.test_project, path=target_dir, recursive=True
|
|
)
|
|
context.added_files = added_files
|
|
context.already_in_context = already_in_context
|
|
context.add_result = (added_files, already_in_context)
|
|
|
|
|
|
@then("all files should be added to context")
|
|
def step_all_files_should_be_added(context: Context) -> None:
|
|
"""Verify all files from directory were added."""
|
|
assert len(context.added_files) == 2, (
|
|
f"Expected 2 files, got {len(context.added_files)}"
|
|
)
|
|
|
|
|
|
@given('I have added "{filename}" to the context')
|
|
def step_have_added_file_to_context(context: Context, filename: str) -> None:
|
|
"""Add a file to context as a given."""
|
|
# Create the file if it doesn't exist
|
|
if not hasattr(context, "test_dir"):
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_context_")
|
|
|
|
def cleanup():
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
test_file = Path(context.test_dir) / filename
|
|
test_file.write_text("test content")
|
|
context.context_service.add_to_context(project=context.test_project, path=test_file)
|
|
|
|
|
|
@when('I remove "{filename}" from context')
|
|
def step_remove_file_from_context(context: Context, filename: str) -> None:
|
|
"""Remove a file from context."""
|
|
test_file = Path(context.test_dir) / filename
|
|
context.removed_count = context.context_service.remove_from_context(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
|
|
|
|
@when('I remove directory "{name}" from context')
|
|
def step_remove_directory_from_context(context: Context, name: str) -> None:
|
|
"""Remove a directory from context."""
|
|
directory_path = Path(context.test_dir) / name
|
|
context.removed_count = context.context_service.remove_from_context(
|
|
project=context.test_project, path=directory_path
|
|
)
|
|
|
|
|
|
@then("the service context should be empty")
|
|
def step_context_should_be_empty(context: Context) -> None:
|
|
"""Verify context is empty."""
|
|
files = context.context_service.list_context(project=context.test_project)
|
|
assert len(files) == 0, f"Expected empty context, got {len(files)} files"
|
|
|
|
|
|
@given("I have added multiple files to context")
|
|
def step_have_added_multiple_files(context: Context) -> None:
|
|
"""Add multiple files to context."""
|
|
if not hasattr(context, "test_dir"):
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_context_")
|
|
|
|
def cleanup():
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
# Add multiple files
|
|
for i in range(3):
|
|
test_file = Path(context.test_dir) / f"file{i}.py"
|
|
test_file.write_text(f"content {i}")
|
|
context.context_service.add_to_context(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
|
|
|
|
@when("I clear the context")
|
|
def step_clear_context(context: Context) -> None:
|
|
"""Clear the context."""
|
|
context.context_service.clear_context(project=context.test_project)
|
|
|
|
|
|
@when("I list the context")
|
|
def step_list_context(context: Context) -> None:
|
|
"""List context files."""
|
|
context.context_files = context.context_service.list_context(
|
|
project=context.test_project
|
|
)
|
|
|
|
|
|
@given("the container lookup for projects fails")
|
|
def step_container_lookup_for_projects_fails(context: Context) -> None:
|
|
"""Override container project service to raise during lookup."""
|
|
from cleveragents.application.container import get_container, override_providers
|
|
|
|
class FailingProjectService:
|
|
def get_current_project(self):
|
|
raise RuntimeError("Simulated project lookup failure")
|
|
|
|
failing_service = FailingProjectService()
|
|
override_providers(project_service=failing_service)
|
|
container = get_container()
|
|
|
|
def cleanup():
|
|
container.project_service.reset_override()
|
|
|
|
add_cleanup(context, cleanup)
|
|
context.project_service_override = failing_service
|
|
context.project_service_override_cleanup = cleanup
|
|
|
|
|
|
@given("the container project service returns the current test project")
|
|
def step_container_project_service_returns_test_project(context: Context) -> None:
|
|
"""Override container project service to return the test project."""
|
|
assert hasattr(context, "test_project"), "Test project not initialized"
|
|
|
|
class CurrentProjectService:
|
|
def __init__(self, behave_context: Context) -> None:
|
|
self._context = behave_context
|
|
|
|
def get_current_project(self):
|
|
return getattr(self._context, "test_project", None)
|
|
|
|
override = CurrentProjectService(context)
|
|
override_providers(project_service=override)
|
|
container = get_container()
|
|
|
|
def cleanup():
|
|
if hasattr(container, "project_service"):
|
|
container.project_service.reset_override()
|
|
|
|
add_cleanup(context, cleanup)
|
|
context.project_service_override = override
|
|
context.project_service_override_cleanup = cleanup
|
|
|
|
|
|
@when("I list context files without providing a project")
|
|
def step_list_context_files_without_project(context: Context) -> None:
|
|
"""List context files relying on container-backed project lookup."""
|
|
context.listed_files = context.context_service.list_files(project=None)
|
|
|
|
|
|
@then("I should receive 1 context file path")
|
|
def step_should_receive_single_context_file_path(context: Context) -> None:
|
|
"""Verify a single context file path is returned."""
|
|
assert hasattr(context, "listed_files"), "No context file listing captured"
|
|
assert len(context.listed_files) == 1, (
|
|
f"Expected 1 context file, got {len(context.listed_files)}"
|
|
)
|
|
|
|
|
|
@then('the list of file paths should include "listed.py"')
|
|
def step_listed_paths_include_listed_py(context: Context) -> None:
|
|
"""Ensure the listed files include the expected filename."""
|
|
assert hasattr(context, "listed_files"), "No context file listing captured"
|
|
file_names = {Path(path).name for path in context.listed_files}
|
|
assert "listed.py" in file_names, f'"listed.py" not found in {file_names}'
|
|
|
|
|
|
@then("I should receive 0 context file paths")
|
|
def step_should_receive_zero_context_file_paths(context: Context) -> None:
|
|
"""Verify no context file paths are returned."""
|
|
assert hasattr(context, "listed_files"), "No context file listing captured"
|
|
assert len(context.listed_files) == 0, (
|
|
f"Expected 0 context files, got {len(context.listed_files)}"
|
|
)
|
|
|
|
|
|
@then("I should see {count:d} files in the list")
|
|
def step_should_see_files_in_list(context: Context, count: int) -> None:
|
|
"""Verify number of files in list."""
|
|
assert len(context.context_files) == count, (
|
|
f"Expected {count} files, got {len(context.context_files)}"
|
|
)
|
|
|
|
|
|
@then('the list should contain "{filename}"')
|
|
def step_list_should_contain_file(context: Context, filename: str) -> None:
|
|
"""Verify a specific file is in the list."""
|
|
# Handle both dict and Context object types
|
|
if context.context_files and hasattr(context.context_files[0], "path"):
|
|
# Context objects
|
|
file_paths = [Path(str(f.path)).name for f in context.context_files]
|
|
else:
|
|
# Dictionaries (legacy)
|
|
file_paths = [Path(f["path"]).name for f in context.context_files]
|
|
assert filename in file_paths, f"File {filename} not in list: {file_paths}"
|
|
|
|
|
|
@then('the list should not contain "hidden.txt"')
|
|
def step_list_should_not_contain_hidden(context: Context) -> None:
|
|
"""Ensure hidden files are excluded from the collected list."""
|
|
file_names = set()
|
|
if getattr(context, "context_files", None):
|
|
files = context.context_files
|
|
if hasattr(files[0], "path"):
|
|
file_names = {Path(str(f.path)).name for f in files}
|
|
else:
|
|
file_names = {Path(f["path"]).name for f in files}
|
|
elif getattr(context, "listed_files", None):
|
|
file_names = {Path(path).name for path in context.listed_files}
|
|
assert "hidden.txt" not in file_names, (
|
|
f'"hidden.txt" unexpectedly found in {file_names}'
|
|
)
|
|
|
|
|
|
@then('the list should not contain "ignored.pyc"')
|
|
def step_list_should_not_contain_ignored_pyc(context: Context) -> None:
|
|
"""Ensure ignored cache files are excluded from the collected list."""
|
|
file_names = set()
|
|
if getattr(context, "context_files", None):
|
|
files = context.context_files
|
|
if hasattr(files[0], "path"):
|
|
file_names = {Path(str(f.path)).name for f in files}
|
|
else:
|
|
file_names = {Path(f["path"]).name for f in files}
|
|
elif getattr(context, "listed_files", None):
|
|
file_names = {Path(path).name for path in context.listed_files}
|
|
assert "ignored.pyc" not in file_names, (
|
|
f'"ignored.pyc" unexpectedly found in {file_names}'
|
|
)
|
|
|
|
|
|
@when('I create a plan named "{name}"')
|
|
def step_create_plan_named(context: Context, name: str) -> None:
|
|
"""Create a plan with a specific name."""
|
|
context.plan = context.plan_service.create_plan(
|
|
project=context.test_project, name=name
|
|
)
|
|
|
|
|
|
@when('I create an empty plan named "{name}"')
|
|
def step_create_empty_plan_named(context: Context, name: str) -> None:
|
|
"""Create an empty plan with a specific name."""
|
|
context.plan = context.plan_service.new_plan(
|
|
project=context.test_project, name=name
|
|
)
|
|
|
|
|
|
@then('the plan name should be "{name}"')
|
|
def step_plan_name_should_be(context: Context, name: str) -> None:
|
|
"""Verify plan has specific name."""
|
|
assert context.plan.name == name, (
|
|
f"Expected name '{name}', got '{context.plan.name}'"
|
|
)
|
|
|
|
|
|
@then("the plan prompt should be empty")
|
|
def step_plan_prompt_should_be_empty(context: Context) -> None:
|
|
"""Verify plan prompt is empty or a placeholder."""
|
|
# Accept either empty or the placeholder prompt used for new empty plans
|
|
assert (
|
|
not context.plan.prompt
|
|
or context.plan.prompt == ""
|
|
or context.plan.prompt == "New plan"
|
|
), f"Expected empty or placeholder prompt, got '{context.plan.prompt}'"
|
|
|
|
|
|
@then("the plan should be the current plan")
|
|
def step_plan_should_be_current(context: Context) -> None:
|
|
"""Verify the plan is the current plan."""
|
|
assert context.plan.current is True, "Plan is not marked as current"
|
|
current = context.plan_service.get_current_plan(project=context.test_project)
|
|
assert current is not None, "No current plan found"
|
|
assert current.id == context.plan.id, (
|
|
f"Plan ID mismatch: expected {context.plan.id}, got {current.id}"
|
|
)
|
|
assert current.name == context.plan.name, (
|
|
f"Plan name mismatch: expected {context.plan.name}, got {current.name}"
|
|
)
|
|
|
|
|
|
@then("the changes should include file operations")
|
|
def step_changes_should_include_file_ops(context: Context) -> None:
|
|
"""Verify changes include file operations."""
|
|
assert any(
|
|
change.operation in ["create", "modify", "delete"] for change in context.changes
|
|
)
|
|
|
|
|
|
@then('the plan should have name "{name}"')
|
|
def step_plan_should_have_name(context: Context, name: str) -> None:
|
|
"""Verify plan has specific name."""
|
|
assert context.plan.name == name, (
|
|
f"Expected name '{name}', got '{context.plan.name}'"
|
|
)
|
|
|
|
|
|
@given("I have created multiple plans")
|
|
def step_given_created_multiple_plans(context: Context) -> None:
|
|
"""Create multiple plans."""
|
|
if not hasattr(context, "plan_service"):
|
|
step_have_plan_service(context)
|
|
|
|
context.plans = []
|
|
for i in range(3):
|
|
plan = context.plan_service.create_plan(
|
|
project=context.test_project, name=f"plan-{i}", prompt=f"Test plan {i}"
|
|
)
|
|
context.plans.append(plan)
|
|
|
|
|
|
@when("I list all plans")
|
|
def step_list_all_plans(context: Context) -> None:
|
|
"""List all plans."""
|
|
context.all_plans = context.plan_service.list_plans(project=context.test_project)
|
|
|
|
|
|
@then("I should see all created plans")
|
|
def step_should_see_all_plans(context: Context) -> None:
|
|
"""Verify all plans are listed."""
|
|
assert len(context.all_plans) >= len(context.plans), (
|
|
f"Expected at least {len(context.plans)} plans, got {len(context.all_plans)}"
|
|
)
|
|
|
|
|
|
@then("each plan should have a name and status")
|
|
def step_each_plan_should_have_name_status(context: Context) -> None:
|
|
"""Verify each plan has name and status."""
|
|
for plan in context.all_plans:
|
|
assert plan.name is not None, "Plan missing name"
|
|
assert plan.status is not None, "Plan missing status"
|
|
|
|
|
|
@given('I have created plan "{name}"')
|
|
def step_given_created_plan_named(context: Context, name: str) -> None:
|
|
"""Create a plan with specific name."""
|
|
if not hasattr(context, "plan_service"):
|
|
step_have_plan_service(context)
|
|
context.plan_service.create_plan(
|
|
project=context.test_project, name=name, prompt=f"Test {name}"
|
|
)
|
|
|
|
|
|
@when('I switch to plan "{name}"')
|
|
def step_switch_to_plan(context: Context, name: str) -> None:
|
|
"""Switch to a specific plan."""
|
|
context.plan_service.switch_to_plan(project=context.test_project, name=name)
|
|
|
|
|
|
@then('"{name}" should be the current service plan')
|
|
def step_should_be_current_plan(context: Context, name: str) -> None:
|
|
"""Verify specific plan is current."""
|
|
current = context.plan_service.get_current_plan(project=context.test_project)
|
|
assert current is not None, "No current plan"
|
|
assert current.name == name, f"Expected current plan '{name}', got '{current.name}'"
|
|
|
|
|
|
@then('"{name}" should not be current')
|
|
def step_should_not_be_current(context: Context, name: str) -> None:
|
|
"""Verify specific plan is not current."""
|
|
current = context.plan_service.get_current_plan(project=context.test_project)
|
|
if current is not None:
|
|
assert current.name != name, f"Plan '{name}' should not be current"
|
|
|
|
|
|
@when('I add "{text}" to the plan')
|
|
def step_add_text_to_plan(context: Context, text: str) -> None:
|
|
"""Add text to the current plan."""
|
|
# We're already in the test directory from plan service creation
|
|
context.plan_service.continue_plan(project=context.test_project, prompt=text)
|
|
|
|
|
|
@then("the plan should contain both instructions")
|
|
def step_plan_should_contain_both(context: Context) -> None:
|
|
"""Verify plan contains both initial and additional instructions."""
|
|
current = context.plan_service.get_current_plan(project=context.test_project)
|
|
assert "Initial instructions" in current.prompt
|
|
assert "Additional instructions" in current.prompt
|
|
|
|
|
|
@given("I am in a project directory")
|
|
def step_in_project_directory(context: Context) -> None:
|
|
"""Set up being in a project directory."""
|
|
if not hasattr(context, "project_service"):
|
|
step_have_project_service(context)
|
|
|
|
# Initialize a project in the test directory
|
|
context.project = context.project_service.initialize_project(
|
|
name="test-project", path=Path(context.test_dir), force=False
|
|
)
|
|
|
|
|
|
@when("I get the current project")
|
|
def step_get_current_project(context: Context) -> None:
|
|
"""Get the current project."""
|
|
import os
|
|
|
|
# Check if we need to change directory or if we're already in the project directory
|
|
if hasattr(context, "test_dir") and context.test_dir:
|
|
# Change to the project directory temporarily
|
|
old_cwd = os.getcwd()
|
|
try:
|
|
os.chdir(context.test_dir)
|
|
context.current_project = context.project_service.get_current_project()
|
|
finally:
|
|
os.chdir(old_cwd)
|
|
else:
|
|
# We're already in the correct directory (e.g., from "And I change to that project directory")
|
|
context.current_project = context.project_service.get_current_project()
|
|
|
|
|
|
@then("I should receive the project information")
|
|
def step_should_receive_project_info(context: Context) -> None:
|
|
"""Verify we got project information."""
|
|
assert context.current_project is not None, "No project information received"
|
|
|
|
|
|
@then("the project should have a name")
|
|
def step_project_should_have_name(context: Context) -> None:
|
|
"""Verify project has a name."""
|
|
assert context.current_project.name is not None, "Project has no name"
|
|
|
|
|
|
@then("the project should have a path")
|
|
def step_project_should_have_path(context: Context) -> None:
|
|
"""Verify project has a path."""
|
|
assert context.current_project.path is not None, "Project has no path"
|
|
|
|
|
|
@given("I have a current project")
|
|
def step_have_current_project(context: Context) -> None:
|
|
"""Ensure we have a current project."""
|
|
if not hasattr(context, "project_service"):
|
|
settings = Settings()
|
|
context.project_service = ProjectService(settings, "sqlite:///test.db")
|
|
|
|
if not hasattr(context, "test_dir"):
|
|
context.test_dir = tempfile.mkdtemp(prefix="test_project_")
|
|
|
|
def cleanup():
|
|
if hasattr(context, "test_dir") and Path(context.test_dir).exists():
|
|
shutil.rmtree(context.test_dir)
|
|
|
|
add_cleanup(context, cleanup)
|
|
|
|
# Initialize a project
|
|
context.project = context.project_service.initialize_project(
|
|
name="test-project", path=Path(context.test_dir), force=False
|
|
)
|
|
|
|
|
|
@when("I get project statistics")
|
|
def step_get_project_statistics(context: Context) -> None:
|
|
"""Get project statistics."""
|
|
context.statistics = context.project_service.get_project_stats(context.project)
|
|
|
|
|
|
@then("I should receive statistics")
|
|
def step_should_receive_statistics(context: Context) -> None:
|
|
"""Verify we got statistics."""
|
|
assert context.statistics is not None, "No statistics received"
|
|
|
|
|
|
@then('the statistics should include "{key}"')
|
|
def step_statistics_should_include(context: Context, key: str) -> None:
|
|
"""Verify statistics include a specific key."""
|
|
assert key in context.statistics, (
|
|
f"Statistics missing key '{key}': {context.statistics}"
|
|
)
|