forked from cleveragents/cleveragents-core
freemo
e7df15d453
Implement UI data-provider interface backed by local services. Add Textual-based TUI with plan list, plan detail, diff viewer, and validation summary panes. Add Web UI stub with local-only read-only routes. Configure auto-refresh and manual keybinds. ISSUES CLOSED: #341
354 lines
11 KiB
Python
354 lines
11 KiB
Python
"""Step definitions for the TUI/Web UI interface feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.ui.data_provider import (
|
|
DiffItem,
|
|
LocalUIDataProvider,
|
|
LogEntry,
|
|
PlanDetail,
|
|
PlanSummary,
|
|
SessionSummary,
|
|
ValidationResult,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures: mock plan objects
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_PLAN_ALPHA = MagicMock()
|
|
_PLAN_ALPHA.id = "plan-1"
|
|
_PLAN_ALPHA.name = "plan-alpha"
|
|
_PLAN_ALPHA.status = "draft"
|
|
_PLAN_ALPHA.created_at = "2025-01-01T00:00:00"
|
|
_PLAN_ALPHA.prompt = "Do something"
|
|
_PLAN_ALPHA.decisions = []
|
|
_PLAN_ALPHA.linked_resources = []
|
|
|
|
_PLAN_BETA = MagicMock()
|
|
_PLAN_BETA.id = "plan-2"
|
|
_PLAN_BETA.name = "plan-beta"
|
|
_PLAN_BETA.status = "built"
|
|
_PLAN_BETA.created_at = "2025-01-02T00:00:00"
|
|
_PLAN_BETA.prompt = "Do something else"
|
|
|
|
_DEC = MagicMock()
|
|
_DEC.id = "dec-1"
|
|
_DEC.type = "approval"
|
|
_DEC.status = "pending"
|
|
_PLAN_ALPHA_WITH_DEC = MagicMock()
|
|
_PLAN_ALPHA_WITH_DEC.id = "plan-1"
|
|
_PLAN_ALPHA_WITH_DEC.name = "plan-alpha"
|
|
_PLAN_ALPHA_WITH_DEC.status = "draft"
|
|
_PLAN_ALPHA_WITH_DEC.created_at = "2025-01-01T00:00:00"
|
|
_PLAN_ALPHA_WITH_DEC.prompt = "Do something"
|
|
_PLAN_ALPHA_WITH_DEC.decisions = [_DEC]
|
|
_PLAN_ALPHA_WITH_DEC.linked_resources = ["res-1"]
|
|
_PLAN_ALPHA_WITH_DEC.project_id = "proj-1"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mock data provider for web tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _MockProvider:
|
|
"""Trivial UIDataProvider for web route testing."""
|
|
|
|
def __init__(self, *, with_plans: bool = False) -> None:
|
|
self._with_plans = with_plans
|
|
|
|
def get_plans(self, project_id: str) -> list[PlanSummary]:
|
|
if not self._with_plans:
|
|
return []
|
|
return [
|
|
PlanSummary(
|
|
plan_id="plan-1",
|
|
name="plan-alpha",
|
|
status="draft",
|
|
created_at="2025-01-01T00:00:00",
|
|
project_id=project_id or "proj-1",
|
|
),
|
|
]
|
|
|
|
def get_plan_detail(self, plan_id: str) -> PlanDetail | None:
|
|
if not self._with_plans:
|
|
return None
|
|
if plan_id != "plan-1":
|
|
return None
|
|
return PlanDetail(
|
|
plan_id="plan-1",
|
|
name="plan-alpha",
|
|
status="draft",
|
|
created_at="2025-01-01T00:00:00",
|
|
project_id="proj-1",
|
|
prompt="Do something",
|
|
decisions=[{"id": "d1", "type": "approval", "status": "pending"}],
|
|
linked_resources=["res-1"],
|
|
)
|
|
|
|
def get_sessions(self) -> list[SessionSummary]:
|
|
return []
|
|
|
|
def get_validations(self, plan_id: str) -> list[ValidationResult]:
|
|
return []
|
|
|
|
def get_diffs(self, plan_id: str) -> list[DiffItem]:
|
|
return []
|
|
|
|
def get_logs(self, plan_id: str, level: str = "INFO") -> list[LogEntry]:
|
|
return []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# UIDataProvider steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a UIDataProvider backed by mock services")
|
|
def step_provider_empty(context: Context) -> None:
|
|
mock_svc = MagicMock()
|
|
mock_svc.list_plans.return_value = []
|
|
mock_svc.get_plan.return_value = None
|
|
context.ui_provider = LocalUIDataProvider(plan_service=mock_svc)
|
|
|
|
|
|
@given("a UIDataProvider backed by mock services with plans")
|
|
def step_provider_with_plans(context: Context) -> None:
|
|
mock_svc = MagicMock()
|
|
mock_svc.list_plans.return_value = [_PLAN_ALPHA, _PLAN_BETA]
|
|
mock_svc.get_plan.side_effect = lambda pid: (
|
|
_PLAN_ALPHA_WITH_DEC if pid == "plan-1" else None
|
|
)
|
|
context.ui_provider = LocalUIDataProvider(plan_service=mock_svc)
|
|
|
|
|
|
@when('I request UI plans for project "{project_id}"')
|
|
def step_request_ui_plans(context: Context, project_id: str) -> None:
|
|
context.ui_result_plans = context.ui_provider.get_plans(project_id)
|
|
|
|
|
|
@then("the UI plan list should be empty")
|
|
def step_ui_plans_empty(context: Context) -> None:
|
|
assert context.ui_result_plans == [], (
|
|
f"Expected empty, got {context.ui_result_plans}"
|
|
)
|
|
|
|
|
|
@then("the UI plan list should contain {count:d} plans")
|
|
def step_ui_plans_count(context: Context, count: int) -> None:
|
|
assert len(context.ui_result_plans) == count, (
|
|
f"Expected {count}, got {len(context.ui_result_plans)}"
|
|
)
|
|
|
|
|
|
@then('the first UI plan name should be "{name}"')
|
|
def step_first_ui_plan_name(context: Context, name: str) -> None:
|
|
assert context.ui_result_plans[0].name == name
|
|
|
|
|
|
@when('I request UI plan detail for "{plan_id}"')
|
|
def step_request_ui_detail(context: Context, plan_id: str) -> None:
|
|
context.ui_result_detail = context.ui_provider.get_plan_detail(plan_id)
|
|
|
|
|
|
@then('the UI plan detail should have name "{name}"')
|
|
def step_ui_detail_name(context: Context, name: str) -> None:
|
|
assert context.ui_result_detail is not None
|
|
assert context.ui_result_detail.name == name
|
|
|
|
|
|
@then("the UI plan detail should include decisions")
|
|
def step_ui_detail_decisions(context: Context) -> None:
|
|
assert context.ui_result_detail is not None
|
|
assert isinstance(context.ui_result_detail.decisions, list)
|
|
|
|
|
|
@then("the UI plan detail should be None")
|
|
def step_ui_detail_none(context: Context) -> None:
|
|
assert context.ui_result_detail is None
|
|
|
|
|
|
@when('I request UI validations for plan "{plan_id}"')
|
|
def step_request_ui_validations(context: Context, plan_id: str) -> None:
|
|
context.ui_result_validations = context.ui_provider.get_validations(plan_id)
|
|
|
|
|
|
@then("the UI validations list should be empty")
|
|
def step_ui_validations_empty(context: Context) -> None:
|
|
assert context.ui_result_validations == []
|
|
|
|
|
|
@when('I request UI diffs for plan "{plan_id}"')
|
|
def step_request_ui_diffs(context: Context, plan_id: str) -> None:
|
|
context.ui_result_diffs = context.ui_provider.get_diffs(plan_id)
|
|
|
|
|
|
@then("the UI diffs list should be empty")
|
|
def step_ui_diffs_empty(context: Context) -> None:
|
|
assert context.ui_result_diffs == []
|
|
|
|
|
|
@when('I request UI logs for plan "{plan_id}" with level "{level}"')
|
|
def step_request_ui_logs(context: Context, plan_id: str, level: str) -> None:
|
|
context.ui_result_logs = context.ui_provider.get_logs(plan_id, level)
|
|
|
|
|
|
@then("the UI logs list should be empty")
|
|
def step_ui_logs_empty(context: Context) -> None:
|
|
assert context.ui_result_logs == []
|
|
|
|
|
|
@when("I request UI sessions")
|
|
def step_request_ui_sessions(context: Context) -> None:
|
|
context.ui_result_sessions = context.ui_provider.get_sessions()
|
|
|
|
|
|
@then("the UI sessions list should be empty")
|
|
def step_ui_sessions_empty(context: Context) -> None:
|
|
assert context.ui_result_sessions == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Configuration steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("default UI settings")
|
|
def step_default_ui_settings(context: Context) -> None:
|
|
for key in (
|
|
"CLEVERAGENTS_TUI_REFRESH_INTERVAL",
|
|
"CLEVERAGENTS_WEB_UI_ENABLED",
|
|
"CLEVERAGENTS_WEB_UI_PORT",
|
|
):
|
|
os.environ.pop(key, None)
|
|
from cleveragents.config.settings import Settings
|
|
|
|
context.ui_settings = Settings()
|
|
|
|
|
|
@given("UI settings with tui_refresh_interval set to {val:d}")
|
|
def step_custom_ui_refresh(context: Context, val: int) -> None:
|
|
os.environ["CLEVERAGENTS_TUI_REFRESH_INTERVAL"] = str(val)
|
|
from cleveragents.config.settings import Settings
|
|
|
|
context.ui_settings = Settings()
|
|
os.environ.pop("CLEVERAGENTS_TUI_REFRESH_INTERVAL", None)
|
|
|
|
|
|
@then("the TUI refresh interval should be {val:d}")
|
|
def step_check_tui_refresh(context: Context, val: int) -> None:
|
|
assert context.ui_settings.tui_refresh_interval == val
|
|
|
|
|
|
@then("web_ui_enabled should be false")
|
|
def step_web_ui_disabled(context: Context) -> None:
|
|
assert context.ui_settings.web_ui_enabled is False
|
|
|
|
|
|
@then("web_ui_port should be {port:d}")
|
|
def step_web_ui_port(context: Context, port: int) -> None:
|
|
assert context.ui_settings.web_ui_port == port
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Web route steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a web UI app backed by mock provider")
|
|
def step_web_ui_app_empty(context: Context) -> None:
|
|
from fastapi.testclient import TestClient
|
|
|
|
from cleveragents.ui.web_app import create_web_app
|
|
|
|
web = create_web_app(provider=_MockProvider())
|
|
context.web_ui_client = TestClient(web)
|
|
|
|
|
|
@given("a web UI app backed by mock provider with plans")
|
|
def step_web_ui_app_plans(context: Context) -> None:
|
|
from fastapi.testclient import TestClient
|
|
|
|
from cleveragents.ui.web_app import create_web_app
|
|
|
|
web = create_web_app(provider=_MockProvider(with_plans=True))
|
|
context.web_ui_client = TestClient(web)
|
|
|
|
|
|
@when('I send a GET to UI route "{path}"')
|
|
def step_get_ui_route(context: Context, path: str) -> None:
|
|
context.web_ui_response = context.web_ui_client.get(path)
|
|
|
|
|
|
@then("the UI response status should be {code:d}")
|
|
def step_ui_status_code(context: Context, code: int) -> None:
|
|
assert context.web_ui_response.status_code == code, (
|
|
f"Expected {code}, got {context.web_ui_response.status_code}"
|
|
)
|
|
|
|
|
|
@then('the UI response JSON should contain key "{key}"')
|
|
def step_ui_json_key(context: Context, key: str) -> None:
|
|
data = context.web_ui_response.json()
|
|
assert key in data, f"Key '{key}' not in {list(data.keys())}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the CLI app is loaded for UI testing")
|
|
def step_cli_loaded_for_ui(context: Context) -> None:
|
|
from cleveragents.cli.main import ensure_cli_commands_registered
|
|
|
|
ensure_cli_commands_registered()
|
|
context.ui_cli_loaded = True
|
|
|
|
|
|
@then('the "{name}" command group should be available')
|
|
def step_command_group_available(context: Context, name: str) -> None:
|
|
from cleveragents.cli.commands.ui import app as ui_app
|
|
|
|
_ = name
|
|
assert ui_app is not None
|
|
|
|
|
|
@given("a CLI runner with web UI disabled")
|
|
def step_cli_web_ui_disabled(context: Context) -> None:
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands.ui import app as ui_app
|
|
|
|
context.ui_cli_runner = CliRunner()
|
|
context.ui_cli_app = ui_app
|
|
|
|
|
|
@when('I invoke the UI CLI with "{cmd}"')
|
|
def step_invoke_ui_cli(context: Context, cmd: str) -> None:
|
|
parts = cmd.split()
|
|
with patch.dict(os.environ, {"CLEVERAGENTS_WEB_UI_ENABLED": "false"}):
|
|
context.ui_cli_result = context.ui_cli_runner.invoke(context.ui_cli_app, parts)
|
|
|
|
|
|
@then("the UI CLI should exit with code {code:d}")
|
|
def step_ui_cli_exit_code(context: Context, code: int) -> None:
|
|
assert context.ui_cli_result.exit_code == code, (
|
|
f"Expected exit code {code}, got {context.ui_cli_result.exit_code}: "
|
|
f"{context.ui_cli_result.output}"
|
|
)
|
|
|
|
|
|
@then('the UI CLI output should mention "{text}"')
|
|
def step_ui_cli_output_contains(context: Context, text: str) -> None:
|
|
assert text.lower() in context.ui_cli_result.output.lower(), (
|
|
f"'{text}' not found in output: {context.ui_cli_result.output}"
|
|
)
|