cleveragents-core/robot/helper_m4_e2e_common.py

"""Shared constants, utilities, and mock factories for M4 E2E verification tests.

Provides reusable test infrastructure for both domain-model tests
(``helper_m4_e2e_domain``), merge tests (``helper_m4_e2e_merge``),
and CLI integration tests (``helper_m4_e2e_cli``, ``helper_m4_e2e_cli_errors``).
"""

from __future__ import annotations

import sys
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, NoReturn
from unittest.mock import MagicMock

# Ensure the src directory is on the import path.  Uses ``append`` rather
# than ``insert(0, ...)`` to avoid shadowing stdlib modules if a file
# in ``src/`` were ever named identically to a stdlib module.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.append(_SRC)

from cleveragents.domain.models.core.plan import (  # noqa: E402
    ExecutionMode,
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
    ProjectLink,
    SubplanAttempt,
    SubplanConfig,
    SubplanMergeStrategy,
    SubplanStatus,
)

if TYPE_CHECKING:
    # Typer's CliRunner wraps Click's Result; importing via click.testing
    # is the canonical path since typer.testing re-exports CliRunner but
    # does not re-export Result.
    from click.testing import Result as CliResult

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

_ROOT_ULID = "01KHDE6WWS2171PWW3GJEBXZ8R"
_CHILD_A_ULID = "01KHDE6WWS2171PWW3GJEBXZ8A"
_CHILD_B_ULID = "01KHDE6WWS2171PWW3GJEBXZ8B"
_CHILD_C_ULID = "01KHDE6WWS2171PWW3GJEBXZ8C"

# Frozen timestamp for deterministic fixtures.  Avoids non-deterministic
# ``datetime.now()`` calls that produce subtly inconsistent timestamps
# across fixture objects within a single test.
_FROZEN_NOW = datetime(2026, 3, 1, 12, 0, 0)

# ---------------------------------------------------------------------------
# Shared test helpers
# ---------------------------------------------------------------------------


def _fail(msg: str) -> NoReturn:
    """Print failure message to stderr and exit 1."""
    print(f"FAIL: {msg}", file=sys.stderr)
    raise SystemExit(1)


def _assert_mock_called_once(mock: MagicMock, label: str) -> None:
    """Wrap ``assert_called_once`` with standardised ``_fail()`` diagnostics.

    Raw ``AssertionError`` from ``unittest.mock`` bypasses the ``FAIL:``
    prefix pattern used by Robot Framework test evaluation; this wrapper
    converts it.
    """
    try:
        mock.assert_called_once()
    except AssertionError as exc:
        _fail(f"{label}: {exc}")


def _assert_mock_called_once_with(
    mock: MagicMock,
    label: str,
    *args: object,
    **kwargs: object,
) -> None:
    """Wrap ``assert_called_once_with`` with ``_fail()`` diagnostics."""
    try:
        mock.assert_called_once_with(*args, **kwargs)
    except AssertionError as exc:
        _fail(f"{label}: {exc}")


def _assert_exit_code(result: CliResult, label: str) -> None:
    """Verify CLI exit code is 0, failing with full diagnostic output."""
    if result.exit_code != 0:
        _fail(
            f"{label} rc={result.exit_code}\n"
            f"output={result.output}\n"
            f"exception={result.exception}"
        )


# ---------------------------------------------------------------------------
# SubplanStatus factory (DRY extraction)
# ---------------------------------------------------------------------------


def _make_subplan_status(
    subplan_id: str,
    *,
    action_name: str = "local/m4-verify-action",
    target_resources: list[str] | None = None,
    status: ProcessingState = ProcessingState.QUEUED,
    started_at: datetime | None = None,
    completed_at: datetime | None = None,
    changeset_summary: str | None = None,
    files_changed: int = 0,
    error: str | None = None,
    attempt_number: int = 1,
    previous_attempts: list[SubplanAttempt] | None = None,
) -> SubplanStatus:
    """Create a ``SubplanStatus`` with sensible test defaults."""
    return SubplanStatus(
        subplan_id=subplan_id,
        action_name=action_name,
        target_resources=target_resources or [],
        status=status,
        started_at=started_at,
        completed_at=completed_at,
        changeset_summary=changeset_summary,
        files_changed=files_changed,
        error=error,
        attempt_number=attempt_number,
        previous_attempts=previous_attempts or [],
    )


def _make_subplan_statuses() -> list[SubplanStatus]:
    """Create three SubplanStatus entries (A=COMPLETE, B=COMPLETE, C=QUEUED).

    This is the default three-child status set used by many M4 tests.
    """
    return [
        _make_subplan_status(
            _CHILD_A_ULID,
            target_resources=["res-api"],
            status=ProcessingState.COMPLETE,
            started_at=_FROZEN_NOW,
            completed_at=_FROZEN_NOW,
            changeset_summary="Updated API endpoints",
            files_changed=3,
        ),
        _make_subplan_status(
            _CHILD_B_ULID,
            target_resources=["res-ui"],
            status=ProcessingState.COMPLETE,
            started_at=_FROZEN_NOW,
            completed_at=_FROZEN_NOW,
            changeset_summary="Updated UI components",
            files_changed=5,
        ),
        _make_subplan_status(
            _CHILD_C_ULID,
            target_resources=["res-db"],
        ),
    ]


# ---------------------------------------------------------------------------
# Plan mock factories
# ---------------------------------------------------------------------------


def _mock_parent_plan(
    phase: PlanPhase = PlanPhase.EXECUTE,
    state: ProcessingState = ProcessingState.PROCESSING,
    subplan_statuses: list[SubplanStatus] | None = None,
    subplan_config: SubplanConfig | None = None,
    *,
    read_only: bool = False,
) -> Plan:
    """Create a parent plan with subplan config."""
    return Plan(
        identity=PlanIdentity(plan_id=_ROOT_ULID),
        namespaced_name=NamespacedName.parse("local/m4-parent-plan"),
        description="M4 verification parent plan with subplans",
        definition_of_done="All subplans complete successfully",
        action_name="local/m4-verify-action",
        phase=phase,
        processing_state=state,
        project_links=[ProjectLink(project_name="local/m4-project")],
        arguments={},
        arguments_order=[],
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        timestamps=PlanTimestamps(created_at=_FROZEN_NOW, updated_at=_FROZEN_NOW),
        created_by=None,
        reusable=True,
        read_only=read_only,
        subplan_config=subplan_config
        or SubplanConfig(
            execution_mode=ExecutionMode.PARALLEL,
            merge_strategy=SubplanMergeStrategy.GIT_THREE_WAY,
            max_parallel=3,
            fail_fast=False,
            retry_failed=True,
            max_retries=2,
        ),
        subplan_statuses=subplan_statuses or [],
    )


def _mock_child_plan(
    child_id: str,
    parent_id: str = _ROOT_ULID,
    root_id: str = _ROOT_ULID,
    phase: PlanPhase = PlanPhase.EXECUTE,
    state: ProcessingState = ProcessingState.QUEUED,
) -> Plan:
    """Create a child subplan."""
    return Plan(
        identity=PlanIdentity(
            plan_id=child_id,
            parent_plan_id=parent_id,
            root_plan_id=root_id,
        ),
        namespaced_name=NamespacedName.parse(f"local/m4-child-{child_id[-1].lower()}"),
        description=f"M4 child subplan {child_id[-1]}",
        definition_of_done="Subplan criteria pass",
        action_name="local/m4-verify-action",
        phase=phase,
        processing_state=state,
        project_links=[ProjectLink(project_name="local/m4-project")],
        arguments={},
        arguments_order=[],
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        timestamps=PlanTimestamps(created_at=_FROZEN_NOW, updated_at=_FROZEN_NOW),
        created_by=None,
        reusable=True,
        read_only=False,
    )