HAL9000
1970fae07b
CI / lint (pull_request) Successful in 39s
CI / typecheck (pull_request) Successful in 49s
CI / push-validation (pull_request) Successful in 33s
CI / helm (pull_request) Successful in 38s
CI / build (pull_request) Successful in 45s
CI / quality (pull_request) Successful in 1m31s
CI / security (pull_request) Successful in 1m38s
CI / unit_tests (pull_request) Successful in 6m57s
CI / docker (pull_request) Successful in 1m33s
CI / coverage (pull_request) Successful in 13m45s
CI / integration_tests (pull_request) Successful in 21m36s
CI / status-check (pull_request) Successful in 3s
Fix formatting to satisfy ruff format --check CI gate.
1102 lines
40 KiB
Python
1102 lines
40 KiB
Python
"""Step definitions for legacy migrator coverage tests."""
|
|
|
|
import json
|
|
import tempfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from sqlalchemy import create_engine
|
|
|
|
from cleveragents.domain.models.core import (
|
|
ContextType,
|
|
OperationType,
|
|
PlanStatus,
|
|
Project,
|
|
ProjectSettings,
|
|
)
|
|
from cleveragents.infrastructure.database.legacy_migrator import (
|
|
LegacyDataMigrator,
|
|
check_and_migrate_legacy_data,
|
|
)
|
|
from cleveragents.infrastructure.database.models import Base
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
|
|
|
|
@given("I have a temporary test directory for legacy migration")
|
|
def step_create_temp_directory_legacy(context: Context) -> None:
|
|
"""Create a temporary directory for testing."""
|
|
context.temp_dir = Path(tempfile.mkdtemp(prefix="test_legacy_"))
|
|
|
|
|
|
@given("I have a Unit of Work instance for testing")
|
|
def step_create_unit_of_work(context: Context) -> None:
|
|
"""Create a Unit of Work instance with in-memory database."""
|
|
context.database_url = "sqlite:///:memory:"
|
|
context.unit_of_work = UnitOfWork(context.database_url)
|
|
|
|
# Initialize the database schema
|
|
engine = create_engine(context.database_url)
|
|
Base.metadata.create_all(engine)
|
|
context.unit_of_work.init_database()
|
|
|
|
|
|
@given("I have a legacy project with all JSON files")
|
|
def step_create_legacy_project_full(context: Context) -> None:
|
|
"""Create a legacy project with all JSON files."""
|
|
project_dir = context.temp_dir / "test_project"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
# Create project.name file
|
|
(cleveragents_dir / "project.name").write_text("test_project")
|
|
|
|
# Create current file
|
|
(cleveragents_dir / "current").write_text("main")
|
|
|
|
# Create plans.json
|
|
plans_data = {
|
|
"main": {
|
|
"prompt": "Initial plan",
|
|
"status": "built",
|
|
"current": True,
|
|
"created_at": "2024-01-01T10:00:00",
|
|
"updated_at": "2024-01-01T10:00:00",
|
|
"build": {"changes": [], "summary": "Test build", "error": None},
|
|
"build_started_at": "2024-01-01T10:00:00",
|
|
"build_completed_at": "2024-01-01T10:01:00",
|
|
"model_used": "mock-gpt",
|
|
"token_count": 100,
|
|
"result": "Success",
|
|
"applied_at": None,
|
|
"files_created": 2,
|
|
"files_modified": 3,
|
|
"files_deleted": 1,
|
|
},
|
|
"feature": {
|
|
"prompt": "Add feature",
|
|
"status": "pending",
|
|
"current": False,
|
|
"created_at": "2024-01-02T10:00:00",
|
|
"updated_at": "2024-01-02T10:00:00",
|
|
},
|
|
}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
# Create contexts.json
|
|
contexts_data = {
|
|
"src/main.py": {
|
|
"content": "print('hello')",
|
|
"file_hash": "abc123",
|
|
"type": "file",
|
|
"added_at": "2024-01-01T10:00:00",
|
|
},
|
|
"src/utils.py": {
|
|
"content": "def util(): pass",
|
|
"file_hash": "def456",
|
|
"type": "file",
|
|
"added_at": "2024-01-01T10:00:00",
|
|
},
|
|
}
|
|
(cleveragents_dir / "contexts.json").write_text(json.dumps(contexts_data))
|
|
|
|
# Create changes.json
|
|
changes_data = [
|
|
{
|
|
"file_path": "src/main.py",
|
|
"operation": "create",
|
|
"original_content": None,
|
|
"new_content": "print('hello')",
|
|
"new_path": None,
|
|
"applied": False,
|
|
"applied_at": None,
|
|
"created_at": "2024-01-01T10:00:00",
|
|
},
|
|
{
|
|
"file_path": "src/old.py",
|
|
"operation": "delete",
|
|
"original_content": "old code",
|
|
"new_content": None,
|
|
"new_path": None,
|
|
"applied": True,
|
|
"applied_at": "2024-01-01T10:05:00",
|
|
"created_at": "2024-01-01T10:00:00",
|
|
},
|
|
]
|
|
(cleveragents_dir / "changes.json").write_text(json.dumps(changes_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a project path without .cleveragents directory")
|
|
def step_create_project_without_cleveragents(context: Context) -> None:
|
|
"""Create a project path without .cleveragents directory."""
|
|
project_dir = context.temp_dir / "no_cleveragents"
|
|
project_dir.mkdir(exist_ok=True)
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a project with .cleveragents directory but no JSON files")
|
|
def step_create_project_without_json(context: Context) -> None:
|
|
"""Create a project with .cleveragents directory but no JSON files."""
|
|
project_dir = context.temp_dir / "no_json"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
# Create only non-JSON files
|
|
(cleveragents_dir / "project.name").write_text("test_project")
|
|
(cleveragents_dir / "current").write_text("main")
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project with only plans.json file")
|
|
def step_create_project_only_plans(context: Context) -> None:
|
|
"""Create a legacy project with only plans.json."""
|
|
project_dir = context.temp_dir / "only_plans"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
plans_data = {
|
|
"main": {
|
|
"prompt": "Initial plan",
|
|
"status": "pending",
|
|
"current": True,
|
|
"created_at": "2024-01-01T10:00:00",
|
|
"updated_at": "2024-01-01T10:00:00",
|
|
}
|
|
}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project with only contexts.json file for current plan")
|
|
def step_create_project_only_contexts(context: Context) -> None:
|
|
"""Create a legacy project with only contexts.json."""
|
|
project_dir = context.temp_dir / "only_contexts"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "current").write_text("main")
|
|
(cleveragents_dir / "project.name").write_text("test_project")
|
|
|
|
# Create a minimal plans.json so migration can map the plan
|
|
plans_data = {
|
|
"main": {
|
|
"prompt": "Test plan",
|
|
"status": "pending",
|
|
"current": True,
|
|
"created_at": "2024-01-01T10:00:00",
|
|
}
|
|
}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
# Create contexts.json
|
|
contexts_data = {
|
|
"src/file.py": {
|
|
"content": "test content",
|
|
"file_hash": "xyz789",
|
|
"type": "file",
|
|
"added_at": "2024-01-01T10:00:00",
|
|
}
|
|
}
|
|
(cleveragents_dir / "contexts.json").write_text(json.dumps(contexts_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project with only changes.json file for current plan")
|
|
def step_create_project_only_changes(context: Context) -> None:
|
|
"""Create a legacy project with only changes.json."""
|
|
project_dir = context.temp_dir / "only_changes"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "current").write_text("main")
|
|
(cleveragents_dir / "project.name").write_text("test_project")
|
|
|
|
# Create a minimal plans.json so migration can map the plan
|
|
plans_data = {
|
|
"main": {
|
|
"prompt": "Test plan",
|
|
"status": "pending",
|
|
"current": True,
|
|
"created_at": "2024-01-01T10:00:00",
|
|
}
|
|
}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
# Create changes.json
|
|
changes_data = [
|
|
{
|
|
"file_path": "src/new.py",
|
|
"operation": "create",
|
|
"new_content": "new code",
|
|
"created_at": "2024-01-01T10:00:00",
|
|
}
|
|
]
|
|
(cleveragents_dir / "changes.json").write_text(json.dumps(changes_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have an existing project in the database for legacy migration")
|
|
def step_create_existing_project(context: Context) -> None:
|
|
"""Create an existing project in the database."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = Project(
|
|
id=None,
|
|
name="existing_project",
|
|
path=Path("/tmp/existing"),
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="mock-gpt",
|
|
),
|
|
)
|
|
context.existing_project = ctx.projects.create(project)
|
|
|
|
|
|
@given("I have a legacy project with matching name")
|
|
def step_create_legacy_with_matching_name(context: Context) -> None:
|
|
"""Create a legacy project with name matching existing database project."""
|
|
project_dir = context.temp_dir / "existing_project"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("existing_project")
|
|
|
|
plans_data = {
|
|
"main": {
|
|
"prompt": "Test plan",
|
|
"status": "pending",
|
|
"current": True,
|
|
"created_at": "2024-01-01T10:00:00",
|
|
}
|
|
}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project without database entry")
|
|
def step_create_legacy_without_db(context: Context) -> None:
|
|
"""Create a legacy project without database entry."""
|
|
project_dir = context.temp_dir / "new_project"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("new_project")
|
|
|
|
plans_data = {"main": {"prompt": "New plan", "status": "pending"}}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project without project.name file")
|
|
def step_create_legacy_without_name_file(context: Context) -> None:
|
|
"""Create a legacy project without project.name file."""
|
|
project_dir = context.temp_dir / "unnamed_project"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
plans_data = {"main": {"prompt": "Test", "status": "pending"}}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project with plans already in database")
|
|
def step_create_project_with_existing_plans(context: Context) -> None:
|
|
"""Create a project with some plans already in database."""
|
|
project_dir = context.temp_dir / "existing_plans"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("existing_plans")
|
|
|
|
# Create project and plan in database first
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = Project(
|
|
id=None,
|
|
name="existing_plans",
|
|
path=project_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="mock-gpt",
|
|
),
|
|
)
|
|
project = ctx.projects.create(project)
|
|
|
|
from cleveragents.domain.models.core import Plan
|
|
|
|
plan = Plan(
|
|
id=None,
|
|
project_id=project.id,
|
|
name="main",
|
|
prompt="Existing plan",
|
|
status=PlanStatus.BUILT,
|
|
current=True,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
build=None,
|
|
build_started_at=None,
|
|
build_completed_at=None,
|
|
model_used=None,
|
|
token_count=None,
|
|
result=None,
|
|
applied_at=None,
|
|
files_created=None,
|
|
files_modified=None,
|
|
files_deleted=None,
|
|
)
|
|
ctx.plans.create(plan)
|
|
|
|
# Create plans.json with both existing and new plans
|
|
plans_data = {
|
|
"main": {"prompt": "Existing plan", "status": "built"},
|
|
"new_plan": {"prompt": "New plan to migrate", "status": "pending"},
|
|
}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a LegacyDataMigrator instance")
|
|
def step_create_migrator(context: Context) -> None:
|
|
"""Create a LegacyDataMigrator instance."""
|
|
if not hasattr(context, "unit_of_work"):
|
|
step_create_unit_of_work(context)
|
|
context.migrator = LegacyDataMigrator(context.unit_of_work)
|
|
|
|
|
|
@given("I have a legacy project with multiple plans")
|
|
def step_create_project_multiple_plans(context: Context) -> None:
|
|
"""Create a legacy project with multiple plans."""
|
|
project_dir = context.temp_dir / "multi_plans"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("multi_plans")
|
|
|
|
plans_data = {
|
|
"main": {"prompt": "Main plan", "status": "built", "current": False},
|
|
"feature": {"prompt": "Feature plan", "status": "pending", "current": False},
|
|
"bugfix": {"prompt": "Bugfix plan", "status": "applied", "current": False},
|
|
}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("one plan is marked as current")
|
|
def step_mark_plan_current(context: Context) -> None:
|
|
"""Mark one plan as current."""
|
|
cleveragents_dir = context.project_path / ".cleveragents"
|
|
|
|
# Update the plans.json to mark feature as current
|
|
plans_data = {
|
|
"main": {"prompt": "Main plan", "status": "built", "current": False},
|
|
"feature": {
|
|
"prompt": "Feature plan",
|
|
"status": "pending",
|
|
"current": True, # Mark this one as current
|
|
},
|
|
"bugfix": {"prompt": "Bugfix plan", "status": "applied", "current": False},
|
|
}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
(cleveragents_dir / "current").write_text("feature")
|
|
|
|
|
|
@given("I have a legacy project with contexts but no current plan")
|
|
def step_create_project_contexts_no_current(context: Context) -> None:
|
|
"""Create a legacy project with contexts but no current plan."""
|
|
project_dir = context.temp_dir / "contexts_no_current"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("contexts_no_current")
|
|
|
|
# Don't create current file
|
|
|
|
contexts_data = {"src/file.py": {"content": "test", "type": "file"}}
|
|
(cleveragents_dir / "contexts.json").write_text(json.dumps(contexts_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project with changes but no current plan")
|
|
def step_create_project_changes_no_current(context: Context) -> None:
|
|
"""Create a legacy project with changes but no current plan."""
|
|
project_dir = context.temp_dir / "changes_no_current"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("changes_no_current")
|
|
|
|
# Don't create current file
|
|
|
|
changes_data = [
|
|
{"file_path": "src/file.py", "operation": "create", "new_content": "test"}
|
|
]
|
|
(cleveragents_dir / "changes.json").write_text(json.dumps(changes_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project with contexts to migrate")
|
|
def step_create_project_with_contexts(context: Context) -> None:
|
|
"""Create a project with contexts to migrate."""
|
|
step_create_project_only_contexts(context)
|
|
|
|
|
|
@given("I have a legacy project with plan containing all fields")
|
|
def step_create_project_plan_all_fields(context: Context) -> None:
|
|
"""Create a legacy project with plan containing all fields."""
|
|
project_dir = context.temp_dir / "plan_all_fields"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("plan_all_fields")
|
|
|
|
plans_data = {
|
|
"main": {
|
|
"prompt": "Complete plan",
|
|
"status": "built",
|
|
"current": True,
|
|
"created_at": "2024-01-01T10:00:00",
|
|
"updated_at": "2024-01-02T10:00:00",
|
|
"build": {
|
|
"changes": ["change1", "change2"],
|
|
"summary": "Built successfully",
|
|
"error": None,
|
|
},
|
|
"build_started_at": "2024-01-01T10:00:00",
|
|
"build_completed_at": "2024-01-01T10:05:00",
|
|
"model_used": "gpt-4",
|
|
"token_count": 1500,
|
|
"result": "Success",
|
|
"applied_at": "2024-01-01T10:10:00",
|
|
"files_created": 5,
|
|
"files_modified": 10,
|
|
"files_deleted": 2,
|
|
}
|
|
}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project with plan containing minimal fields")
|
|
def step_create_project_plan_minimal(context: Context) -> None:
|
|
"""Create a legacy project with plan containing minimal fields."""
|
|
project_dir = context.temp_dir / "plan_minimal"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("plan_minimal")
|
|
|
|
plans_data = {"main": {"prompt": "Minimal plan"}}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project with JSON files")
|
|
def step_create_legacy_for_function(context: Context) -> None:
|
|
"""Create a legacy project for testing the convenience function."""
|
|
step_create_legacy_project_full(context)
|
|
|
|
|
|
@given("I have a legacy project with invalid JSON files")
|
|
def step_create_project_invalid_json(context: Context) -> None:
|
|
"""Create a legacy project with invalid JSON files."""
|
|
project_dir = context.temp_dir / "invalid_json"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("invalid_json")
|
|
|
|
# Write invalid JSON
|
|
(cleveragents_dir / "plans.json").write_text("{invalid json content")
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@given("I have a legacy project with read-only JSON files")
|
|
def step_create_project_readonly_json(context: Context) -> None:
|
|
"""Create a legacy project with read-only JSON files."""
|
|
project_dir = context.temp_dir / "readonly_json"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("readonly_json")
|
|
|
|
plans_data = {"main": {"prompt": "Test", "status": "pending"}}
|
|
plans_file = cleveragents_dir / "plans.json"
|
|
plans_file.write_text(json.dumps(plans_data))
|
|
|
|
# Make the file read-only
|
|
import os
|
|
import stat
|
|
|
|
current = os.stat(plans_file)
|
|
os.chmod(
|
|
plans_file, current.st_mode & ~stat.S_IWUSR & ~stat.S_IWGRP & ~stat.S_IWOTH
|
|
)
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@when("I run the legacy data migration")
|
|
def step_run_migration(context: Context) -> None:
|
|
"""Run the legacy data migration."""
|
|
migrator = LegacyDataMigrator(context.unit_of_work)
|
|
context.migration_result = migrator.migrate_project_data(context.project_path)
|
|
|
|
|
|
@when('I map plan status "{status}" to enum')
|
|
def step_map_plan_status(context: Context, status: str) -> None:
|
|
"""Map a plan status string to enum."""
|
|
context.mapped_result = context.migrator._map_plan_status(status)
|
|
|
|
|
|
@when("I map plan build data with valid dictionary")
|
|
def step_map_plan_build_valid(context: Context) -> None:
|
|
"""Map valid plan build data."""
|
|
build_data = {
|
|
"changes": ["change1"],
|
|
"summary": "Test summary",
|
|
"error": "Some error",
|
|
}
|
|
context.mapped_result = context.migrator._map_plan_build(build_data)
|
|
|
|
|
|
@when("I map plan build data with None")
|
|
def step_map_plan_build_none(context: Context) -> None:
|
|
"""Map None plan build data."""
|
|
context.mapped_result = context.migrator._map_plan_build(None)
|
|
|
|
|
|
@when("I map plan build data with empty dictionary")
|
|
def step_map_plan_build_empty(context: Context) -> None:
|
|
"""Map empty dictionary plan build data."""
|
|
context.mapped_result = context.migrator._map_plan_build({})
|
|
|
|
|
|
@when("I map plan build data with non-dict value")
|
|
def step_map_plan_build_nondict(context: Context) -> None:
|
|
"""Map non-dict plan build data."""
|
|
context.mapped_result = context.migrator._map_plan_build("not a dict")
|
|
|
|
|
|
@when('I map context type "{type_str}" to enum')
|
|
def step_map_context_type(context: Context, type_str: str) -> None:
|
|
"""Map a context type string to enum."""
|
|
context.mapped_result = context.migrator._map_context_type(type_str)
|
|
|
|
|
|
@when('I map operation "{op_str}" to enum')
|
|
def step_map_operation(context: Context, op_str: str) -> None:
|
|
"""Map an operation string to enum."""
|
|
context.mapped_result = context.migrator._map_operation(op_str)
|
|
|
|
|
|
@when('I parse datetime string "{dt_str}"')
|
|
def step_parse_datetime(context: Context, dt_str: str) -> None:
|
|
"""Parse a datetime string."""
|
|
context.mapped_result = context.migrator._parse_datetime(dt_str)
|
|
|
|
|
|
@when("I parse datetime string None")
|
|
def step_parse_datetime_none(context: Context) -> None:
|
|
"""Parse None as datetime."""
|
|
context.mapped_result = context.migrator._parse_datetime(None)
|
|
|
|
|
|
@when('I parse datetime string ""')
|
|
def step_parse_datetime_empty(context: Context) -> None:
|
|
"""Parse empty string as datetime."""
|
|
context.mapped_result = context.migrator._parse_datetime("")
|
|
|
|
|
|
@when("I call check_and_migrate_legacy_data function")
|
|
def step_call_check_and_migrate(context: Context) -> None:
|
|
"""Call the check_and_migrate_legacy_data function."""
|
|
context.migration_result = check_and_migrate_legacy_data(
|
|
context.project_path, context.unit_of_work
|
|
)
|
|
|
|
|
|
@then("the migration should return true")
|
|
def step_migration_returns_true(context: Context) -> None:
|
|
"""Check that migration returned true."""
|
|
assert context.migration_result is True, (
|
|
f"Expected True, got {context.migration_result}"
|
|
)
|
|
|
|
|
|
@then("the migration should return false")
|
|
def step_migration_returns_false(context: Context) -> None:
|
|
"""Check that migration returned false."""
|
|
assert context.migration_result is False, (
|
|
f"Expected False, got {context.migration_result}"
|
|
)
|
|
|
|
|
|
@then("the JSON files should be backed up")
|
|
def step_check_json_backup(context: Context) -> None:
|
|
"""Check that JSON files were backed up."""
|
|
cleveragents_dir = context.project_path / ".cleveragents"
|
|
|
|
# Check for backup files
|
|
assert (cleveragents_dir / "plans.json.backup").exists()
|
|
assert (cleveragents_dir / "contexts.json.backup").exists()
|
|
assert (cleveragents_dir / "changes.json.backup").exists()
|
|
|
|
|
|
@then("the plans.json file should be backed up")
|
|
def step_check_plans_backup(context: Context) -> None:
|
|
"""Check that plans.json was backed up."""
|
|
cleveragents_dir = context.project_path / ".cleveragents"
|
|
assert (cleveragents_dir / "plans.json.backup").exists()
|
|
|
|
|
|
@then("the contexts.json file should be backed up")
|
|
def step_check_contexts_backup(context: Context) -> None:
|
|
"""Check that contexts.json was backed up."""
|
|
cleveragents_dir = context.project_path / ".cleveragents"
|
|
assert (cleveragents_dir / "contexts.json.backup").exists()
|
|
|
|
|
|
@then("the changes.json file should be backed up")
|
|
def step_check_changes_backup(context: Context) -> None:
|
|
"""Check that changes.json was backed up."""
|
|
cleveragents_dir = context.project_path / ".cleveragents"
|
|
assert (cleveragents_dir / "changes.json.backup").exists()
|
|
|
|
|
|
@then("the database should contain the migrated project data")
|
|
def step_check_project_in_db(context: Context) -> None:
|
|
"""Check that project was migrated to database."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("test_project")
|
|
assert project is not None
|
|
assert project.name == "test_project"
|
|
|
|
|
|
@then("the database should contain the migrated plans")
|
|
def step_check_plans_in_db(context: Context) -> None:
|
|
"""Check that plans were migrated to database."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("test_project")
|
|
if project:
|
|
plans = ctx.plans.get_all_for_project(project.id)
|
|
assert len(plans) > 0
|
|
|
|
|
|
@then("the database should contain the migrated contexts")
|
|
def step_check_contexts_in_db(context: Context) -> None:
|
|
"""Check that contexts were migrated to database."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("test_project")
|
|
if project:
|
|
plan = ctx.plans.get_current_for_project(project.id)
|
|
if plan:
|
|
contexts = ctx.contexts.get_for_plan(plan.id)
|
|
assert len(contexts) > 0
|
|
|
|
|
|
@then("the database should contain the migrated changes")
|
|
def step_check_changes_in_db(context: Context) -> None:
|
|
"""Check that changes were migrated to database."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("test_project")
|
|
if project:
|
|
plan = ctx.plans.get_current_for_project(project.id)
|
|
if plan:
|
|
changes = ctx.changes.get_for_plan(plan.id)
|
|
assert len(changes) > 0
|
|
|
|
|
|
@then("the migration should use the existing project")
|
|
def step_check_existing_project_used(context: Context) -> None:
|
|
"""Check that existing project was used."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
projects = ctx.projects.get_all()
|
|
# Should still only have one project
|
|
assert len(projects) == 1
|
|
assert projects[0].name == "existing_project"
|
|
|
|
|
|
@then("a new project should be created in the database")
|
|
def step_check_new_project_created(context: Context) -> None:
|
|
"""Check that a new project was created."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("new_project")
|
|
assert project is not None
|
|
|
|
|
|
@then("the project name should use the directory name")
|
|
def step_check_directory_name_used(context: Context) -> None:
|
|
"""Check that directory name was used as project name."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("unnamed_project")
|
|
assert project is not None
|
|
|
|
|
|
@then("existing plans should not be duplicated")
|
|
def step_check_no_duplicate_plans(context: Context) -> None:
|
|
"""Check that existing plans were not duplicated."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("existing_plans")
|
|
if project:
|
|
plans = ctx.plans.get_all_for_project(project.id)
|
|
# Check that main plan appears only once
|
|
main_plans = [p for p in plans if p.name == "main"]
|
|
assert len(main_plans) == 1
|
|
|
|
|
|
@then("the migrator result should be PlanStatus.PENDING")
|
|
def step_check_status_pending(context: Context) -> None:
|
|
"""Check that result is PlanStatus.PENDING."""
|
|
assert context.mapped_result == PlanStatus.PENDING
|
|
|
|
|
|
@then("the migrator result should be PlanStatus.BUILDING")
|
|
def step_check_status_building(context: Context) -> None:
|
|
"""Check that result is PlanStatus.BUILDING."""
|
|
assert context.mapped_result == PlanStatus.BUILDING
|
|
|
|
|
|
@then("the migrator result should be PlanStatus.BUILT")
|
|
def step_check_status_built(context: Context) -> None:
|
|
"""Check that result is PlanStatus.BUILT."""
|
|
assert context.mapped_result == PlanStatus.BUILT
|
|
|
|
|
|
@then("the migrator result should be PlanStatus.ERROR")
|
|
def step_check_status_failed(context: Context) -> None:
|
|
"""Check that result is PlanStatus.FAILED."""
|
|
assert context.mapped_result == PlanStatus.ERROR
|
|
|
|
|
|
@then("the migrator result should be PlanStatus.APPLIED")
|
|
def step_check_status_applied(context: Context) -> None:
|
|
"""Check that result is PlanStatus.APPLIED."""
|
|
assert context.mapped_result == PlanStatus.APPLIED
|
|
|
|
|
|
@then("the migrator result should be None")
|
|
def step_check_result_none(context: Context) -> None:
|
|
"""Check that result is None."""
|
|
assert context.mapped_result is None
|
|
|
|
|
|
@then("the migrator result should be ContextType.FILE")
|
|
def step_check_context_type_file(context: Context) -> None:
|
|
"""Check that result is ContextType.FILE."""
|
|
assert context.mapped_result == ContextType.FILE
|
|
|
|
|
|
@then("the migrator result should be ContextType.DIRECTORY")
|
|
def step_check_context_type_directory(context: Context) -> None:
|
|
"""Check that result is ContextType.DIRECTORY."""
|
|
assert context.mapped_result == ContextType.DIRECTORY
|
|
|
|
|
|
@then("the migrator result should be ContextType.URL")
|
|
def step_check_context_type_url(context: Context) -> None:
|
|
"""Check that result is ContextType.URL."""
|
|
assert context.mapped_result == ContextType.URL
|
|
|
|
|
|
@then("the migrator result should be ContextType.NOTE")
|
|
def step_check_context_type_note(context: Context) -> None:
|
|
"""Check that result is ContextType.NOTE."""
|
|
assert context.mapped_result == ContextType.NOTE
|
|
|
|
|
|
@then("the migrator result should be OperationType.CREATE")
|
|
def step_check_operation_create(context: Context) -> None:
|
|
"""Check that result is OperationType.CREATE."""
|
|
assert context.mapped_result == OperationType.CREATE
|
|
|
|
|
|
@then("the migrator result should be OperationType.MODIFY")
|
|
def step_check_operation_modify(context: Context) -> None:
|
|
"""Check that result is OperationType.MODIFY."""
|
|
assert context.mapped_result == OperationType.MODIFY
|
|
|
|
|
|
@then("the migrator result should be OperationType.DELETE")
|
|
def step_check_operation_delete(context: Context) -> None:
|
|
"""Check that result is OperationType.DELETE."""
|
|
assert context.mapped_result == OperationType.DELETE
|
|
|
|
|
|
@then("the migrator result should be OperationType.MOVE")
|
|
def step_check_operation_move(context: Context) -> None:
|
|
"""Check that result is OperationType.MOVE."""
|
|
assert context.mapped_result == OperationType.MOVE
|
|
|
|
|
|
@then("the migrator result should be a valid datetime object")
|
|
def step_check_datetime_valid(context: Context) -> None:
|
|
"""Check that result is a valid datetime."""
|
|
assert isinstance(context.mapped_result, datetime)
|
|
|
|
|
|
@then("the current plan should be set in the project")
|
|
def step_check_current_plan_set(context: Context) -> None:
|
|
"""Check that current plan was set in project."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("multi_plans")
|
|
if project:
|
|
current_plan = ctx.plans.get_current_for_project(project.id)
|
|
assert current_plan is not None
|
|
assert current_plan.name == "feature"
|
|
|
|
|
|
@then("the contexts should not be migrated")
|
|
def step_check_contexts_not_migrated(context: Context) -> None:
|
|
"""Check that contexts were not migrated."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
# Since there's no current plan, contexts should not be migrated
|
|
# Check by verifying the project has no current plan and no contexts
|
|
project = ctx.projects.get_by_name("contexts_no_current")
|
|
if project:
|
|
current_plan = ctx.plans.get_current_for_project(project.id)
|
|
# If there's no current plan, contexts should not have been created
|
|
if current_plan:
|
|
contexts = ctx.contexts.get_for_plan(current_plan.id)
|
|
assert len(contexts) == 0
|
|
else:
|
|
# No current plan exists, which is correct
|
|
assert True
|
|
|
|
|
|
@then("the changes should not be migrated")
|
|
def step_check_changes_not_migrated(context: Context) -> None:
|
|
"""Check that changes were not migrated."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
# Since there's no current plan, changes should not be migrated
|
|
# Check by verifying the project has no current plan and no changes
|
|
project = ctx.projects.get_by_name("changes_no_current")
|
|
if project:
|
|
current_plan = ctx.plans.get_current_for_project(project.id)
|
|
# If there's no current plan, changes should not have been created
|
|
if current_plan:
|
|
changes = ctx.changes.get_for_plan(current_plan.id)
|
|
assert len(changes) == 0
|
|
else:
|
|
# No current plan exists, which is correct
|
|
assert True
|
|
|
|
|
|
@then("the contexts should be migrated successfully")
|
|
def step_check_contexts_migrated(context: Context) -> None:
|
|
"""Check that contexts were migrated successfully."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
contexts = ctx.contexts.get_all()
|
|
# Should have at least one context
|
|
assert len(contexts) > 0
|
|
|
|
|
|
@then("all plan fields should be migrated correctly")
|
|
def step_check_all_fields_migrated(context: Context) -> None:
|
|
"""Check that all plan fields were migrated."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("plan_all_fields")
|
|
if project:
|
|
plans = ctx.plans.get_all_for_project(project.id)
|
|
plan = next((p for p in plans if p.name == "main"), None)
|
|
assert plan is not None
|
|
assert plan.prompt == "Complete plan"
|
|
assert plan.status == PlanStatus.BUILT
|
|
assert plan.current is True
|
|
assert plan.model_used == "gpt-4"
|
|
assert plan.token_count == 1500
|
|
assert plan.files_created == 5
|
|
assert plan.files_modified == 10
|
|
assert plan.files_deleted == 2
|
|
|
|
|
|
@then("the plan should be migrated with defaults")
|
|
def step_check_plan_with_defaults(context: Context) -> None:
|
|
"""Check that plan was migrated with default values."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("plan_minimal")
|
|
if project:
|
|
plans = ctx.plans.get_all_for_project(project.id)
|
|
plan = next((p for p in plans if p.name == "main"), None)
|
|
assert plan is not None
|
|
assert plan.prompt == "Minimal plan"
|
|
assert plan.status == PlanStatus.PENDING # Default status
|
|
|
|
|
|
@then("the function should return true")
|
|
def step_function_returns_true(context: Context) -> None:
|
|
"""Check that function returned true."""
|
|
assert context.migration_result is True
|
|
|
|
|
|
@then("the migration should be performed")
|
|
def step_check_migration_performed(context: Context) -> None:
|
|
"""Check that migration was performed."""
|
|
cleveragents_dir = context.project_path / ".cleveragents"
|
|
# Check that at least one backup file exists
|
|
backups_exist = (
|
|
(cleveragents_dir / "plans.json.backup").exists()
|
|
or (cleveragents_dir / "contexts.json.backup").exists()
|
|
or (cleveragents_dir / "changes.json.backup").exists()
|
|
)
|
|
assert backups_exist
|
|
|
|
|
|
@then("the migration should handle errors gracefully")
|
|
def step_check_error_handling(context: Context) -> None:
|
|
"""Check that migration handled errors gracefully."""
|
|
# If we got here without raising an exception, it handled errors
|
|
assert True
|
|
|
|
|
|
@then("the migration should handle backup errors gracefully")
|
|
def step_check_backup_error_handling(context: Context) -> None:
|
|
"""Check that migration handled backup errors gracefully."""
|
|
# The migration may fail or succeed, but shouldn't crash
|
|
assert context.migration_result in [True, False]
|
|
|
|
|
|
@given("I have a legacy project with multiple plans to migrate efficiently")
|
|
def step_create_project_multiple_plans_efficiency(context: Context) -> None:
|
|
"""Create a legacy project with multiple plans for efficiency testing."""
|
|
project_dir = context.temp_dir / "efficiency_test"
|
|
project_dir.mkdir(exist_ok=True)
|
|
|
|
cleveragents_dir = project_dir / ".cleveragents"
|
|
cleveragents_dir.mkdir(exist_ok=True)
|
|
|
|
(cleveragents_dir / "project.name").write_text("efficiency_test")
|
|
|
|
# Create multiple plans to ensure the loop runs more than once
|
|
plans_data = {
|
|
"plan_alpha": {"prompt": "Alpha plan", "status": "pending", "current": False},
|
|
"plan_beta": {"prompt": "Beta plan", "status": "built", "current": False},
|
|
"plan_gamma": {"prompt": "Gamma plan", "status": "applied", "current": True},
|
|
"plan_delta": {"prompt": "Delta plan", "status": "pending", "current": False},
|
|
}
|
|
(cleveragents_dir / "plans.json").write_text(json.dumps(plans_data))
|
|
(cleveragents_dir / "current").write_text("plan_gamma")
|
|
|
|
context.project_path = project_dir
|
|
|
|
|
|
@when("I run the legacy data migration with query tracking")
|
|
def step_run_migration_with_tracking(context: Context) -> None:
|
|
"""Run migration while tracking how many times get_all_for_project is called."""
|
|
# Find the plans repository class to patch at the class level
|
|
with context.unit_of_work.transaction() as ctx:
|
|
repo_class = type(ctx.plans)
|
|
|
|
call_tracker = {"count": 0}
|
|
original_repo_method = repo_class.get_all_for_project
|
|
|
|
def tracking_method(self, project_id):
|
|
call_tracker["count"] += 1
|
|
return original_repo_method(self, project_id)
|
|
|
|
repo_class.get_all_for_project = tracking_method
|
|
try:
|
|
migrator = LegacyDataMigrator(context.unit_of_work)
|
|
context.migration_result = migrator.migrate_project_data(context.project_path)
|
|
finally:
|
|
repo_class.get_all_for_project = original_repo_method
|
|
|
|
context.get_all_for_project_call_count = call_tracker["count"]
|
|
|
|
|
|
@then("get_all_for_project should be called exactly once")
|
|
def step_check_get_all_called_once(context: Context) -> None:
|
|
"""Verify get_all_for_project was called exactly once (not once per plan)."""
|
|
call_count = context.get_all_for_project_call_count
|
|
assert call_count == 1, (
|
|
f"Expected get_all_for_project to be called exactly once, "
|
|
f"but it was called {call_count} times. "
|
|
f"This indicates an N+1 query pattern is still present."
|
|
)
|
|
|
|
|
|
@then("all plans should be migrated correctly")
|
|
def step_check_all_plans_migrated(context: Context) -> None:
|
|
"""Verify all plans from the efficiency test were migrated."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("efficiency_test")
|
|
assert project is not None, "Project should have been created"
|
|
plans = ctx.plans.get_all_for_project(project.id)
|
|
plan_names = {p.name for p in plans}
|
|
expected_names = {"plan_alpha", "plan_beta", "plan_gamma", "plan_delta"}
|
|
assert plan_names == expected_names, (
|
|
f"Expected plans {expected_names}, got {plan_names}"
|
|
)
|
|
|
|
|
|
@then("the existing plan id should be mapped without type suppression")
|
|
def step_check_existing_plan_id_mapped(context: Context) -> None:
|
|
"""Check that existing plan id was mapped correctly via assert type narrowing."""
|
|
with context.unit_of_work.transaction() as ctx:
|
|
project = ctx.projects.get_by_name("existing_plans")
|
|
assert project is not None, "Project 'existing_plans' should exist in database"
|
|
plans = ctx.plans.get_all_for_project(project.id)
|
|
main_plans = [p for p in plans if p.name == "main"]
|
|
assert len(main_plans) == 1, "Should have exactly one 'main' plan"
|
|
assert main_plans[0].id is not None, (
|
|
"Existing plan id must be non-None (assert narrowing validates this)"
|
|
)
|