cleveragents-core/features/steps/validation_pipeline_concurrency_steps.py

"""Step definitions for concurrent validation pipeline scenarios.

Tests that concurrent ValidationPipeline.run() calls correctly restore
sys.stdout and sys.stderr to the true originals after all pipelines finish,
and that each pipeline captures its own per-thread output independently.

The concurrency test avoids nested threading (threads-within-threads) by
using a simple sequential approach: install the stream wrappers twice
(simulating two concurrent pipelines), verify the reference count is 2,
then release twice and verify restoration.
"""

from __future__ import annotations

import sys
import threading
from typing import Any

from behave import given, then, use_step_matcher, when

from cleveragents.application.services.validation_pipeline import (
    ValidationCommand,
    ValidationPipeline,
    _ThreadLocalStream,
)
from cleveragents.domain.models.core.tool import ValidationMode

use_step_matcher("re")


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


class _FastPrintingExecutor:
    """Executor that prints a unique message and returns passed immediately."""

    def __init__(self, pipeline_id: str) -> None:
        self._pipeline_id = pipeline_id

    def __call__(
        self, validation_name: str, arguments: dict[str, Any]
    ) -> dict[str, Any]:
        print(f"output-from-{self._pipeline_id}")
        return {"passed": True, "message": f"{validation_name} ok"}


# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------


@given(r"two concurrent validation pipelines are prepared")
def step_given_two_concurrent_pipelines(context: Any) -> None:
    context.vp_concurrent_summaries: list[Any] = [None, None]
    context.vp_concurrent_errors: list[Exception | None] = [None, None]
    context.vp_stdout_before: Any = sys.stdout
    context.vp_stderr_before: Any = sys.stderr

    def _make_command(name: str, resource: str) -> ValidationCommand:
        return ValidationCommand(
            validation_name=name,
            resource_id="R00001",
            resource_name=resource,
            mode=ValidationMode.REQUIRED,
            arguments={},
        )

    context.vp_pipeline_a = ValidationPipeline(
        commands=[_make_command("check-a", "resource-a")],
        executor=_FastPrintingExecutor("pipeline-a"),
        max_workers=1,
    )
    context.vp_pipeline_b = ValidationPipeline(
        commands=[_make_command("check-b", "resource-b")],
        executor=_FastPrintingExecutor("pipeline-b"),
        max_workers=1,
    )


# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------


@when(r"both pipelines run concurrently using a barrier")
def step_when_both_pipelines_run_concurrently(context: Any) -> None:
    """Run both pipelines in separate threads and collect results."""
    summaries: list[Any] = [None, None]
    errors: list[Exception | None] = [None, None]

    def _run(idx: int, pipeline: ValidationPipeline) -> None:
        try:
            summaries[idx] = pipeline.run()
        except Exception as exc:
            errors[idx] = exc

    t_a = threading.Thread(target=_run, args=(0, context.vp_pipeline_a))
    t_b = threading.Thread(target=_run, args=(1, context.vp_pipeline_b))
    t_a.start()
    t_b.start()
    t_a.join(timeout=30.0)
    t_b.join(timeout=30.0)

    context.vp_concurrent_summaries = summaries
    context.vp_concurrent_errors = errors


# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------


@then(r"sys\.stdout is not a _ThreadLocalStream after both pipelines finish")
def step_then_stdout_restored(context: Any) -> None:
    assert not isinstance(sys.stdout, _ThreadLocalStream), (
        f"sys.stdout was not restored: {type(sys.stdout)}"
    )


@then(r"sys\.stderr is not a _ThreadLocalStream after both pipelines finish")
def step_then_stderr_restored(context: Any) -> None:
    assert not isinstance(sys.stderr, _ThreadLocalStream), (
        f"sys.stderr was not restored: {type(sys.stderr)}"
    )


@then(r"each pipeline captured its own stdout output independently")
def step_then_each_pipeline_captured_independently(context: Any) -> None:
    summaries = context.vp_concurrent_summaries
    errors = context.vp_concurrent_errors

    for i, err in enumerate(errors):
        assert err is None, f"Pipeline {i} raised an error: {err}"

    for i, summary in enumerate(summaries):
        assert summary is not None, f"Pipeline {i} produced no summary"
        assert summary.total == 1, (
            f"Pipeline {i} expected 1 result, got {summary.total}"
        )
        assert summary.all_required_passed, f"Pipeline {i} had required failures"

    # Verify captured stdout is isolated per pipeline
    result_a = summaries[0].results[0]
    result_b = summaries[1].results[0]

    data_a = result_a.data or {}
    data_b = result_b.data or {}

    captured_a = data_a.get("_captured_stdout", "")
    captured_b = data_b.get("_captured_stdout", "")

    assert "pipeline-a" in captured_a, (
        f"Pipeline A did not capture its own output. Got: {captured_a!r}"
    )
    assert "pipeline-b" not in captured_a, (
        f"Pipeline A captured pipeline B's output: {captured_a!r}"
    )
    assert "pipeline-b" in captured_b, (
        f"Pipeline B did not capture its own output. Got: {captured_b!r}"
    )
    assert "pipeline-a" not in captured_b, (
        f"Pipeline B captured pipeline A's output: {captured_b!r}"
    )


# Reset to default parse matcher
use_step_matcher("parse")