cleveragents-core/features/steps/coverage_easy_gains_steps.py

"""Step definitions for ``coverage_easy_gains.feature``.

Targets three distinct uncovered-line areas:
  - plan.py 876:           Plan validate_phase_state_consistency defaults QUEUED
  - config_parser.py 434:  _build_from_v3 propagates options to graph nodes
  - resource_file_watcher.py 378:  _on_file_changed early-return on unknown event
"""

from __future__ import annotations

import tempfile
from pathlib import Path
from unittest.mock import patch

from behave import given, then, when
from behave.runner import Context

# ------------------------------------------------------------------
# Plan model coverage (plan.py 876)
# ------------------------------------------------------------------


@given("a Plan model factory is available")
def step_eg_given_plan_factory(context: Context) -> None:
    from cleveragents.domain.models.core.plan import ProcessingState, ProjectLink

    context.eg_processing_state_queued = ProcessingState.QUEUED
    context.eg_project_link = ProjectLink(
        project_name="local/demo-action",
        alias=None,
    )


@when(
    "I create an action-phase plan without specifying processing_state",
    target="plan.py:876",
)
def step_eg_create_action_plan_no_state(context: Context) -> None:
    from cleveragents.domain.models.core.plan import (
        NamespacedName,
        Plan,
        PlanIdentity,
        PlanPhase,
    )

    context.eg_plan = Plan(
        identity=PlanIdentity(plan_id="01HGZ6FE0AQDYTR4BXVQZ6EA02"),
        namespaced_name=NamespacedName(namespace="local", name="easy-gains"),
        action_name="local/demo-action",
        description="Plan for easy-gains coverage",
        phase=PlanPhase.ACTION,
    )


@when(
    "I create an action-phase plan with pre-set QUEUED state",
    target="plan.py:876",
)
def step_eg_create_action_plan_with_queued(context: Context) -> None:
    from cleveragents.domain.models.core.plan import (
        NamespacedName,
        Plan,
        PlanIdentity,
        PlanPhase,
        ProcessingState,
    )

    context.eg_plan = Plan(
        identity=PlanIdentity(plan_id="01HGZ6FE0AQDYTR4BXVQZ6EA03"),
        namespaced_name=NamespacedName(namespace="local", name="easy-gains-2"),
        action_name="local/demo-action",
        description="Plan with explicit QUEUED state",
        phase=PlanPhase.ACTION,
        processing_state=ProcessingState.QUEUED,
    )


@then("the plan should have processing_state == ProcessingState.QUEUED")
def step_eg_plan_has_queued(context: Context) -> None:
    from cleveragents.domain.models.core.plan import ProcessingState

    assert context.eg_plan.processing_state is ProcessingState.QUEUED, (
        f"Expected QUEUED, got {context.eg_plan.processing_state}"
    )


@then("its phase should still be PlanPhase.ACTION")
def step_eg_phase_is_action(context: Context) -> None:
    from cleveragents.domain.models.core.plan import PlanPhase

    assert context.eg_plan.phase is PlanPhase.ACTION, (
        f"Expected ACTION, got {context.eg_plan.phase}"
    )


# ------------------------------------------------------------------
# Config parser coverage (config_parser.py 434)
# ------------------------------------------------------------------


@given("a ReactiveConfig builder is available")
def step_eg_given_config_builder(context: Context) -> None:
    from cleveragents.reactive import config_parser

    context.eg_build_from_v3 = config_parser.ReactiveConfigParser._build_from_v3


@when('I build a graph actor config with actor-level "options" block', target="434")
def step_eg_build_graph_with_options(context: Context) -> None:
    from cleveragents.reactive.config_parser import ReactiveConfig

    rc = ReactiveConfig()

    raw_data = {
        "type": "graph",
        "name": "graph-with-options",
        "model": "gpt-4o",
        "route": {
            "nodes": [
                {"id": "start", "config": {"system_prompt": "Begin"}},
                {"id": "middle", "config": {"system_prompt": "Middle step"}},
                {"id": "end", "config": {"system_prompt": "Done"}},
            ],
            "edges": [
                {"from": "start", "to": "middle"},
                {"from": "middle", "to": "end"},
            ],
            "entry_node": "start",
        },
        "options": {
            "openai_api_base": "https://proxy.example.com/v1",
            "max_retries": 5,
            "timeout_seconds": 30.0,
        },
    }

    context.eg_config = context.eg_build_from_v3(raw_data, rc)


@then("the resulting nodes should each receive the same options dict")
def step_eg_graph_nodes_receive_options(context: Context) -> None:
    expected_options = {
        "openai_api_base": "https://proxy.example.com/v1",
        "max_retries": 5,
        "timeout_seconds": 30.0,
    }

    for node_id, agent in context.eg_config.agents.items():
        config_dict = agent.config or {}
        assert "options" in config_dict, (
            f"Agent '{node_id}' missing 'options' in config"
        )
        actual_options = config_dict["options"]
        for key, val in expected_options.items():
            assert actual_options[key] == val, (
                f"Node '{node_id}': expected options['{key}']={val}, "
                f"got {actual_options.get(key)!r}"
            )


# ------------------------------------------------------------------
# Resource file watcher coverage (resource_file_watcher.py 378)
# ------------------------------------------------------------------


@given("a ResourceFileWatcher with a watched path is available")
def step_eg_given_watcher_with_path(context: Context) -> None:
    from cleveragents.application.services.resource_file_watcher import (
        ResourceFileWatcher,
    )

    tmpdir_obj = tempfile.TemporaryDirectory()
    temp_dir = Path(tmpdir_obj.name)
    watch_file = temp_dir / "watched_target.txt"
    watch_file.write_text("hello world")

    watcher = ResourceFileWatcher()
    watcher._running = True
    # Simulate a watched file by seeding the internal path map directly.
    watcher._watched_paths[str(watch_file.resolve())] = (
        "test-resource-id-001",
        "test-project",
    )

    context.eg_watcher_tmpdir = tmpdir_obj
    context.eg_watcher = watcher
    context.eg_watch_file = watch_file


@when(
    "I trigger an unrecognized file-system event on the watched path",
    target="resource_file_watcher.py:378",
)
def step_eg_trigger_unrecognized_event(context: Context) -> None:
    from cleveragents.application.services.resource_file_watcher import (
        ResourceFileWatcher,
    )

    # Send a custom generic event that is NOT FileCreatedEvent / FileModifiedEvent
    # / FileDeletedEvent / FileMovedEvent, so line 378's ``else: return`` fires.
    from watchdog.events import FileSystemEvent

    class _UnknownEvent(FileSystemEvent):
        """Artificially craft an unrecognized event type."""

        def __init__(self, src_path: str) -> None:
            super().__init__(src_path=src_path)
            self.is_directory = False

    unknown_event = _UnknownEvent(str(context.eg_watch_file))

    # Capture whether the handler actually tried to do any work.
    with patch.object(
        ResourceFileWatcher,
        "_fire_change",
        spec_set=True,
    ) as mock_fire:
        context.eg_watcher._handle_fs_event(unknown_event)
        context.eg_fire_called = mock_fire.called


@then("no resource change callback should fire")
def step_eg_no_callback_fired(context: Context) -> None:
    assert context.eg_fire_called is False, (
        "_fire_change was unexpectedly invoked for an unrecognized event"
    )


@then("the watcher should still be running")
def step_eg_watcher_still_running(context: Context) -> None:
    # The early-return path does not modify _running; assert it remains True.
    assert getattr(context.eg_watcher, "_running", False) is True, (
        "Watcher should still be running after unrecognized event"
    )