fix(lint): address ruff lint failures in parallel_subplan_scheduler_steps.py
CI / push-validation (pull_request) Successful in 32s
CI / helm (pull_request) Successful in 41s
CI / build (pull_request) Successful in 56s
CI / lint (pull_request) Failing after 1m9s
CI / quality (pull_request) Successful in 1m29s
CI / typecheck (pull_request) Successful in 1m34s
CI / security (pull_request) Successful in 2m2s
CI / coverage (pull_request) Has been skipped
CI / unit_tests (pull_request) Failing after 3m11s
CI / docker (pull_request) Has been skipped
CI / e2e_tests (pull_request) Successful in 3m41s
CI / integration_tests (pull_request) Successful in 4m36s
CI / status-check (pull_request) Failing after 3s
CI / push-validation (pull_request) Successful in 32s
CI / helm (pull_request) Successful in 41s
CI / build (pull_request) Successful in 56s
CI / lint (pull_request) Failing after 1m9s
CI / quality (pull_request) Successful in 1m29s
CI / typecheck (pull_request) Successful in 1m34s
CI / security (pull_request) Successful in 2m2s
CI / coverage (pull_request) Has been skipped
CI / unit_tests (pull_request) Failing after 3m11s
CI / docker (pull_request) Has been skipped
CI / e2e_tests (pull_request) Successful in 3m41s
CI / integration_tests (pull_request) Successful in 4m36s
CI / status-check (pull_request) Failing after 3s
This commit is contained in:
@@ -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 (
|
||||
ParallelSubplanScheduler,
|
||||
SchedulerState,
|
||||
SubplanQueue,
|
||||
)
|
||||
from cleveragents.application.services.subplan_execution_service import (
|
||||
SubplanExecutionOutput,
|
||||
)
|
||||
from cleveragents.domain.models.core.plan import (
|
||||
ExecutionMode,
|
||||
ProcessingState,
|
||||
SubplanConfig,
|
||||
SubplanMergeStrategy,
|
||||
SubplanStatus,
|
||||
)
|
||||
from cleveragents.application.services.parallel_subplan_scheduler import ParallelSubplanScheduler, SchedulerState, 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
|
||||
|
||||
Reference in New Issue
Block a user