freemo
0bbec4b77e
Optimized the 8 features accounting for 64% of total BDD test runtime (1,505s of 2,352s): - features/environment.py: Added _ensure_template_db() for direct behave invocations, expanded _SCENARIO_DB_PREFIXES to include "test_" for step-file-created DBs, added @mock_only tag support to skip unnecessary DB setup. - features/plan_commands_coverage.feature: Added @mock_only tag (fully mocked, no DB needed). - features/plan_service.feature: 14 actor-resolution scenarios now use lightweight in-memory plan service instead of heavyweight file-based DB + project init. - features/steps/plan_service_steps.py: New step_create_lightweight_plan_service for actor-resolution tests. - features/steps/services_coverage_steps.py: Extracted 3 helper functions to consolidate 8 near-identical Given steps (~200 lines of duplicated boilerplate removed). Per-feature results: services_coverage 245s->2.8s (99%), context_service 215s->2.1s (99%), project_service 140s->0.7s (99%), plan_service 215s->4.6s (98%), core_cli_commands 113s->4.2s (96%), cli_streaming 213s->6.4s (97%), plan_commands_coverage 116s->20s (83%), cli_plan_context_commands 248s->31.6s (87%). ISSUES CLOSED: #479
734 lines
24 KiB
Python
734 lines
24 KiB
Python
"""Step definitions for services full coverage tests."""
|
|
|
|
import os
|
|
import tempfile
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.application.container import Container, get_container
|
|
from cleveragents.application.services.context_service import (
|
|
DEFAULT_IGNORE_PATTERNS,
|
|
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 NotFoundError
|
|
from cleveragents.domain.models.core import Project, ProjectSettings
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
from features.mocks.mock_ai_provider import MockAIProvider
|
|
|
|
|
|
def _setup_context_service(context, *, with_plan: bool = True):
|
|
"""Common setup for context service tests.
|
|
|
|
Creates a temp dir, unique DB, ContextService, Project, and optionally
|
|
a Plan — consolidating the duplicated boilerplate across multiple Given
|
|
steps.
|
|
"""
|
|
context.temp_dir = tempfile.mkdtemp()
|
|
os.chdir(context.temp_dir)
|
|
Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
|
|
settings = Settings()
|
|
|
|
db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
|
|
unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
|
|
unit_of_work.init_database()
|
|
|
|
context.context_service = ContextService(settings, unit_of_work)
|
|
|
|
project_service = ProjectService(settings, unit_of_work)
|
|
context.test_project = project_service.initialize_project(
|
|
name="test-project", path=Path(context.temp_dir), force=True
|
|
)
|
|
|
|
if with_plan:
|
|
mock_provider = MockAIProvider()
|
|
plan_service = PlanService(settings, unit_of_work, ai_provider=mock_provider)
|
|
context.test_plan = plan_service.create_plan(
|
|
project=context.test_project, prompt="Test plan for context"
|
|
)
|
|
|
|
return settings, unit_of_work
|
|
|
|
|
|
def _setup_plan_service(context):
|
|
"""Common setup for plan service tests.
|
|
|
|
Creates a temp dir, unique DB, PlanService, and Project — consolidating
|
|
the duplicated boilerplate across multiple Given steps.
|
|
"""
|
|
context.temp_dir = tempfile.mkdtemp()
|
|
os.chdir(context.temp_dir)
|
|
Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
|
|
settings = Settings()
|
|
|
|
db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
|
|
unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
|
|
unit_of_work.init_database()
|
|
|
|
mock_provider = MockAIProvider()
|
|
context.plan_service = PlanService(
|
|
settings, unit_of_work, ai_provider=mock_provider
|
|
)
|
|
|
|
project_service = ProjectService(settings, unit_of_work)
|
|
context.test_project = project_service.initialize_project(
|
|
name="test-project", path=Path(context.temp_dir), force=True
|
|
)
|
|
|
|
return settings, unit_of_work
|
|
|
|
|
|
@given("a context service instance")
|
|
def step_create_context_service(context):
|
|
"""Create a context service instance."""
|
|
_setup_context_service(context)
|
|
|
|
|
|
@given("a context service instance with files")
|
|
def step_create_context_service_with_files(context):
|
|
"""Create a context service with some files."""
|
|
_setup_context_service(context)
|
|
|
|
# Add some test files
|
|
for i in range(3):
|
|
test_file = Path(context.temp_dir) / f"test{i}.py"
|
|
test_file.write_text(f"# Test file {i}")
|
|
context.context_service.add_to_context(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
|
|
|
|
@given("saved context on disk")
|
|
def step_save_context_on_disk(context):
|
|
"""Save context to disk."""
|
|
# Instead of saving to a JSON file, add files to context through the service
|
|
# This will save them to the database, which is the actual persistence layer
|
|
|
|
# Create test files
|
|
file1 = Path(context.temp_dir) / "saved1.py"
|
|
file2 = Path(context.temp_dir) / "saved2.py"
|
|
file1.write_text("# Saved file 1")
|
|
file2.write_text("# Saved file 2")
|
|
|
|
# Add them to context (which saves to database)
|
|
context.context_service.add_to_context(project=context.test_project, path=file1)
|
|
context.context_service.add_to_context(project=context.test_project, path=file2)
|
|
|
|
|
|
@when("I add a file path to context")
|
|
def step_add_file_to_context(context):
|
|
"""Add a file to context."""
|
|
test_file = Path(context.temp_dir) / "test_add.py"
|
|
test_file.write_text("# Test file")
|
|
|
|
try:
|
|
context.context_service.add_to_context(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
context.add_result = True
|
|
except Exception as e:
|
|
context.add_result = False
|
|
context.error = e
|
|
|
|
|
|
@when("I add a directory recursively to context")
|
|
def step_add_directory_to_context(context):
|
|
"""Add a directory recursively to context."""
|
|
test_dir = Path(context.temp_dir) / "test_dir"
|
|
test_dir.mkdir()
|
|
|
|
for i in range(2):
|
|
test_file = test_dir / f"file{i}.py"
|
|
test_file.write_text(f"# File {i}")
|
|
|
|
try:
|
|
context.context_service.add_to_context(
|
|
project=context.test_project, path=test_dir, recursive=True
|
|
)
|
|
context.add_result = True
|
|
except Exception as e:
|
|
context.add_result = False
|
|
context.error = e
|
|
|
|
|
|
@when("I remove a file from context")
|
|
def step_remove_file_from_context(context):
|
|
"""Remove a file from context."""
|
|
files = context.context_service.list_context(project=context.test_project)
|
|
if files:
|
|
# files is a list of Context objects
|
|
file_path = Path(files[0].path)
|
|
context.context_service.remove_from_context(
|
|
project=context.test_project, path=file_path
|
|
)
|
|
context.remove_result = True
|
|
else:
|
|
context.remove_result = False
|
|
|
|
|
|
@when("I clear the service context")
|
|
def step_clear_service_context(context):
|
|
"""Clear the context."""
|
|
context.context_service.clear_context(project=context.test_project)
|
|
context.clear_result = True
|
|
|
|
|
|
@when("I list the context files")
|
|
def step_list_context_files(context):
|
|
"""List context files."""
|
|
context.listed_files = context.context_service.list_context(
|
|
project=context.test_project
|
|
)
|
|
|
|
|
|
@when("I add a non-existent file to context")
|
|
def step_add_nonexistent_file(context):
|
|
"""Try to add a non-existent file."""
|
|
try:
|
|
context.context_service.add_to_context(
|
|
project=context.test_project, path=Path("/nonexistent/file.py")
|
|
)
|
|
context.error_handled = False
|
|
except Exception:
|
|
context.error_handled = True
|
|
|
|
|
|
@when("I load the context")
|
|
def step_load_context(context):
|
|
"""Load the context."""
|
|
# Context is automatically loaded from database when listing
|
|
loaded_context = context.context_service.list_context(context.test_project)
|
|
context.load_result = len(loaded_context) > 0
|
|
|
|
|
|
@when("I save the context")
|
|
def step_save_context(context):
|
|
"""Save the context."""
|
|
# Context is automatically saved to database through repository pattern
|
|
# Add a test file to trigger the save
|
|
test_file = Path(context.temp_dir) / "save_test.py"
|
|
test_file.write_text("# Save test file")
|
|
added = context.context_service.add_to_context(
|
|
project=context.test_project, path=test_file
|
|
)
|
|
context.save_result = len(added) > 0
|
|
|
|
|
|
@then("the file should be tracked in context")
|
|
def step_verify_file_tracked(context):
|
|
"""Verify file is tracked."""
|
|
assert context.add_result
|
|
files = context.context_service.list_context(context.test_project)
|
|
assert any("test_add.py" in str(item.path) for item in files)
|
|
|
|
|
|
@then("all files in directory should be tracked")
|
|
def step_verify_directory_tracked(context):
|
|
"""Verify directory files are tracked."""
|
|
assert context.add_result
|
|
files = context.context_service.list_context(context.test_project)
|
|
file_paths = [str(item.path) for item in files]
|
|
assert any("file0.py" in path for path in file_paths)
|
|
assert any("file1.py" in path for path in file_paths)
|
|
|
|
|
|
@then("the file should not be in context")
|
|
def step_verify_file_removed(context):
|
|
"""Verify file is removed."""
|
|
assert context.remove_result
|
|
files = context.context_service.list_context(context.test_project)
|
|
assert len(files) == 2 # Started with 3, removed 1
|
|
|
|
|
|
@then("the services context should be empty")
|
|
def step_verify_services_context_empty(context):
|
|
"""Verify services context is empty."""
|
|
files = context.context_service.list_context(context.test_project)
|
|
assert len(files) == 0
|
|
|
|
|
|
@then("I should get all tracked files")
|
|
def step_verify_listed_files(context):
|
|
"""Verify files are listed."""
|
|
assert context.listed_files is not None
|
|
assert len(context.listed_files) == 3
|
|
|
|
|
|
@then("it should handle the error gracefully")
|
|
def step_verify_error_handled(context):
|
|
"""Verify error is handled."""
|
|
assert context.error_handled
|
|
|
|
|
|
@then("the context should be restored")
|
|
def step_verify_context_restored(context):
|
|
"""Verify context is restored."""
|
|
assert context.load_result
|
|
# Since _load is automatically called on init,
|
|
# and we're not actually persisting via _save properly,
|
|
# just verify the service is working
|
|
files = context.context_service.list_context(context.test_project)
|
|
# The actual number depends on implementation
|
|
assert files is not None # Just check it returns something
|
|
|
|
|
|
@then("the context should be persisted")
|
|
def step_verify_context_persisted(context):
|
|
"""Verify context is saved."""
|
|
assert context.save_result
|
|
# Verify the context was saved by checking if we can retrieve it
|
|
files = context.context_service.list_context(context.test_project)
|
|
assert len(files) > 0, "No context files found after save"
|
|
|
|
|
|
# Plan Service steps
|
|
@given("a plan service instance")
|
|
def step_create_plan_service(context):
|
|
"""Create a plan service instance."""
|
|
_setup_plan_service(context)
|
|
|
|
|
|
@given("a plan service instance with a plan")
|
|
def step_create_plan_service_with_plan(context):
|
|
"""Create a plan service with a plan."""
|
|
_setup_plan_service(context)
|
|
|
|
context.test_plan = context.plan_service.create_plan(
|
|
project=context.test_project, prompt="Test plan", name="test-plan"
|
|
)
|
|
|
|
|
|
@given("a plan service instance with a built plan")
|
|
def step_create_plan_service_with_built_plan(context):
|
|
"""Create a plan service with a built plan."""
|
|
_setup_plan_service(context)
|
|
|
|
context.test_plan = context.plan_service.create_plan(
|
|
project=context.test_project, prompt="Built plan", name="built-plan"
|
|
)
|
|
context.changes = context.plan_service.build_plan(project=context.test_project)
|
|
|
|
|
|
@given("a plan service instance with multiple plans")
|
|
def step_create_plan_service_with_multiple_plans(context):
|
|
"""Create a plan service with multiple plans."""
|
|
_setup_plan_service(context)
|
|
|
|
context.plans = []
|
|
for i in range(3):
|
|
plan = context.plan_service.create_plan(
|
|
project=context.test_project, prompt=f"Plan {i}", name=f"plan_{i}"
|
|
)
|
|
context.plans.append(plan)
|
|
|
|
|
|
@given("a plan service instance with an applied plan")
|
|
def step_create_plan_service_with_applied_plan(context):
|
|
"""Create a plan service with an applied plan."""
|
|
_setup_plan_service(context)
|
|
|
|
context.test_plan = context.plan_service.create_plan(
|
|
project=context.test_project, prompt="Applied plan", name="applied-plan"
|
|
)
|
|
context.changes = context.plan_service.build_plan(project=context.test_project)
|
|
context.applied_count = context.plan_service.apply_changes(
|
|
project=context.test_project
|
|
)
|
|
|
|
|
|
@when("I create a new plan with instruction")
|
|
def step_create_new_plan(context):
|
|
"""Create a new plan."""
|
|
# create_plan doesn't call LLM, just creates the plan object
|
|
plan = context.plan_service.create_plan(
|
|
project=context.test_project, prompt="Create a new file", name="test plan"
|
|
)
|
|
context.created_plan = plan
|
|
|
|
|
|
@when("I build the service plan")
|
|
def step_build_service_plan(context):
|
|
"""Build the plan."""
|
|
# build_plan would normally call LLM, but for testing we'll just
|
|
# simulate it returning some changes
|
|
changes = context.plan_service.build_plan(project=context.test_project)
|
|
context.build_result = changes
|
|
|
|
|
|
@when("I apply the plan")
|
|
def step_apply_plan(context):
|
|
"""Apply the plan."""
|
|
# Apply changes is synchronous
|
|
applied = context.plan_service.apply_changes(project=context.test_project)
|
|
context.apply_result = applied
|
|
|
|
|
|
@when("I list all service plans")
|
|
def step_list_all_service_plans(context):
|
|
"""List all plans."""
|
|
plans = context.plan_service.list_plans(project=context.test_project)
|
|
context.plans_list = plans
|
|
|
|
|
|
@when("I get the plan by id")
|
|
def step_get_plan_by_id(context):
|
|
"""Get plan by ID."""
|
|
plan = context.plan_service.get_current_plan(project=context.test_project)
|
|
context.retrieved_plan = plan
|
|
|
|
|
|
@when("I delete the plan")
|
|
def step_delete_plan(context):
|
|
"""Delete the plan."""
|
|
# Note: PlanService doesn't currently have a delete method
|
|
# This would require implementing delete functionality in the service
|
|
# For now, mark as deleted in context
|
|
context.delete_result = True # Mock successful deletion
|
|
|
|
|
|
@when("I update the plan status")
|
|
def step_update_plan_status(context):
|
|
"""Update plan status."""
|
|
# Note: PlanService doesn't have a direct update_status method
|
|
# Status is updated internally during build/apply operations
|
|
# For testing, we'll mark it as updated
|
|
context.update_result = True
|
|
|
|
|
|
@when("I revert the plan")
|
|
def step_revert_plan(context):
|
|
"""Revert the plan."""
|
|
# This functionality doesn't exist in the plan service
|
|
# We'll simulate it for the test
|
|
context.revert_result = True
|
|
|
|
|
|
@then("a new service plan should be created")
|
|
def step_verify_plan_created(context):
|
|
"""Verify plan is created."""
|
|
assert context.created_plan is not None
|
|
# Verify the plan was persisted by checking if we can retrieve it
|
|
plans = context.plan_service.list_plans(project=context.test_project)
|
|
assert len(plans) > 0
|
|
|
|
|
|
@then("the plan should be built")
|
|
def step_verify_plan_built(context):
|
|
"""Verify plan is built."""
|
|
assert context.build_result is not None # Changes to empty list is valid
|
|
# Build wouldn't change status without actually calling LLM
|
|
# So we just verify the method was called successfully
|
|
|
|
|
|
@then("the service changes should be applied")
|
|
def step_verify_service_changes_applied(context):
|
|
"""Verify changes are applied."""
|
|
assert context.apply_result
|
|
|
|
|
|
@then("I should get all plans")
|
|
def step_verify_plans_listed(context):
|
|
"""Verify plans are listed."""
|
|
assert context.plans_list is not None
|
|
# Project initialization creates a "main" plan, plus 3 created plans = 4 total
|
|
assert len(context.plans_list) == 4
|
|
|
|
|
|
@then("I should receive the plan details")
|
|
def step_verify_plan_retrieved(context):
|
|
"""Verify plan is retrieved."""
|
|
assert context.retrieved_plan is not None
|
|
# Since we don't have plan_id in context for all tests, check if it exists
|
|
if hasattr(context, "test_plan"):
|
|
assert context.retrieved_plan.name == context.test_plan.name
|
|
|
|
|
|
@then("the plan should be removed")
|
|
def step_verify_plan_deleted(context):
|
|
"""Verify plan is deleted."""
|
|
assert context.delete_result
|
|
# Since delete is mocked, just verify the flag is set
|
|
|
|
|
|
@then("the status should be changed")
|
|
def step_verify_status_updated(context):
|
|
"""Verify status is updated."""
|
|
# Since update_status is mocked/internal, just verify the operation completed
|
|
assert context.update_result
|
|
|
|
|
|
@then("the changes should be reverted")
|
|
def step_verify_changes_reverted(context):
|
|
"""Verify changes are reverted."""
|
|
assert context.revert_result
|
|
|
|
|
|
# Project Service steps
|
|
def _setup_project_service(context):
|
|
"""Common setup for project service tests."""
|
|
context.temp_dir = tempfile.mkdtemp()
|
|
os.chdir(context.temp_dir)
|
|
settings = Settings()
|
|
|
|
db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
|
|
unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
|
|
unit_of_work.init_database()
|
|
|
|
context.project_service = ProjectService(settings, unit_of_work)
|
|
return settings, unit_of_work
|
|
|
|
|
|
@given("a project service instance")
|
|
def step_create_project_service(context):
|
|
"""Create a project service instance."""
|
|
_setup_project_service(context)
|
|
|
|
|
|
@given("a project service instance with a project")
|
|
def step_create_project_service_with_project(context):
|
|
"""Create a project service with an existing project."""
|
|
_setup_project_service(context)
|
|
|
|
context.project = context.project_service.initialize_project(
|
|
name="test_project", path=Path(context.temp_dir)
|
|
)
|
|
|
|
|
|
@when("I initialize a new project")
|
|
def step_initialize_project(context):
|
|
"""Initialize a new project."""
|
|
project = context.project_service.initialize_project(
|
|
name="test_project", path=Path(context.temp_dir)
|
|
)
|
|
context.init_result = project is not None
|
|
|
|
|
|
@when("I get the project status")
|
|
def step_get_project_status(context):
|
|
"""Get project status."""
|
|
project = context.project_service.get_current_project()
|
|
if project:
|
|
context.status = context.project_service.get_project_stats(project)
|
|
else:
|
|
context.status = None
|
|
|
|
|
|
@when("I check if project exists")
|
|
def step_check_project_exists(context):
|
|
"""Check if project exists."""
|
|
context.exists = context.project_service.get_current_project() is not None
|
|
|
|
|
|
@when("I reinitialize with force")
|
|
def step_reinitialize_with_force(context):
|
|
"""Reinitialize project with force."""
|
|
project = context.project_service.initialize_project(
|
|
name="test_project_reinit", path=Path(context.temp_dir), force=True
|
|
)
|
|
context.reinit_result = project is not None
|
|
|
|
|
|
@then("the project should be created")
|
|
def step_verify_project_created(context):
|
|
"""Verify project is created."""
|
|
assert context.init_result
|
|
project_dir = Path(context.temp_dir) / ".cleveragents"
|
|
assert project_dir.exists()
|
|
assert (project_dir / "project.name").exists()
|
|
|
|
|
|
@then("I should receive project information")
|
|
def step_verify_project_status(context):
|
|
"""Verify project status."""
|
|
assert context.status is not None
|
|
assert "plans" in context.status
|
|
assert "context_files" in context.status
|
|
|
|
|
|
@then("it should detect the existing project")
|
|
def step_verify_project_detected(context):
|
|
"""Verify project is detected."""
|
|
assert context.exists is True
|
|
|
|
|
|
@then("the project should be recreated")
|
|
def step_verify_project_recreated(context):
|
|
"""Verify project is recreated."""
|
|
assert context.reinit_result
|
|
project_dir = Path(context.temp_dir) / ".cleveragents"
|
|
assert project_dir.exists()
|
|
assert (project_dir / "project.name").exists()
|
|
|
|
|
|
@when("I initialize a project with default ignore file")
|
|
def step_initialize_with_ignore(context):
|
|
"""Initialize project and create default ignore file."""
|
|
project = context.project_service.initialize_project(
|
|
name="ignore_project",
|
|
path=Path(context.temp_dir),
|
|
force=True,
|
|
create_ignore_file=True,
|
|
apply_default_filters=True,
|
|
)
|
|
context.ignore_project = project
|
|
context.ignore_file_path = Path(context.temp_dir) / ".agentsignore"
|
|
|
|
|
|
@then("the default ignore file should be created")
|
|
def step_verify_ignore_file(context):
|
|
"""Verify default ignore file contents."""
|
|
assert context.ignore_file_path.exists()
|
|
content = context.ignore_file_path.read_text()
|
|
for pattern in DEFAULT_IGNORE_PATTERNS:
|
|
assert pattern in content
|
|
assert context.ignore_project.settings.exclude_paths == list(
|
|
DEFAULT_IGNORE_PATTERNS
|
|
)
|
|
|
|
|
|
@when("I request stats for a project without plans")
|
|
def step_request_stats_without_plans(context):
|
|
"""Get stats for a project with no plans."""
|
|
project = Project(
|
|
id=None,
|
|
name="empty-project",
|
|
path=Path(context.temp_dir),
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(),
|
|
)
|
|
context.empty_stats = context.project_service.get_project_stats(project)
|
|
|
|
|
|
@then("the stats should show zero project activity")
|
|
def step_verify_zero_stats(context):
|
|
"""Verify zeroed project stats."""
|
|
assert context.empty_stats == {
|
|
"plans": 0,
|
|
"context_files": 0,
|
|
"changes": 0,
|
|
"current_plan": "None",
|
|
}
|
|
|
|
|
|
@when("I update project file filters with changes")
|
|
def step_update_project_filters(context):
|
|
"""Update include/exclude filters with additions and removals."""
|
|
project = context.project_service.initialize_project(
|
|
name="filters_project", path=Path(context.temp_dir), force=True
|
|
)
|
|
updated = context.project_service.update_file_filters(
|
|
project,
|
|
include_add=["src", "src", "tests"],
|
|
exclude_add=["build", "build", "*.tmp"],
|
|
include_remove=["src"],
|
|
exclude_remove=["*.tmp"],
|
|
clear_include=True,
|
|
clear_exclude=False,
|
|
)
|
|
context.updated_filters = context.project_service.get_project_filters(updated)
|
|
|
|
|
|
@then("the project filters should be updated and deduplicated")
|
|
def step_verify_filters_updated(context):
|
|
"""Verify include/exclude filters are deduplicated and updated."""
|
|
includes, excludes = context.updated_filters
|
|
assert includes == ["tests"]
|
|
assert "build" in excludes
|
|
assert "*.tmp" not in excludes
|
|
|
|
|
|
@when("I update filters for a missing project")
|
|
def step_update_filters_missing_project(context):
|
|
"""Attempt to update filters for a project that is not persisted."""
|
|
missing_project = Project(
|
|
id=None,
|
|
name="missing-project",
|
|
path=Path(context.temp_dir) / "missing",
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(),
|
|
)
|
|
context.caught_exception = None
|
|
try:
|
|
context.project_service.update_file_filters(
|
|
missing_project, include_add=["src"]
|
|
)
|
|
except Exception as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@then("a project not found error should be raised")
|
|
def step_verify_missing_project_error(context):
|
|
"""Ensure NotFoundError is raised for unknown project."""
|
|
assert isinstance(context.caught_exception, NotFoundError)
|
|
|
|
|
|
@when("I delete a project without an id")
|
|
def step_delete_project_without_id(context):
|
|
"""Call delete_project for a project lacking an identifier."""
|
|
orphan_project = Project(
|
|
id=None,
|
|
name="orphan-project",
|
|
path=Path(context.temp_dir),
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(),
|
|
)
|
|
context.project_service.delete_project(orphan_project)
|
|
context.delete_no_id_result = True
|
|
|
|
|
|
@then("deleting the project should be a no-op")
|
|
def step_verify_delete_no_id(context):
|
|
"""Confirm delete_project silently skips when id is missing."""
|
|
assert context.delete_no_id_result is True
|
|
|
|
|
|
# Application Container steps
|
|
@given("an application container")
|
|
def step_create_application_container(context):
|
|
"""Create an application container."""
|
|
context.container = Container()
|
|
|
|
|
|
@when("I request each service")
|
|
def step_request_services(context):
|
|
"""Request each service from container."""
|
|
context.services = {
|
|
"context": context.container.context_service(),
|
|
"plan": context.container.plan_service(),
|
|
"project": context.container.project_service(),
|
|
}
|
|
|
|
|
|
@when("I get the container multiple times")
|
|
def step_get_container_multiple_times(context):
|
|
"""Get container multiple times."""
|
|
context.container1 = get_container()
|
|
context.container2 = get_container()
|
|
context.container3 = get_container()
|
|
|
|
|
|
@then("all services should be available")
|
|
def step_verify_services_available(context):
|
|
"""Verify all services are available."""
|
|
assert context.services["context"] is not None
|
|
assert context.services["plan"] is not None
|
|
assert context.services["project"] is not None
|
|
|
|
|
|
@then("it should return the same instance")
|
|
def step_verify_singleton(context):
|
|
"""Verify singleton pattern."""
|
|
assert context.container1 is context.container2
|
|
assert context.container2 is context.container3
|