freemo
051ee7c290
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 21s
CI / quality (pull_request) Successful in 31s
CI / typecheck (pull_request) Successful in 47s
CI / security (pull_request) Successful in 52s
CI / build (pull_request) Successful in 56s
CI / e2e_tests (pull_request) Successful in 5m1s
CI / integration_tests (pull_request) Successful in 5m30s
CI / unit_tests (pull_request) Successful in 5m42s
CI / docker (pull_request) Successful in 58s
CI / coverage (pull_request) Successful in 7m35s
CI / build (push) Successful in 21s
CI / docker (push) Has been skipped
CI / benchmark-regression (pull_request) Failing after 49m24s
CI / lint (push) Successful in 22s
CI / quality (push) Successful in 39s
CI / security (push) Successful in 48s
CI / typecheck (push) Successful in 1m26s
CI / benchmark-regression (push) Has been skipped
CI / e2e_tests (push) Successful in 5m53s
CI / coverage (push) Successful in 9m4s
CI / benchmark-publish (push) Successful in 19m10s
CI / integration_tests (push) Failing after 19m18s
CI / unit_tests (push) Failing after 19m20s
Added 52 new .feature files and corresponding _steps.py files targeting previously uncovered code paths in the following areas: - TUI layer: app, commands, persona (state/schema/registry), widgets, input (shell_exec, reference_parser) - Application services: plan lifecycle/service/executor, session, project, repo indexing, correction, checkpoint, actor, llm_actors, strategy coordinator, resource file watcher, service retry wiring - CLI commands: session, resource, repl, plan, db, automation_profile - Domain models: retry_policy, resource_type, cost_budget, docker_compose_analyzer, detail_level, _sql_string_aware, _postgresql_helpers - Core: circuit_breaker, retry_service_patterns - Infrastructure: repositories, transaction_sandbox, strategy_registry, plugins/loader, container - Config: settings - Agents: plan_generation, context_analysis, auto_debug - A2A: facade All new tests follow the Behave/Gherkin BDD standard. Resolved step definition collisions with unique prefixes. Fixed Alembic fileConfig logger disabling issue (disable_existing_loggers=False). ISSUES CLOSED: #1068
1672 lines
55 KiB
Python
1672 lines
55 KiB
Python
"""Step definitions for database infrastructure tests."""
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from cleveragents.domain.models.core import (
|
|
Change,
|
|
Context,
|
|
ContextType,
|
|
OperationType,
|
|
Plan,
|
|
PlanStatus,
|
|
Project,
|
|
ProjectSettings,
|
|
)
|
|
from cleveragents.infrastructure.database.models import (
|
|
Base,
|
|
ChangeModel,
|
|
ContextModel,
|
|
PlanModel,
|
|
ProjectModel,
|
|
)
|
|
from cleveragents.infrastructure.database.repositories import (
|
|
ChangeRepository,
|
|
ContextRepository,
|
|
PlanRepository,
|
|
ProjectRepository,
|
|
)
|
|
|
|
|
|
@given("I have imported the database infrastructure modules")
|
|
def step_import_database_modules(context):
|
|
"""Import all database infrastructure modules."""
|
|
context.database_url = "sqlite:///:memory:"
|
|
context.engine = create_engine(context.database_url)
|
|
context.SessionLocal = sessionmaker(
|
|
bind=context.engine, autoflush=False, autocommit=False
|
|
)
|
|
|
|
# Create all tables
|
|
Base.metadata.create_all(context.engine)
|
|
|
|
|
|
@given("I have a test database session")
|
|
def step_create_test_session(context):
|
|
"""Create a test database session."""
|
|
if not hasattr(context, "SessionLocal"):
|
|
step_import_database_modules(context)
|
|
context.db_session = context.SessionLocal()
|
|
|
|
|
|
# ProjectModel scenarios
|
|
@given('I create a ProjectModel with name "{name}" and path "{path}"')
|
|
def step_create_project_model(context, name, path):
|
|
"""Create a ProjectModel instance."""
|
|
context.project_model = ProjectModel(
|
|
name=name, path=path, created_at=datetime.now(), updated_at=datetime.now()
|
|
)
|
|
|
|
|
|
@when(
|
|
'I set the project settings with model provider "{provider}" and temperature {temperature:f}'
|
|
)
|
|
def step_set_project_settings(context, provider, temperature):
|
|
"""Set project settings."""
|
|
settings = {"model_provider": provider, "temperature": temperature}
|
|
context.project_model.settings = json.dumps(settings)
|
|
|
|
|
|
@when("I save the ProjectModel to the database")
|
|
def step_save_project_model(context):
|
|
"""Save ProjectModel to database."""
|
|
context.db_session.add(context.project_model)
|
|
context.db_session.commit()
|
|
context.db_session.refresh(context.project_model)
|
|
|
|
|
|
@then("the ProjectModel should be persisted with correct attributes")
|
|
def step_verify_project_model_persisted(context):
|
|
"""Verify ProjectModel is persisted."""
|
|
assert context.project_model.id is not None
|
|
saved = (
|
|
context.db_session.query(ProjectModel)
|
|
.filter_by(id=context.project_model.id)
|
|
.first()
|
|
)
|
|
assert saved is not None
|
|
assert saved.name == context.project_model.name
|
|
assert saved.path == context.project_model.path
|
|
|
|
|
|
@then("the ProjectModel should have created_at and updated_at timestamps")
|
|
def step_verify_project_timestamps(context):
|
|
"""Verify ProjectModel timestamps."""
|
|
assert context.project_model.created_at is not None
|
|
assert context.project_model.updated_at is not None
|
|
|
|
|
|
@then("the ProjectModel settings should be correctly stored as JSON")
|
|
def step_verify_project_settings_json(context):
|
|
"""Verify ProjectModel settings are stored as JSON."""
|
|
settings = json.loads(context.project_model.settings)
|
|
assert settings["model_provider"] == "openai"
|
|
assert settings["temperature"] == 0.7
|
|
|
|
|
|
# PlanModel scenarios
|
|
@given("I have an existing ProjectModel with id {project_id:d}")
|
|
def step_create_existing_project(context, project_id):
|
|
"""Create an existing project."""
|
|
project = ProjectModel(
|
|
id=project_id,
|
|
name=f"project-{project_id}",
|
|
path=f"/tmp/project-{project_id}",
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
settings="{}", # Add default settings
|
|
)
|
|
context.db_session.add(project)
|
|
context.db_session.commit()
|
|
context.project_id = project_id
|
|
|
|
|
|
@when('I create a PlanModel for project {project_id:d} with name "{name}"')
|
|
def step_create_plan_model(context, project_id, name):
|
|
"""Create a PlanModel instance."""
|
|
context.plan_model = PlanModel(
|
|
project_id=project_id,
|
|
name=name,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
|
|
|
|
@when('I set the plan prompt to "{prompt}"')
|
|
def step_set_plan_prompt(context, prompt):
|
|
"""Set plan prompt."""
|
|
context.plan_model.prompt = prompt
|
|
|
|
|
|
@when('I set the plan status to "{status}"')
|
|
def step_set_plan_status(context, status):
|
|
"""Set plan status."""
|
|
context.plan_model.status = status
|
|
|
|
|
|
@when("I save the PlanModel to the database")
|
|
def step_save_plan_model(context):
|
|
"""Save PlanModel to database."""
|
|
context.db_session.add(context.plan_model)
|
|
context.db_session.commit()
|
|
context.db_session.refresh(context.plan_model)
|
|
|
|
|
|
@then("the PlanModel should be persisted with correct attributes")
|
|
def step_verify_plan_model_persisted(context):
|
|
"""Verify PlanModel is persisted."""
|
|
assert context.plan_model.id is not None
|
|
saved = (
|
|
context.db_session.query(PlanModel).filter_by(id=context.plan_model.id).first()
|
|
)
|
|
assert saved is not None
|
|
assert saved.name == context.plan_model.name
|
|
assert saved.prompt == context.plan_model.prompt
|
|
assert saved.status == context.plan_model.status
|
|
|
|
|
|
@then("the PlanModel should be linked to the correct project")
|
|
def step_verify_plan_project_link(context):
|
|
"""Verify PlanModel is linked to project."""
|
|
assert context.plan_model.project_id == context.project_id
|
|
project = (
|
|
context.db_session.query(ProjectModel).filter_by(id=context.project_id).first()
|
|
)
|
|
assert project is not None
|
|
|
|
|
|
@then("the PlanModel should have empty contexts and changes collections")
|
|
def step_verify_plan_collections_empty(context):
|
|
"""Verify PlanModel collections are empty."""
|
|
# Refresh to get relationship data
|
|
context.db_session.refresh(context.plan_model)
|
|
assert len(context.plan_model.contexts) == 0
|
|
assert len(context.plan_model.changes) == 0
|
|
|
|
|
|
# ContextModel scenarios
|
|
@given("I have an existing PlanModel with id {plan_id:d}")
|
|
def step_create_existing_plan(context, plan_id):
|
|
"""Create an existing plan."""
|
|
if not hasattr(context, "project_id"):
|
|
step_create_existing_project(context, 1)
|
|
|
|
plan = PlanModel(
|
|
id=plan_id,
|
|
project_id=context.project_id,
|
|
name=f"plan-{plan_id}",
|
|
prompt="Test prompt", # Add required prompt field
|
|
status="pending", # Add default status
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
context.db_session.add(plan)
|
|
context.db_session.commit()
|
|
context.plan_id = plan_id
|
|
|
|
|
|
@when('I create a ContextModel for plan {plan_id:d} with type "{ctx_type}"')
|
|
def step_create_context_model(context, plan_id, ctx_type):
|
|
"""Create a ContextModel instance."""
|
|
context.context_model = ContextModel(
|
|
plan_id=plan_id, type=ctx_type, added_at=datetime.now()
|
|
)
|
|
|
|
|
|
@when('I set the context path to "{path}"')
|
|
def step_set_context_path(context, path):
|
|
"""Set context path."""
|
|
context.context_model.path = path
|
|
|
|
|
|
@when('I set the context file_hash to "{file_hash}"')
|
|
def step_set_context_file_hash(context, file_hash):
|
|
"""Set context file_hash."""
|
|
context.context_model.file_hash = file_hash
|
|
|
|
|
|
@when("I set the context size to {size:d}")
|
|
def step_set_context_size(context, size):
|
|
"""Set context size."""
|
|
context.context_model.size = size
|
|
|
|
|
|
@when("I save the ContextModel to the database")
|
|
def step_save_context_model(context):
|
|
"""Save ContextModel to database."""
|
|
context.db_session.add(context.context_model)
|
|
context.db_session.commit()
|
|
context.db_session.refresh(context.context_model)
|
|
|
|
|
|
@then("the ContextModel should be persisted with correct attributes")
|
|
def step_verify_context_model_persisted(context):
|
|
"""Verify ContextModel is persisted."""
|
|
assert context.context_model.id is not None
|
|
saved = (
|
|
context.db_session.query(ContextModel)
|
|
.filter_by(id=context.context_model.id)
|
|
.first()
|
|
)
|
|
assert saved is not None
|
|
assert saved.path == context.context_model.path
|
|
assert saved.content == context.context_model.content
|
|
assert saved.type == context.context_model.type
|
|
|
|
|
|
@then("the ContextModel should be linked to the correct plan")
|
|
def step_verify_context_plan_link(context):
|
|
"""Verify ContextModel is linked to plan."""
|
|
assert context.context_model.plan_id == context.plan_id
|
|
plan = context.db_session.query(PlanModel).filter_by(id=context.plan_id).first()
|
|
assert plan is not None
|
|
|
|
|
|
# ChangeModel scenarios
|
|
@when('I create a ChangeModel for plan {plan_id:d} with file_path "{file_path}"')
|
|
def step_create_change_model(context, plan_id, file_path):
|
|
"""Create a ChangeModel instance."""
|
|
context.change_model = ChangeModel(
|
|
plan_id=plan_id, file_path=file_path, created_at=datetime.now()
|
|
)
|
|
|
|
|
|
@when('I set the change operation to "{operation}"')
|
|
def step_set_change_operation(context, operation):
|
|
"""Set change operation."""
|
|
# Handle both ChangeModel and ChangeRepository scenarios
|
|
if hasattr(context, "change_model"):
|
|
context.change_model.operation = operation
|
|
elif hasattr(context, "created_change"):
|
|
# For ChangeRepository scenario - already set in add_change, this is a no-op
|
|
pass
|
|
|
|
|
|
@when('I set the original_content to "{content}"')
|
|
def step_set_original_content(context, content):
|
|
"""Set original content."""
|
|
context.change_model.original_content = content
|
|
|
|
|
|
@when('I set the new_content to "{content}"')
|
|
def step_set_new_content(context, content):
|
|
"""Set new content."""
|
|
# Handle both ChangeModel and ChangeRepository scenarios
|
|
if hasattr(context, "change_model"):
|
|
context.change_model.new_content = content
|
|
elif hasattr(context, "created_change"):
|
|
# For ChangeRepository scenario - already set in add_change, this is a no-op
|
|
pass
|
|
|
|
|
|
@when("I save the ChangeModel to the database")
|
|
def step_save_change_model(context):
|
|
"""Save ChangeModel to database."""
|
|
context.db_session.add(context.change_model)
|
|
context.db_session.commit()
|
|
context.db_session.refresh(context.change_model)
|
|
|
|
|
|
@then("the ChangeModel should be persisted with correct attributes")
|
|
def step_verify_change_model_persisted(context):
|
|
"""Verify ChangeModel is persisted."""
|
|
assert context.change_model.id is not None
|
|
saved = (
|
|
context.db_session.query(ChangeModel)
|
|
.filter_by(id=context.change_model.id)
|
|
.first()
|
|
)
|
|
assert saved is not None
|
|
assert saved.file_path == context.change_model.file_path
|
|
assert saved.operation == context.change_model.operation
|
|
|
|
|
|
@then("the ChangeModel should be linked to the correct plan")
|
|
def step_verify_change_plan_link(context):
|
|
"""Verify ChangeModel is linked to plan."""
|
|
assert context.change_model.plan_id == context.plan_id
|
|
plan = context.db_session.query(PlanModel).filter_by(id=context.plan_id).first()
|
|
assert plan is not None
|
|
|
|
|
|
@then("the ChangeModel should have applied flag set to False")
|
|
def step_verify_change_not_applied(context):
|
|
"""Verify ChangeModel is not applied."""
|
|
assert not context.change_model.applied
|
|
assert context.change_model.applied_at is None
|
|
|
|
|
|
@then("the ChangeModel should have applied flag as False by default")
|
|
def step_verify_change_applied_default(context):
|
|
"""Verify ChangeModel applied flag defaults to False."""
|
|
assert not context.change_model.applied
|
|
assert context.change_model.applied_at is None
|
|
|
|
|
|
# Repository scenarios
|
|
# Repository scenarios
|
|
@when('I initialize the database with URL "sqlite:///:memory:"')
|
|
def step_initialize_database_with_url(context):
|
|
"""Initialize database with specific URL."""
|
|
from cleveragents.infrastructure.database import init_database
|
|
|
|
context.engine = init_database("sqlite:///:memory:")
|
|
context.database_initialized = True
|
|
|
|
|
|
@then("all database tables should be created")
|
|
def step_verify_all_tables_created(context):
|
|
"""Verify all database tables are created."""
|
|
from sqlalchemy import inspect
|
|
|
|
inspector = inspect(context.engine)
|
|
tables = inspector.get_table_names()
|
|
assert "projects" in tables
|
|
assert "plans" in tables
|
|
assert "contexts" in tables
|
|
assert "changes" in tables
|
|
|
|
|
|
@then("the database engine should be returned")
|
|
def step_verify_engine_returned(context):
|
|
"""Verify database engine is returned."""
|
|
from sqlalchemy.engine import Engine
|
|
|
|
assert context.engine is not None
|
|
assert isinstance(context.engine, Engine)
|
|
|
|
|
|
@then("I should be able to get a working session")
|
|
def step_verify_working_session(context):
|
|
"""Verify we can get a working session."""
|
|
SessionLocal = sessionmaker(bind=context.engine)
|
|
session = SessionLocal()
|
|
try:
|
|
# Try to query something to verify it works
|
|
from cleveragents.infrastructure.database.models import ProjectModel
|
|
|
|
result = session.query(ProjectModel).all()
|
|
assert result is not None # Even if empty, should not error
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@when("I initialize the database")
|
|
def step_initialize_database(context):
|
|
"""Initialize the database."""
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import create_engine
|
|
|
|
from cleveragents.infrastructure.database.models import Base
|
|
|
|
# Create a temporary database if engine doesn't exist
|
|
if not hasattr(context, "engine"):
|
|
context.temp_dir = tempfile.mkdtemp()
|
|
db_file = Path(context.temp_dir) / "test.db"
|
|
context.engine = create_engine(f"sqlite:///{db_file}")
|
|
|
|
Base.metadata.create_all(context.engine)
|
|
context.database_initialized = True
|
|
|
|
|
|
@then("the database tables should be created")
|
|
def step_verify_database_tables(context):
|
|
"""Verify database tables are created."""
|
|
from sqlalchemy import inspect
|
|
|
|
assert hasattr(context, "engine"), "Database engine not initialized"
|
|
inspector = inspect(context.engine)
|
|
tables = inspector.get_table_names()
|
|
assert "projects" in tables
|
|
assert "plans" in tables
|
|
assert "contexts" in tables
|
|
assert "changes" in tables
|
|
|
|
|
|
@then("the database should be ready for use")
|
|
def step_verify_database_ready(context):
|
|
"""Verify database is ready."""
|
|
assert context.database_initialized is True
|
|
|
|
|
|
# ProjectRepository scenarios
|
|
@given("I have a ProjectRepository instance")
|
|
def step_create_project_repository(context):
|
|
"""Create ProjectRepository instance."""
|
|
if not hasattr(context, "db_session"):
|
|
step_create_test_session(context)
|
|
context.project_repo = ProjectRepository(context.db_session)
|
|
|
|
|
|
@when('I create a project with name "{name}" and path "{path}"')
|
|
def step_create_project_with_repo(context, name, path):
|
|
"""Create project using repository."""
|
|
project = Project(
|
|
name=name,
|
|
path=Path(path),
|
|
settings=ProjectSettings(),
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
context.created_project = context.project_repo.create(project)
|
|
|
|
|
|
@when("I save the project to the database")
|
|
def step_save_project_with_repo(context):
|
|
"""Save project (commit)."""
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the project should have an assigned ID")
|
|
def step_verify_project_has_id(context):
|
|
"""Verify project has ID."""
|
|
assert context.created_project.id is not None
|
|
|
|
|
|
@then("the project should be saved with an assigned ID")
|
|
def step_verify_project_saved_with_id(context):
|
|
"""Verify project is saved with ID."""
|
|
assert context.created_project.id is not None
|
|
# Also verify it's in the database
|
|
retrieved = context.project_repo.get_by_id(context.created_project.id)
|
|
assert retrieved is not None
|
|
|
|
|
|
@then("I should get the same project back")
|
|
def step_verify_same_project_back(context):
|
|
"""Verify we got the same project."""
|
|
assert context.retrieved_project is not None
|
|
assert context.retrieved_project.id == context.created_project.id
|
|
assert context.retrieved_project.name == context.created_project.name
|
|
|
|
|
|
@when('I retrieve the project by name "{name}"')
|
|
def step_retrieve_project_by_specific_name(context, name):
|
|
"""Retrieve project by specific name."""
|
|
context.retrieved_project = context.project_repo.get_by_name(name)
|
|
|
|
|
|
@then("I should get the same project back with correct attributes")
|
|
def step_verify_same_project_with_attributes(context):
|
|
"""Verify we got the same project with correct attributes."""
|
|
assert context.retrieved_project is not None
|
|
assert context.retrieved_project.name == context.created_project.name
|
|
# Verify path matches what was created
|
|
assert str(context.retrieved_project.path) == str(context.created_project.path)
|
|
|
|
|
|
@when('I update the created plan prompt to "{prompt}"')
|
|
def step_update_created_plan_prompt(context, prompt):
|
|
"""Update the created plan's prompt."""
|
|
context.created_plan.prompt = prompt
|
|
|
|
|
|
@then("the plan should be saved with an assigned ID")
|
|
def step_verify_plan_saved_with_id(context):
|
|
"""Verify plan is saved with ID."""
|
|
assert context.created_plan.id is not None
|
|
# Also verify it's in the database
|
|
retrieved = context.plan_repo.get_by_id(context.created_plan.id)
|
|
assert retrieved is not None
|
|
|
|
|
|
@then("I should get the same plan back with correct attributes")
|
|
def step_verify_same_plan_with_attributes(context):
|
|
"""Verify we got the same plan with correct attributes."""
|
|
assert context.retrieved_plan is not None
|
|
assert context.retrieved_plan.id == context.created_plan.id
|
|
assert context.retrieved_plan.name == context.created_plan.name
|
|
if hasattr(context.created_plan, "prompt"):
|
|
assert context.retrieved_plan.prompt == context.created_plan.prompt
|
|
|
|
|
|
@when('I update the project settings default_model to "{model}"')
|
|
def step_update_project_settings_model(context, model):
|
|
"""Update project settings default_model."""
|
|
context.created_project.settings.default_model = model
|
|
|
|
|
|
@when("I save the updates")
|
|
def step_save_updates(context):
|
|
"""Save updates (generic)."""
|
|
# This depends on what we're updating
|
|
if hasattr(context, "created_project") and hasattr(context, "project_repo"):
|
|
context.project_repo.update(context.created_project)
|
|
elif hasattr(context, "created_plan") and hasattr(context, "plan_repo"):
|
|
context.plan_repo.update(context.created_plan)
|
|
context.db_session.commit()
|
|
|
|
|
|
@given("I have created multiple plans for the project")
|
|
def step_create_multiple_plans_for_project(context):
|
|
"""Create multiple plans for the project."""
|
|
if not hasattr(context, "plan_repo"):
|
|
step_create_plan_repository(context)
|
|
if not hasattr(context, "project_id"):
|
|
step_ensure_project_exists(context, 1)
|
|
|
|
context.multiple_plans = []
|
|
for i in range(3):
|
|
plan = Plan(
|
|
project_id=context.project_id,
|
|
name=f"plan-{i + 1}",
|
|
prompt=f"Test prompt {i + 1}",
|
|
status=PlanStatus.PENDING,
|
|
is_current=(i == 1), # Make plan 2 current
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
created = context.plan_repo.create(plan)
|
|
context.multiple_plans.append(created)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("only plan 2 should be marked as current")
|
|
def step_verify_only_plan_2_current(context):
|
|
"""Verify only plan 2 is current."""
|
|
all_plans = context.plan_repo.get_all_for_project(context.project_id)
|
|
current_plans = [p for p in all_plans if p.current]
|
|
assert len(current_plans) == 1
|
|
assert current_plans[0].name == "plan-2"
|
|
|
|
|
|
@when("I retrieve the project by ID")
|
|
def step_db_retrieve_project_by_id(context):
|
|
"""Retrieve project by ID."""
|
|
context.retrieved_project = context.project_repo.get_by_id(
|
|
context.created_project.id
|
|
)
|
|
|
|
|
|
@then("I should get the correct project back")
|
|
def step_verify_retrieved_project(context):
|
|
"""Verify retrieved project."""
|
|
assert context.retrieved_project is not None
|
|
assert context.retrieved_project.id == context.created_project.id
|
|
assert context.retrieved_project.name == context.created_project.name
|
|
|
|
|
|
@given('I have created a project with name "{name}"')
|
|
def step_given_project_created(context, name):
|
|
"""Create a project."""
|
|
step_create_project_repository(context)
|
|
step_create_project_with_repo(context, name, f"/tmp/{name}")
|
|
step_save_project_with_repo(context)
|
|
|
|
|
|
@when('I update the project name to "{new_name}"')
|
|
def step_update_project_name(context, new_name):
|
|
"""Update project name."""
|
|
context.created_project.name = new_name
|
|
context.updated_project = context.project_repo.update(context.created_project)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the project name should be updated in the database")
|
|
def step_verify_project_name_updated(context):
|
|
"""Verify project name is updated."""
|
|
retrieved = context.project_repo.get_by_id(context.created_project.id)
|
|
assert retrieved.name == context.updated_project.name
|
|
|
|
|
|
# PlanRepository scenarios
|
|
@given("I have a PlanRepository instance")
|
|
def step_create_plan_repository(context):
|
|
"""Create PlanRepository instance."""
|
|
if not hasattr(context, "db_session"):
|
|
step_create_test_session(context)
|
|
context.plan_repo = PlanRepository(context.db_session)
|
|
|
|
|
|
@given("I ensure a project exists with id {project_id:d}")
|
|
def step_ensure_project_exists(context, project_id):
|
|
"""Ensure project exists."""
|
|
project = context.db_session.query(ProjectModel).filter_by(id=project_id).first()
|
|
if not project:
|
|
project = ProjectModel(
|
|
id=project_id,
|
|
name=f"project-{project_id}",
|
|
path=f"/tmp/project-{project_id}",
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
settings="{}", # Add default settings
|
|
)
|
|
context.db_session.add(project)
|
|
context.db_session.commit()
|
|
context.project_id = project_id
|
|
|
|
|
|
@when('I create a plan with name "{name}" for project {project_id:d}')
|
|
def step_create_plan_with_repo(context, name, project_id):
|
|
"""Create plan using repository."""
|
|
plan = Plan(
|
|
project_id=project_id,
|
|
name=name,
|
|
prompt="Test prompt",
|
|
status=PlanStatus.PENDING,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
context.created_plan = context.plan_repo.create(plan)
|
|
|
|
|
|
@when("I save the plan to the database")
|
|
def step_save_plan_with_repo(context):
|
|
"""Save plan (commit)."""
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the plan should have an assigned ID")
|
|
def step_verify_plan_has_id(context):
|
|
"""Verify plan has ID."""
|
|
assert context.created_plan.id is not None
|
|
|
|
|
|
@when("I retrieve the plan by ID")
|
|
def step_db_retrieve_plan_by_id(context):
|
|
"""Retrieve plan by ID."""
|
|
context.retrieved_plan = context.plan_repo.get_by_id(context.created_plan.id)
|
|
|
|
|
|
@then("I should get the correct plan back")
|
|
def step_verify_retrieved_plan(context):
|
|
"""Verify retrieved plan."""
|
|
assert context.retrieved_plan is not None
|
|
assert context.retrieved_plan.id == context.created_plan.id
|
|
assert context.retrieved_plan.name == context.created_plan.name
|
|
|
|
|
|
@when("I set plan {plan_id:d} as current for project {project_id:d}")
|
|
def step_set_current_plan(context, plan_id, project_id):
|
|
"""Set current plan for project."""
|
|
plan = context.plan_repo.get_by_id(plan_id)
|
|
if plan:
|
|
context.plan_repo.set_current(project_id, plan_id)
|
|
context.db_session.commit()
|
|
|
|
|
|
@when("I get the current plan for project {project_id:d}")
|
|
def step_get_current_plan(context, project_id):
|
|
"""Get current plan for project."""
|
|
context.current_plan = context.plan_repo.get_current(project_id)
|
|
|
|
|
|
@then("I should receive plan {plan_id:d}")
|
|
def step_verify_current_plan_id(context, plan_id):
|
|
"""Verify current plan ID."""
|
|
assert context.current_plan is not None
|
|
assert context.current_plan.id == plan_id
|
|
|
|
|
|
@given("I have created {count:d} plans for the project")
|
|
def step_create_multiple_plans(context, count):
|
|
"""Create multiple plans."""
|
|
if not hasattr(context, "plan_repo"):
|
|
step_create_plan_repository(context)
|
|
|
|
context.created_plans = []
|
|
for i in range(count):
|
|
plan = Plan(
|
|
project_id=context.project_id,
|
|
name=f"plan-{i + 1}",
|
|
prompt=f"Test prompt {i + 1}",
|
|
status=PlanStatus.PENDING,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
created = context.plan_repo.create(plan)
|
|
context.created_plans.append(created)
|
|
context.db_session.commit()
|
|
|
|
|
|
@when("I get all plans for project {project_id:d}")
|
|
def step_get_all_plans(context, project_id):
|
|
"""Get all plans for project."""
|
|
context.all_plans = context.plan_repo.get_all_for_project(project_id)
|
|
|
|
|
|
@then("I should receive a list of {count:d} plans")
|
|
def step_verify_plan_count(context, count):
|
|
"""Verify plan count."""
|
|
assert len(context.all_plans) == count
|
|
|
|
|
|
@then("all plans should belong to project {project_id:d}")
|
|
def step_verify_plans_project(context, project_id):
|
|
"""Verify all plans belong to project."""
|
|
for plan in context.all_plans:
|
|
assert plan.project_id == project_id
|
|
|
|
|
|
# Additional repository test steps
|
|
@given("I have created a plan with id {plan_id:d}")
|
|
def step_given_plan_created(context, plan_id):
|
|
"""Create a plan with specific ID."""
|
|
if not hasattr(context, "plan_repo"):
|
|
step_create_plan_repository(context)
|
|
if not hasattr(context, "project_id"):
|
|
step_ensure_project_exists(context, 1)
|
|
|
|
plan = Plan(
|
|
id=plan_id,
|
|
project_id=context.project_id,
|
|
name=f"plan-{plan_id}",
|
|
prompt="Test prompt",
|
|
status=PlanStatus.PENDING,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
context.created_plan = context.plan_repo.create(plan)
|
|
context.db_session.commit()
|
|
|
|
|
|
@when("I update the plan with build started_at timestamp")
|
|
def step_update_plan_build_started(context):
|
|
"""Update plan with build started timestamp."""
|
|
from cleveragents.domain.models import PlanBuild
|
|
|
|
# Create or update the build object
|
|
if not context.created_plan.build:
|
|
context.created_plan.build = PlanBuild(
|
|
plan_id=context.created_plan.id or 1,
|
|
started_at=datetime.now(),
|
|
model_used="gpt-4", # Will be updated in next step
|
|
token_count=0,
|
|
)
|
|
else:
|
|
context.created_plan.build.started_at = datetime.now()
|
|
|
|
|
|
@when('I update the plan with model_used "{model}"')
|
|
def step_update_plan_model(context, model):
|
|
"""Update plan with model used."""
|
|
from cleveragents.domain.models import PlanBuild
|
|
|
|
if not context.created_plan.build:
|
|
context.created_plan.build = PlanBuild(
|
|
plan_id=context.created_plan.id or 1,
|
|
started_at=datetime.now(),
|
|
model_used=model,
|
|
token_count=0,
|
|
)
|
|
else:
|
|
context.created_plan.build.model_used = model
|
|
|
|
|
|
@when("I update the plan with token_count {count:d}")
|
|
def step_update_plan_token_count(context, count):
|
|
"""Update plan with token count."""
|
|
from cleveragents.domain.models import PlanBuild
|
|
|
|
if not context.created_plan.build:
|
|
context.created_plan.build = PlanBuild(
|
|
plan_id=context.created_plan.id or 1,
|
|
started_at=datetime.now(),
|
|
model_used="gpt-4", # Default value
|
|
token_count=count,
|
|
)
|
|
else:
|
|
context.created_plan.build.token_count = count
|
|
|
|
|
|
@when("I save the plan updates")
|
|
def step_save_plan_updates(context):
|
|
"""Save plan updates."""
|
|
context.plan_repo.update(context.created_plan)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the plan should have the build information persisted")
|
|
def step_verify_plan_build_info(context):
|
|
"""Verify plan build information."""
|
|
if not hasattr(context, "plan_repo"):
|
|
step_create_plan_repository(context)
|
|
retrieved = context.plan_repo.get_by_id(context.created_plan.id)
|
|
assert retrieved.build is not None
|
|
assert retrieved.build.started_at is not None
|
|
assert retrieved.build.model_used == "gpt-4"
|
|
assert retrieved.build.token_count == 1500
|
|
|
|
|
|
@when("I update the plan with applied_at timestamp")
|
|
def step_update_plan_applied(context):
|
|
"""Update plan with applied timestamp."""
|
|
from cleveragents.domain.models import PlanResult
|
|
|
|
if not context.created_plan.result:
|
|
context.created_plan.result = PlanResult(
|
|
success=True,
|
|
files_created=0,
|
|
files_modified=0,
|
|
files_deleted=0,
|
|
applied_at=datetime.now(),
|
|
)
|
|
else:
|
|
context.created_plan.result.applied_at = datetime.now()
|
|
|
|
|
|
@when("I update the plan with files_created {count:d}")
|
|
def step_update_plan_files_created(context, count):
|
|
"""Update plan with files created."""
|
|
from cleveragents.domain.models import PlanResult
|
|
|
|
if not context.created_plan.result:
|
|
context.created_plan.result = PlanResult(
|
|
success=True,
|
|
files_created=count,
|
|
files_modified=0,
|
|
files_deleted=0,
|
|
applied_at=datetime.now(),
|
|
)
|
|
else:
|
|
context.created_plan.result.files_created = count
|
|
|
|
|
|
@when("I update the plan with files_modified {count:d}")
|
|
def step_update_plan_files_modified(context, count):
|
|
"""Update plan with files modified."""
|
|
from cleveragents.domain.models import PlanResult
|
|
|
|
if not context.created_plan.result:
|
|
context.created_plan.result = PlanResult(
|
|
success=True,
|
|
files_created=0,
|
|
files_modified=count,
|
|
files_deleted=0,
|
|
applied_at=datetime.now(),
|
|
)
|
|
else:
|
|
context.created_plan.result.files_modified = count
|
|
|
|
|
|
@when("I update the plan with files_deleted {count:d}")
|
|
def step_update_plan_files_deleted(context, count):
|
|
"""Update plan with files deleted."""
|
|
from cleveragents.domain.models import PlanResult
|
|
|
|
if not context.created_plan.result:
|
|
context.created_plan.result = PlanResult(
|
|
success=True,
|
|
files_created=0,
|
|
files_modified=0,
|
|
files_deleted=count,
|
|
applied_at=datetime.now(),
|
|
)
|
|
else:
|
|
context.created_plan.result.files_deleted = count
|
|
|
|
|
|
@then("the plan should have the result information persisted")
|
|
def step_verify_plan_result_info(context):
|
|
"""Verify plan result information."""
|
|
if not hasattr(context, "plan_repo"):
|
|
step_create_plan_repository(context)
|
|
retrieved = context.plan_repo.get_by_id(context.created_plan.id)
|
|
assert retrieved.result is not None
|
|
assert retrieved.result.applied_at is not None
|
|
assert retrieved.result.files_created == 2
|
|
assert retrieved.result.files_modified == 3
|
|
assert retrieved.result.files_deleted == 1
|
|
|
|
|
|
# ContextRepository scenarios
|
|
@given("I have a ContextRepository instance")
|
|
def step_create_context_repository(context):
|
|
"""Create ContextRepository instance."""
|
|
if not hasattr(context, "db_session"):
|
|
step_create_test_session(context)
|
|
context.context_repo = ContextRepository(context.db_session)
|
|
|
|
|
|
@given("I ensure a plan exists with id {plan_id:d}")
|
|
def step_ensure_plan_exists(context, plan_id):
|
|
"""Ensure plan exists."""
|
|
plan = context.db_session.query(PlanModel).filter_by(id=plan_id).first()
|
|
if not plan:
|
|
if not hasattr(context, "project_id"):
|
|
step_ensure_project_exists(context, 1)
|
|
plan = PlanModel(
|
|
id=plan_id,
|
|
project_id=context.project_id,
|
|
name=f"plan-{plan_id}",
|
|
prompt="Test prompt",
|
|
status="pending",
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
context.db_session.add(plan)
|
|
context.db_session.commit()
|
|
context.plan_id = plan_id
|
|
|
|
|
|
@when('I add a context item with path "{path}"')
|
|
def step_add_context_item(context, path):
|
|
"""Add context item."""
|
|
context_item = Context(
|
|
plan_id=1,
|
|
type=ContextType.FILE,
|
|
path=path, # Keep as string, don't convert to Path
|
|
content="file contents",
|
|
added_at=datetime.now(),
|
|
)
|
|
context.created_context = context.context_repo.add(context_item)
|
|
|
|
|
|
@when('I set the context type to "{ctx_type}"')
|
|
def step_set_context_type(context, ctx_type):
|
|
"""Set context type (no-op, already set in previous step)."""
|
|
pass
|
|
|
|
|
|
@when('I set the context content to "{content}"')
|
|
def step_set_repository_context_content(context, content):
|
|
"""Set context content for repository test (no-op, already set in add step)."""
|
|
# Check if this is being called in the context of ContextModel or ContextRepository
|
|
if hasattr(context, "context_model"):
|
|
# This is for ContextModel scenario
|
|
context.context_model.content = content
|
|
else:
|
|
# This is for ContextRepository scenario - content already set in add_context_item step
|
|
pass
|
|
|
|
|
|
@then("the context should be saved with an assigned ID")
|
|
def step_verify_context_saved(context):
|
|
"""Verify context is saved."""
|
|
context.db_session.commit()
|
|
assert context.created_context.id is not None
|
|
|
|
|
|
@when("I get all context items for plan {plan_id:d}")
|
|
def step_get_all_context(context, plan_id):
|
|
"""Get all context for plan."""
|
|
context.all_contexts = context.context_repo.get_for_plan(plan_id)
|
|
|
|
|
|
@then("I should receive the added context item")
|
|
def step_verify_context_retrieved(context):
|
|
"""Verify context is retrieved."""
|
|
assert len(context.all_contexts) > 0
|
|
assert context.all_contexts[0].path == context.created_context.path
|
|
|
|
|
|
@given("I have added {count:d} context items to the plan")
|
|
def step_add_multiple_contexts(context, count):
|
|
"""Add multiple context items."""
|
|
if not hasattr(context, "context_repo"):
|
|
step_create_context_repository(context)
|
|
|
|
context.added_contexts = []
|
|
for i in range(count):
|
|
ctx = Context(
|
|
plan_id=1,
|
|
type=ContextType.FILE,
|
|
path=str(Path(f"/context/file{i + 1}.py")), # Convert Path to string
|
|
content=f"content {i + 1}",
|
|
added_at=datetime.now(),
|
|
)
|
|
added = context.context_repo.add(ctx)
|
|
context.added_contexts.append(added)
|
|
context.db_session.commit()
|
|
|
|
|
|
@when("I remove context item with id {ctx_id:d}")
|
|
def step_remove_context(context, ctx_id):
|
|
"""Remove context item."""
|
|
context.context_repo.remove(ctx_id)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the context item should be deleted")
|
|
def step_verify_context_deleted(context):
|
|
"""Verify context is deleted."""
|
|
# Check in database
|
|
from cleveragents.infrastructure.database.models import ContextModel
|
|
|
|
deleted = context.db_session.query(ContextModel).filter_by(id=2).first()
|
|
assert deleted is None
|
|
|
|
|
|
@then("I should receive {count:d} context items")
|
|
def step_verify_context_count(context, count):
|
|
"""Verify context count."""
|
|
assert len(context.all_contexts) == count
|
|
|
|
|
|
@then("context item with id {ctx_id:d} should not be present")
|
|
def step_verify_context_not_present(context, ctx_id):
|
|
"""Verify context is not present."""
|
|
for ctx in context.all_contexts:
|
|
assert ctx.id != ctx_id
|
|
|
|
|
|
# Removed duplicate - using step_add_multiple_contexts instead
|
|
|
|
|
|
@when("I clear all context for plan {plan_id:d}")
|
|
def step_clear_context(context, plan_id):
|
|
"""Clear all context for plan."""
|
|
context.context_repo.clear_for_plan(plan_id)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("all context items should be deleted")
|
|
def step_verify_all_context_deleted(context):
|
|
"""Verify all context deleted."""
|
|
contexts = context.context_repo.get_for_plan(1)
|
|
assert len(contexts) == 0
|
|
|
|
|
|
@then("I should receive an empty list")
|
|
def step_verify_empty_list(context):
|
|
"""Verify empty list."""
|
|
# Handle both context and change scenarios
|
|
if hasattr(context, "all_contexts"):
|
|
assert len(context.all_contexts) == 0
|
|
elif hasattr(context, "all_changes"):
|
|
assert len(context.all_changes) == 0
|
|
else:
|
|
raise AssertionError("No list to verify (expected all_contexts or all_changes)")
|
|
|
|
|
|
# ChangeRepository scenarios
|
|
@given("I have a ChangeRepository instance")
|
|
def step_create_change_repository(context):
|
|
"""Create ChangeRepository instance."""
|
|
if not hasattr(context, "db_session"):
|
|
step_create_test_session(context)
|
|
context.change_repo = ChangeRepository(context.db_session)
|
|
|
|
|
|
@when('I add a change with file_path "{file_path}"')
|
|
def step_add_change(context, file_path):
|
|
"""Add change."""
|
|
change = Change(
|
|
plan_id=1,
|
|
file_path=file_path, # Keep as string, don't convert to Path
|
|
operation=OperationType.CREATE,
|
|
new_content="new file content",
|
|
created_at=datetime.now(),
|
|
)
|
|
context.created_change = context.change_repo.add(change)
|
|
|
|
|
|
# Removed - conflicts with ChangeModel step
|
|
|
|
|
|
# Removed - conflicts with ChangeModel step
|
|
|
|
|
|
@then("the change should be saved with an assigned ID")
|
|
def step_verify_change_saved(context):
|
|
"""Verify change is saved."""
|
|
context.db_session.commit()
|
|
assert context.created_change.id is not None
|
|
|
|
|
|
@when("I get all changes for plan {plan_id:d}")
|
|
def step_get_all_changes(context, plan_id):
|
|
"""Get all changes for plan."""
|
|
context.all_changes = context.change_repo.get_for_plan(plan_id)
|
|
|
|
|
|
@then("I should receive the added change")
|
|
def step_verify_change_retrieved(context):
|
|
"""Verify change is retrieved."""
|
|
assert len(context.all_changes) > 0
|
|
assert context.all_changes[0].file_path == context.created_change.file_path
|
|
|
|
|
|
@given("I have added a change with id {change_id:d}")
|
|
def step_add_change_with_id(context, change_id):
|
|
"""Add change with ID."""
|
|
if not hasattr(context, "change_repo"):
|
|
step_create_change_repository(context)
|
|
|
|
change = Change(
|
|
id=change_id,
|
|
plan_id=1,
|
|
file_path=f"/change/file{change_id}.py", # Keep as string
|
|
operation=OperationType.CREATE,
|
|
new_content="new content",
|
|
created_at=datetime.now(),
|
|
)
|
|
context.created_change = context.change_repo.add(change)
|
|
context.db_session.commit()
|
|
|
|
|
|
@when("I mark change {change_id:d} as applied")
|
|
def step_mark_change_applied(context, change_id):
|
|
"""Mark change as applied."""
|
|
context.change_repo.mark_applied(change_id)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the change should have applied flag set to True")
|
|
def step_verify_change_applied_flag(context):
|
|
"""Verify change applied flag."""
|
|
from cleveragents.infrastructure.database.models import ChangeModel
|
|
|
|
change = context.db_session.query(ChangeModel).filter_by(id=1).first()
|
|
assert change.applied is True
|
|
|
|
|
|
@then("the change should have applied_at timestamp")
|
|
def step_verify_change_applied_timestamp(context):
|
|
"""Verify change applied timestamp."""
|
|
from cleveragents.infrastructure.database.models import ChangeModel
|
|
|
|
change = context.db_session.query(ChangeModel).filter_by(id=1).first()
|
|
assert change.applied_at is not None
|
|
|
|
|
|
@then("the retrieved change should show as applied")
|
|
def step_verify_retrieved_change_applied(context):
|
|
"""Verify retrieved change is applied."""
|
|
changes = context.change_repo.get_for_plan(1)
|
|
assert len(changes) > 0
|
|
assert changes[0].applied is True
|
|
|
|
|
|
@given("I have added {count:d} changes to the plan")
|
|
def step_add_multiple_changes(context, count):
|
|
"""Add multiple changes."""
|
|
if not hasattr(context, "change_repo"):
|
|
step_create_change_repository(context)
|
|
|
|
context.added_changes = []
|
|
for i in range(count):
|
|
change = Change(
|
|
plan_id=1,
|
|
file_path=f"/change/file{i + 1}.py", # Keep as string
|
|
operation=OperationType.CREATE,
|
|
new_content=f"content {i + 1}",
|
|
created_at=datetime.now(),
|
|
)
|
|
added = context.change_repo.add(change)
|
|
context.added_changes.append(added)
|
|
context.db_session.commit()
|
|
|
|
|
|
@when("I clear all changes for plan {plan_id:d}")
|
|
def step_clear_changes(context, plan_id):
|
|
"""Clear all changes for plan."""
|
|
context.change_repo.clear_for_plan(plan_id)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("all changes should be deleted")
|
|
def step_verify_all_changes_deleted(context):
|
|
"""Verify all changes deleted."""
|
|
changes = context.change_repo.get_for_plan(1)
|
|
assert len(changes) == 0
|
|
|
|
|
|
# Integration scenarios
|
|
@given("I have all repository instances")
|
|
def step_create_all_repositories(context):
|
|
"""Create all repository instances."""
|
|
step_create_test_session(context)
|
|
context.project_repo = ProjectRepository(context.db_session)
|
|
context.plan_repo = PlanRepository(context.db_session)
|
|
context.context_repo = ContextRepository(context.db_session)
|
|
context.change_repo = ChangeRepository(context.db_session)
|
|
|
|
|
|
@given("I have created a project with plans, contexts, and changes")
|
|
def step_create_full_project_data(context):
|
|
"""Create full project data."""
|
|
# Create project
|
|
project = Project(
|
|
name="integration-project",
|
|
path=Path("/tmp/integration"),
|
|
settings=ProjectSettings(),
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
context.project = context.project_repo.create(project)
|
|
|
|
# Create plan
|
|
plan = Plan(
|
|
project_id=context.project.id,
|
|
name="integration-plan",
|
|
prompt="Test prompt",
|
|
status=PlanStatus.PENDING,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
build_started_at=datetime.now(),
|
|
model_used="gpt-4",
|
|
token_count=1000,
|
|
applied_at=datetime.now(),
|
|
files_created=1,
|
|
files_modified=2,
|
|
files_deleted=0,
|
|
)
|
|
context.plan = context.plan_repo.create(plan)
|
|
context.plan_repo.set_current(context.project.id, context.plan.id)
|
|
|
|
# Add context
|
|
ctx = Context(
|
|
plan_id=context.plan.id,
|
|
type=ContextType.FILE,
|
|
path="/test/file.py", # String, not Path
|
|
content="test content",
|
|
added_at=datetime.now(),
|
|
)
|
|
context.context_repo.add(ctx)
|
|
|
|
# Add change
|
|
change = Change(
|
|
plan_id=context.plan.id,
|
|
file_path="/test/changed.py", # String, not Path
|
|
operation=OperationType.MODIFY,
|
|
original_content="old",
|
|
new_content="new",
|
|
created_at=datetime.now(),
|
|
)
|
|
context.change_repo.add(change)
|
|
|
|
context.db_session.commit()
|
|
|
|
|
|
@when("I retrieve the project by ID for integration")
|
|
def step_retrieve_project_integration(context):
|
|
"""Retrieve project for integration test."""
|
|
context.retrieved_project = context.project_repo.get_by_id(context.project.id)
|
|
|
|
|
|
@then("I should get a domain Project model")
|
|
def step_verify_domain_project(context):
|
|
"""Verify domain Project model."""
|
|
assert isinstance(context.retrieved_project, Project)
|
|
assert context.retrieved_project.name == "integration-project"
|
|
|
|
|
|
@when("I retrieve the current plan")
|
|
def step_retrieve_current_plan_integration(context):
|
|
"""Retrieve current plan."""
|
|
context.current_plan = context.plan_repo.get_current(context.project.id)
|
|
|
|
|
|
@then("I should get a domain Plan model with build and result data")
|
|
def step_verify_domain_plan(context):
|
|
"""Verify domain Plan model."""
|
|
assert isinstance(context.current_plan, Plan)
|
|
# Just verify we have a plan with the expected name
|
|
assert context.current_plan.name == "integration-plan"
|
|
# The backward compatibility fields or build/result objects should exist
|
|
assert (context.current_plan.model_used or context.current_plan.build) is not None
|
|
|
|
|
|
@when("I retrieve contexts for the plan")
|
|
def step_retrieve_contexts_integration(context):
|
|
"""Retrieve contexts for plan."""
|
|
context.contexts = context.context_repo.get_for_plan(context.plan.id)
|
|
|
|
|
|
@then("I should get a list of domain Context models")
|
|
def step_verify_domain_contexts(context):
|
|
"""Verify domain Context models."""
|
|
assert len(context.contexts) > 0
|
|
assert isinstance(context.contexts[0], Context)
|
|
assert context.contexts[0].path == "/test/file.py"
|
|
|
|
|
|
@when("I retrieve changes for the plan")
|
|
def step_retrieve_changes_integration(context):
|
|
"""Retrieve changes for plan."""
|
|
context.changes = context.change_repo.get_for_plan(context.plan.id)
|
|
|
|
|
|
@then("I should get a list of domain Change models")
|
|
def step_verify_domain_changes(context):
|
|
"""Verify domain Change models."""
|
|
assert len(context.changes) > 0
|
|
assert isinstance(context.changes[0], Change)
|
|
assert context.changes[0].file_path == "/test/changed.py"
|
|
|
|
|
|
# Database repository test steps
|
|
@given("I have a database session")
|
|
def step_have_database_session(context):
|
|
"""Ensure database session exists."""
|
|
step_create_test_session(context)
|
|
|
|
|
|
@given("I have a project repository")
|
|
def step_have_project_repository(context):
|
|
"""Ensure project repository exists."""
|
|
step_create_project_repository(context)
|
|
|
|
|
|
@when('I create a new project named "{name}"')
|
|
def step_create_new_project(context, name):
|
|
"""Create new project."""
|
|
step_create_project_with_repo(context, name, f"/tmp/{name}")
|
|
|
|
|
|
@then("the project should be saved to database")
|
|
def step_project_saved_to_db(context):
|
|
"""Verify project saved."""
|
|
step_save_project_with_repo(context)
|
|
assert context.created_project.id is not None
|
|
|
|
|
|
@then("I should be able to retrieve the project by ID")
|
|
def step_can_retrieve_project(context):
|
|
"""Verify can retrieve project."""
|
|
step_db_retrieve_project_by_id(context)
|
|
step_verify_retrieved_project(context)
|
|
|
|
|
|
@then("I should be able to retrieve the project by name")
|
|
def step_can_retrieve_project_by_name(context):
|
|
"""Verify can retrieve project by name."""
|
|
retrieved = context.project_repo.get_by_name(context.created_project.name)
|
|
assert retrieved is not None
|
|
assert retrieved.name == context.created_project.name
|
|
|
|
|
|
@given("I have an existing project in the database")
|
|
def step_existing_project_in_db(context):
|
|
"""Create existing project."""
|
|
step_have_project_repository(context)
|
|
step_create_project_with_repo(context, "existing-project", "/tmp/existing")
|
|
step_save_project_with_repo(context)
|
|
|
|
|
|
@when("I update the project settings")
|
|
def step_update_project_settings(context):
|
|
"""Update project settings."""
|
|
context.created_project.settings.auto_build = True
|
|
context.created_project.settings.default_model = "gpt-4"
|
|
context.updated_project = context.project_repo.update(context.created_project)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the changes should be persisted")
|
|
def step_changes_persisted(context):
|
|
"""Verify changes persisted."""
|
|
retrieved = context.project_repo.get_by_id(context.created_project.id)
|
|
assert retrieved.settings.auto_build
|
|
assert retrieved.settings.default_model == "gpt-4"
|
|
|
|
|
|
@then("the updated_at timestamp should change")
|
|
def step_updated_timestamp_changed(context):
|
|
"""Verify updated timestamp changed."""
|
|
assert context.updated_project.updated_at > context.created_project.created_at
|
|
|
|
|
|
@given("I have a plan repository")
|
|
def step_have_plan_repository(context):
|
|
"""Ensure plan repository exists."""
|
|
step_create_plan_repository(context)
|
|
|
|
|
|
@given("I have a project with ID {project_id:d}")
|
|
def step_have_project_with_id(context, project_id):
|
|
"""Ensure project with ID exists."""
|
|
step_ensure_project_exists(context, project_id)
|
|
|
|
|
|
@when("I create a new plan for the project")
|
|
def step_create_new_plan(context):
|
|
"""Create new plan."""
|
|
step_create_plan_with_repo(context, "new-plan", context.project_id)
|
|
|
|
|
|
@then("the plan should be saved to database")
|
|
def step_plan_saved_to_db(context):
|
|
"""Verify plan saved."""
|
|
step_save_plan_with_repo(context)
|
|
assert context.created_plan.id is not None
|
|
|
|
|
|
@then("I should be able to retrieve the plan by ID")
|
|
def step_can_retrieve_plan(context):
|
|
"""Verify can retrieve plan."""
|
|
retrieved = context.plan_repo.get_by_id(context.created_plan.id)
|
|
assert retrieved is not None
|
|
assert retrieved.name == context.created_plan.name
|
|
|
|
|
|
@then("I should be able to get all plans for the project")
|
|
def step_can_get_all_plans(context):
|
|
"""Verify can get all plans."""
|
|
plans = context.plan_repo.get_all_for_project(context.project_id)
|
|
assert len(plans) > 0
|
|
|
|
|
|
@given("I have multiple plans for a project")
|
|
def step_have_multiple_plans(context):
|
|
"""Create multiple plans."""
|
|
step_have_plan_repository(context)
|
|
step_ensure_project_exists(context, 1)
|
|
step_create_multiple_plans(context, 3)
|
|
|
|
|
|
@when("I set one plan as current")
|
|
def step_set_one_current(context):
|
|
"""Set one plan as current."""
|
|
context.plan_repo.set_current(context.project_id, context.created_plans[1].id)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("only that plan should be marked as current")
|
|
def step_only_one_current(context):
|
|
"""Verify only one plan current."""
|
|
current = context.plan_repo.get_current(context.project_id)
|
|
assert current.id == context.created_plans[1].id
|
|
assert current.current is True
|
|
|
|
|
|
@then("other plans should not be current")
|
|
def step_others_not_current(context):
|
|
"""Verify other plans not current."""
|
|
all_plans = context.plan_repo.get_all_for_project(context.project_id)
|
|
current_count = sum(1 for p in all_plans if p.current)
|
|
assert current_count == 1
|
|
|
|
|
|
@given("I have a context repository")
|
|
def step_have_context_repository(context):
|
|
"""Ensure context repository exists."""
|
|
step_create_context_repository(context)
|
|
|
|
|
|
@given("I have a plan with ID {plan_id:d}")
|
|
def step_have_plan_with_id(context, plan_id):
|
|
"""Ensure plan with ID exists."""
|
|
step_given_plan_created(context, plan_id)
|
|
|
|
|
|
@when("I add context items to the plan")
|
|
def step_add_context_items(context):
|
|
"""Add context items."""
|
|
for i in range(3):
|
|
ctx = Context(
|
|
plan_id=1,
|
|
type=ContextType.FILE,
|
|
path=str(Path(f"/test/file{i}.py")), # Convert Path to string
|
|
content=f"content {i}",
|
|
added_at=datetime.now(),
|
|
)
|
|
context.context_repo.add(ctx)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the context should be saved to database")
|
|
def step_context_saved_to_db(context):
|
|
"""Verify context saved."""
|
|
contexts = context.context_repo.get_for_plan(1)
|
|
assert len(contexts) == 3
|
|
|
|
|
|
@then("I should be able to retrieve all context for the plan")
|
|
def step_can_retrieve_all_context(context):
|
|
"""Verify can retrieve all context."""
|
|
contexts = context.context_repo.get_for_plan(1)
|
|
assert len(contexts) == 3
|
|
assert all(isinstance(c, Context) for c in contexts)
|
|
|
|
|
|
@given("I have a plan with context items")
|
|
def step_have_plan_with_context(context):
|
|
"""Create plan with context."""
|
|
step_have_context_repository(context)
|
|
step_have_plan_with_id(context, 1)
|
|
step_add_context_items(context)
|
|
|
|
|
|
@when("I clear the context for the plan")
|
|
def step_clear_plan_context(context):
|
|
"""Clear context for plan."""
|
|
context.context_repo.clear_for_plan(1)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the plan should have no context items")
|
|
def step_plan_no_context(context):
|
|
"""Verify plan has no context."""
|
|
contexts = context.context_repo.get_for_plan(1)
|
|
assert len(contexts) == 0
|
|
|
|
|
|
@given("I have a change repository")
|
|
def step_have_change_repository(context):
|
|
"""Ensure change repository exists."""
|
|
step_create_change_repository(context)
|
|
|
|
|
|
@when("I add changes to the plan")
|
|
def step_add_changes_to_plan(context):
|
|
"""Add changes to plan."""
|
|
for i in range(2):
|
|
change = Change(
|
|
plan_id=1,
|
|
file_path=str(Path(f"/test/change{i}.py")), # Convert Path to string
|
|
operation=OperationType.CREATE,
|
|
new_content=f"new content {i}",
|
|
created_at=datetime.now(),
|
|
)
|
|
context.change_repo.add(change)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the changes should be saved to database")
|
|
def step_changes_saved_to_db(context):
|
|
"""Verify changes saved."""
|
|
changes = context.change_repo.get_for_plan(1)
|
|
assert len(changes) == 2
|
|
|
|
|
|
@then("I should be able to retrieve all changes for the plan")
|
|
def step_can_retrieve_all_changes(context):
|
|
"""Verify can retrieve all changes."""
|
|
changes = context.change_repo.get_for_plan(1)
|
|
assert len(changes) == 2
|
|
assert all(isinstance(c, Change) for c in changes)
|
|
|
|
|
|
@given("I have unapplied changes")
|
|
def step_have_unapplied_changes(context):
|
|
"""Create unapplied changes."""
|
|
step_have_change_repository(context)
|
|
step_have_plan_with_id(context, 1)
|
|
change = Change(
|
|
plan_id=1,
|
|
file_path=str(Path("/test/unapplied.py")), # Convert Path to string
|
|
operation=OperationType.CREATE,
|
|
new_content="new content",
|
|
created_at=datetime.now(),
|
|
)
|
|
context.unapplied_change = context.change_repo.add(change)
|
|
context.db_session.commit()
|
|
|
|
|
|
@when("I mark a change as applied")
|
|
def step_mark_one_change_applied(context):
|
|
"""Mark one change as applied."""
|
|
context.change_repo.mark_applied(context.unapplied_change.id)
|
|
context.db_session.commit()
|
|
|
|
|
|
@then("the change should be marked as applied")
|
|
def step_change_marked_applied(context):
|
|
"""Verify change marked applied."""
|
|
changes = context.change_repo.get_for_plan(1)
|
|
applied = next((c for c in changes if c.id == context.unapplied_change.id), None)
|
|
assert applied is not None
|
|
assert applied.applied is True
|
|
|
|
|
|
@then("the applied_at timestamp should be set")
|
|
def step_applied_timestamp_set(context):
|
|
"""Verify applied timestamp set."""
|
|
changes = context.change_repo.get_for_plan(1)
|
|
applied = next((c for c in changes if c.id == context.unapplied_change.id), None)
|
|
assert applied.applied_at is not None
|
|
|
|
|
|
@when("I attempt a failing transaction")
|
|
def step_attempt_failing_transaction(context):
|
|
"""Attempt a transaction that will fail."""
|
|
# Create UnitOfWork from existing database session
|
|
if not hasattr(context, "unit_of_work"):
|
|
from cleveragents.infrastructure.database.unit_of_work import (
|
|
UnitOfWork,
|
|
)
|
|
|
|
# Use existing engine from context to create UnitOfWork
|
|
# This ensures we use the same database that was initialized in previous steps
|
|
if not hasattr(context, "engine"):
|
|
step_create_test_session(context)
|
|
|
|
# Create a UnitOfWork but share the existing engine
|
|
context.unit_of_work = UnitOfWork("sqlite:///:memory:")
|
|
# Replace its engine with the existing one
|
|
context.unit_of_work._engine = context.engine
|
|
context.unit_of_work._session_factory = context.SessionLocal
|
|
|
|
try:
|
|
with context.unit_of_work.transaction() as ctx:
|
|
# Create a project
|
|
project = Project(
|
|
name="rollback-test",
|
|
path=Path("/tmp/rollback"),
|
|
settings=ProjectSettings(),
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
created = ctx.projects.create(project)
|
|
context.rollback_project_id = created.id
|
|
|
|
# Force an error
|
|
raise ValueError("Forced error for testing rollback")
|
|
except ValueError:
|
|
pass # Expected error
|
|
|
|
|
|
@then("the transaction should be rolled back")
|
|
def step_transaction_rolled_back(context):
|
|
"""Verify transaction rolled back."""
|
|
# Try to find the project that should have been rolled back
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("rollback-test")
|
|
assert project is None
|
|
|
|
|
|
@then("no data should be persisted")
|
|
def step_no_data_persisted(context):
|
|
"""Verify no data persisted."""
|
|
# Already verified in previous step
|
|
pass
|