fix(lint): address ruff lint failures in parallel_subplan_scheduler_steps.py

features/steps/parallel_subplan_scheduler_steps.py

@@ -1,29 +1,15 @@
 """Step definitions for parallel subplan scheduler BDD tests."""
-
 from __future__ import annotations
 
-import time
 from datetime import UTC, datetime
+import time
 from typing import Any
-from unittest.mock import MagicMock
 
 from behave import given, then, when
 
-from cleveragents.application.services.parallel_subplan_scheduler import (
+from cleveragents.application.services.parallel_subplan_scheduler import ParallelSubplanScheduler, SchedulerState, SubplanQueue
-    ParallelSubplanScheduler,
+from cleveragents.application.services.subplan_execution_service import SubplanExecutionOutput
-    SchedulerState,
+from cleveragents.domain.models.core.plan import ExecutionMode, ProcessingState, SubplanConfig, SubplanMergeStrategy, SubplanStatus
-    SubplanQueue,
-)
-from cleveragents.application.services.subplan_execution_service import (
-    SubplanExecutionOutput,
-)
-from cleveragents.domain.models.core.plan import (
-    ExecutionMode,
-    ProcessingState,
-    SubplanConfig,
-    SubplanMergeStrategy,
-    SubplanStatus,
-)
 
 
 # --- Fixtures and helpers ---
@@ -53,7 +39,7 @@ def create_executor_fn(
 
     def executor(status: SubplanStatus) -> SubplanExecutionOutput:
         subplan_id = status.subplan_id
-        
+
         # Track execution for concurrency testing
         if not hasattr(context, "concurrent_executions"):
             context.concurrent_executions = []
@@ -61,16 +47,16 @@ def create_executor_fn(
             context.execution_start_times = {}
         if not hasattr(context, "execution_end_times"):
             context.execution_end_times = {}
-        
+
         context.execution_start_times[subplan_id] = time.time()
         context.concurrent_executions.append(subplan_id)
-        
+
         # Simulate blocking if configured
         if subplan_id in block_seconds:
             time.sleep(block_seconds[subplan_id])
-        
+
         context.execution_end_times[subplan_id] = time.time()
-        
+
         # Simulate failure if configured
         if subplan_id in fail_ids:
             return SubplanExecutionOutput(
@@ -78,7 +64,7 @@ def create_executor_fn(
                 success=False,
                 error="Test failure",
             )
-        
+
         return SubplanExecutionOutput(
             subplan_id=subplan_id,
             success=True,
@@ -86,7 +72,7 @@ def create_executor_fn(
             files_changed=1,
             changeset_summary="Test changes",
         )
-    
+
     return executor
 
 
@@ -252,7 +238,7 @@ def step_create_subplans_with_failure(context: Any, count: int) -> None:
         create_subplan_status(f"subplan-{i:03d}")
         for i in range(count)
     ]
-    fail_ids = {f"subplan-001"}
+    fail_ids = {"subplan-001"}
     context.scheduler._executor_fn = create_executor_fn(context, fail_ids=fail_ids)
 
 
@@ -263,7 +249,7 @@ def step_create_subplans_with_first_failure(context: Any, count: int) -> None:
         create_subplan_status(f"subplan-{i:03d}")
         for i in range(count)
     ]
-    fail_ids = {f"subplan-000"}
+    fail_ids = {"subplan-000"}
     context.scheduler._executor_fn = create_executor_fn(context, fail_ids=fail_ids)
 
 
@@ -274,7 +260,7 @@ def step_create_subplans_with_blocking(context: Any, count: int, seconds: int) -
         create_subplan_status(f"subplan-{i:03d}")
         for i in range(count)
     ]
-    block_seconds = {f"subplan-000": seconds}
+    block_seconds = {"subplan-000": seconds}
     context.scheduler._executor_fn = create_executor_fn(context, block_seconds=block_seconds)
 
 
@@ -287,7 +273,7 @@ def step_create_subplans_with_blocking_and_quick(
         create_subplan_status(f"subplan-{i:03d}")
         for i in range(count)
     ]
-    block_seconds = {f"subplan-000": seconds}
+    block_seconds = {"subplan-000": seconds}
     context.scheduler._executor_fn = create_executor_fn(context, block_seconds=block_seconds)
 
 
@@ -495,25 +481,25 @@ def step_check_peak_concurrency(context: Any, max_parallel: int) -> None:
     """Verify peak concurrency doesn't exceed limit."""
     if not hasattr(context, "execution_start_times"):
         return  # Skip if not tracking
-    
+
     # Calculate peak concurrency
     events = []
     for subplan_id, start_time in context.execution_start_times.items():
         end_time = context.execution_end_times.get(subplan_id, start_time)
         events.append((start_time, "start"))
         events.append((end_time, "end"))
-    
+
     events.sort()
     current_concurrent = 0
     peak_concurrent = 0
-    
+
     for _, event_type in events:
         if event_type == "start":
             current_concurrent += 1
             peak_concurrent = max(peak_concurrent, current_concurrent)
         else:
             current_concurrent -= 1
-    
+
     assert peak_concurrent <= max_parallel
 
 
@@ -522,7 +508,7 @@ def step_check_execution_order(context: Any) -> None:
     """Verify subplans executed in order."""
     if not hasattr(context, "execution_start_times"):
         return  # Skip if not tracking
-    
+
     subplan_ids = list(context.execution_start_times.keys())
     for i in range(len(subplan_ids) - 1):
         start_i = context.execution_start_times[subplan_ids[i]]
@@ -821,7 +807,7 @@ def step_verify_timeout_error(context: Any) -> None:
         if status.status == ProcessingState.ERRORED:
             assert "timeout" in (status.error or "").lower()
             return
-    assert False, "No timeout error found"
+    raise AssertionError("No timeout error found")
 
 
 @then("the second subplan should complete successfully")
@@ -858,11 +844,11 @@ def step_verify_dependency_order(context: Any) -> None:
     """Verify dependency order is respected."""
     assert context.result is not None
     status_map = {s.subplan_id: s for s in context.result.statuses}
-    
+
     # Verify A before B and C
     assert status_map["subplan-A"].completed_at < status_map["subplan-B"].completed_at
     assert status_map["subplan-A"].completed_at < status_map["subplan-C"].completed_at
-    
+
     # Verify B and C before D
     assert status_map["subplan-B"].completed_at < status_map["subplan-D"].completed_at
     assert status_map["subplan-C"].completed_at < status_map["subplan-D"].completed_at
@@ -873,25 +859,25 @@ def step_verify_min_concurrency(context: Any, min_concurrent: int) -> None:
     """Verify minimum concurrency."""
     if not hasattr(context, "execution_start_times"):
         return  # Skip if not tracking
-    
+
     # Calculate peak concurrency
     events = []
     for subplan_id, start_time in context.execution_start_times.items():
         end_time = context.execution_end_times.get(subplan_id, start_time)
         events.append((start_time, "start"))
         events.append((end_time, "end"))
-    
+
     events.sort()
     current_concurrent = 0
     peak_concurrent = 0
-    
+
     for _, event_type in events:
         if event_type == "start":
             current_concurrent += 1
             peak_concurrent = max(peak_concurrent, current_concurrent)
         else:
             current_concurrent -= 1
-    
+
     assert peak_concurrent >= min_concurrent
 
 
@@ -900,7 +886,7 @@ def step_verify_c_after_ab(context: Any) -> None:
     """Verify C starts after A and B complete."""
     assert context.result is not None
     status_map = {s.subplan_id: s for s in context.result.statuses}
-    
+
     # C should start after both A and B complete
     assert status_map["subplan-C"].started_at > status_map["subplan-A"].completed_at
     assert status_map["subplan-C"].started_at > status_map["subplan-B"].completed_at