Luis Mendes
ab911dbdc4
The format_output() function returned a string that callers passed to Rich console.print(), which wraps long lines at terminal width. This injected literal newline characters into JSON string values (e.g. in definition_of_done fields), producing invalid JSON that downstream parsers could not decode (JSONDecodeError: Invalid control character). For machine-readable formats (json, yaml, plain), format_output() now writes the rendered output directly to sys.stdout and returns an empty string. This preserves the exact serialization from json.dumps/ yaml.dump without Rich text processing artifacts. Refs: #746
291 lines
9.3 KiB
Python
291 lines
9.3 KiB
Python
"""Helper utilities for plan diff and artifacts Robot integration tests.
|
|
|
|
Covers:
|
|
- Diff output in multiple formats (rich, plain, JSON, YAML)
|
|
- Artifacts summary rendering
|
|
- Empty changeset guard
|
|
- Apply summary persistence
|
|
- Merge failure handling
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
|
|
from cleveragents.application.services.plan_apply_service import PlanApplyService
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.core.exceptions import PlanError
|
|
from cleveragents.domain.models.core.change import (
|
|
ChangeEntry,
|
|
ChangeOperation,
|
|
InMemoryChangeSetStore,
|
|
SpecChangeSet,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
PlanPhase,
|
|
ProcessingState,
|
|
)
|
|
|
|
|
|
def _create_services() -> tuple[
|
|
PlanLifecycleService, InMemoryChangeSetStore, PlanApplyService
|
|
]:
|
|
"""Create service instances for testing."""
|
|
settings = Settings()
|
|
lifecycle = PlanLifecycleService(settings=settings)
|
|
store = InMemoryChangeSetStore()
|
|
apply_svc = PlanApplyService(lifecycle_service=lifecycle, changeset_store=store)
|
|
return lifecycle, store, apply_svc
|
|
|
|
|
|
def _create_plan_with_changeset(
|
|
lifecycle: PlanLifecycleService,
|
|
store: InMemoryChangeSetStore,
|
|
) -> str:
|
|
"""Create a test action, plan, and changeset. Return plan_id."""
|
|
lifecycle.create_action(
|
|
name="local/robot-diff-test",
|
|
description="Robot diff test",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/planner",
|
|
execution_actor="local/executor",
|
|
)
|
|
plan = lifecycle.use_action(action_name="local/robot-diff-test")
|
|
plan_id = plan.identity.plan_id
|
|
|
|
# Advance through strategize and execute
|
|
lifecycle.start_strategize(plan_id)
|
|
lifecycle.complete_strategize(plan_id)
|
|
plan = lifecycle.get_plan(plan_id)
|
|
if plan.phase == PlanPhase.STRATEGIZE:
|
|
lifecycle.execute_plan(plan_id)
|
|
plan = lifecycle.get_plan(plan_id)
|
|
if plan.phase == PlanPhase.EXECUTE:
|
|
lifecycle.start_execute(plan_id)
|
|
lifecycle.complete_execute(plan_id)
|
|
|
|
# Create changeset
|
|
cs = SpecChangeSet(
|
|
changeset_id="robot-cs-001",
|
|
plan_id=plan_id,
|
|
entries=[
|
|
ChangeEntry(
|
|
plan_id=plan_id,
|
|
resource_id="res-001",
|
|
tool_name="file-write",
|
|
operation=ChangeOperation.CREATE,
|
|
path="src/robot_new.py",
|
|
after_hash="abc123",
|
|
),
|
|
ChangeEntry(
|
|
plan_id=plan_id,
|
|
resource_id="res-002",
|
|
tool_name="file-edit",
|
|
operation=ChangeOperation.MODIFY,
|
|
path="src/robot_existing.py",
|
|
before_hash="old111",
|
|
after_hash="new222",
|
|
),
|
|
],
|
|
)
|
|
store._store[cs.changeset_id] = cs
|
|
|
|
plan = lifecycle.get_plan(plan_id)
|
|
plan.changeset_id = cs.changeset_id
|
|
plan.sandbox_refs = ["sandbox-robot-001"]
|
|
lifecycle._commit_plan(plan)
|
|
|
|
return plan_id
|
|
|
|
|
|
def _create_empty_changeset_plan(
|
|
lifecycle: PlanLifecycleService,
|
|
store: InMemoryChangeSetStore,
|
|
) -> str:
|
|
"""Create a plan with empty changeset. Return plan_id."""
|
|
lifecycle.create_action(
|
|
name="local/robot-empty-test",
|
|
description="Robot empty test",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/planner",
|
|
execution_actor="local/executor",
|
|
)
|
|
plan = lifecycle.use_action(action_name="local/robot-empty-test")
|
|
plan_id = plan.identity.plan_id
|
|
|
|
lifecycle.start_strategize(plan_id)
|
|
lifecycle.complete_strategize(plan_id)
|
|
plan = lifecycle.get_plan(plan_id)
|
|
if plan.phase == PlanPhase.STRATEGIZE:
|
|
lifecycle.execute_plan(plan_id)
|
|
plan = lifecycle.get_plan(plan_id)
|
|
if plan.phase == PlanPhase.EXECUTE:
|
|
lifecycle.start_execute(plan_id)
|
|
lifecycle.complete_execute(plan_id)
|
|
|
|
cs = SpecChangeSet(changeset_id="robot-cs-empty", plan_id=plan_id, entries=[])
|
|
store._store[cs.changeset_id] = cs
|
|
plan = lifecycle.get_plan(plan_id)
|
|
plan.changeset_id = cs.changeset_id
|
|
lifecycle._commit_plan(plan)
|
|
|
|
return plan_id
|
|
|
|
|
|
def _advance_to_apply_processing(lifecycle: PlanLifecycleService, plan_id: str) -> None:
|
|
"""Advance plan to apply/processing."""
|
|
plan = lifecycle.get_plan(plan_id)
|
|
if plan.phase == PlanPhase.EXECUTE and plan.state == ProcessingState.COMPLETE:
|
|
lifecycle.apply_plan(plan_id)
|
|
plan = lifecycle.get_plan(plan_id)
|
|
if plan.phase == PlanPhase.APPLY and plan.state == ProcessingState.QUEUED:
|
|
lifecycle.start_apply(plan_id)
|
|
|
|
|
|
def _diff_output() -> None:
|
|
"""Test diff output in rich format."""
|
|
lifecycle, store, apply_svc = _create_services()
|
|
plan_id = _create_plan_with_changeset(lifecycle, store)
|
|
|
|
output = apply_svc.diff(plan_id, fmt="rich")
|
|
assert "robot-cs-001" in output, f"Changeset ID missing: {output[:200]}"
|
|
assert "src/robot_new.py" in output, f"Path missing: {output[:200]}"
|
|
assert "src/robot_existing.py" in output, f"Path missing: {output[:200]}"
|
|
print("diff-output-ok")
|
|
|
|
|
|
def _artifacts_output() -> None:
|
|
"""Test artifacts output.
|
|
|
|
``format_output`` for machine-readable formats writes directly to
|
|
``sys.stdout`` and returns ``""``. We redirect stdout so the JSON
|
|
payload is captured for parsing.
|
|
"""
|
|
from io import StringIO
|
|
|
|
lifecycle, store, apply_svc = _create_services()
|
|
plan_id = _create_plan_with_changeset(lifecycle, store)
|
|
|
|
buf = StringIO()
|
|
old_stdout = sys.stdout
|
|
sys.stdout = buf
|
|
try:
|
|
output = apply_svc.artifacts(plan_id, fmt="json")
|
|
finally:
|
|
sys.stdout = old_stdout
|
|
# Combine: return value or captured stdout
|
|
effective = output or buf.getvalue().strip()
|
|
parsed = json.loads(effective)
|
|
assert parsed["changeset_id"] == "robot-cs-001"
|
|
assert "sandbox-robot-001" in parsed["sandbox_refs"]
|
|
assert len(parsed["files_changed"]) == 2
|
|
print("artifacts-output-ok")
|
|
|
|
|
|
def _empty_guard() -> None:
|
|
"""Test empty changeset guard blocks apply."""
|
|
lifecycle, store, apply_svc = _create_services()
|
|
plan_id = _create_empty_changeset_plan(lifecycle, store)
|
|
|
|
try:
|
|
apply_svc.guard_empty_changeset(plan_id)
|
|
raise AssertionError("Should have raised PlanError")
|
|
except PlanError as e:
|
|
assert "empty" in e.message.lower()
|
|
print("empty-guard-ok")
|
|
|
|
|
|
def _empty_guard_allow() -> None:
|
|
"""Test empty changeset guard allows with flag."""
|
|
lifecycle, store, apply_svc = _create_services()
|
|
plan_id = _create_empty_changeset_plan(lifecycle, store)
|
|
|
|
result = apply_svc.guard_empty_changeset(plan_id, allow_empty=True)
|
|
assert result is True
|
|
print("empty-guard-allow-ok")
|
|
|
|
|
|
def _apply_summary() -> None:
|
|
"""Test apply summary persistence."""
|
|
lifecycle, store, apply_svc = _create_services()
|
|
plan_id = _create_plan_with_changeset(lifecycle, store)
|
|
_advance_to_apply_processing(lifecycle, plan_id)
|
|
|
|
apply_svc.persist_apply_summary(plan_id, files_changed=7, validations_run=2)
|
|
|
|
plan = lifecycle.get_plan(plan_id)
|
|
assert plan.error_details is not None
|
|
assert plan.error_details["apply_files_changed"] == "7"
|
|
assert plan.error_details["apply_validations_run"] == "2"
|
|
assert "apply_completed_at" in plan.error_details
|
|
print("apply-summary-ok")
|
|
|
|
|
|
def _merge_failure() -> None:
|
|
"""Test merge failure handling."""
|
|
lifecycle, store, apply_svc = _create_services()
|
|
plan_id = _create_plan_with_changeset(lifecycle, store)
|
|
_advance_to_apply_processing(lifecycle, plan_id)
|
|
|
|
apply_svc.handle_merge_failure(plan_id, conflict_details="Conflict in src/main.py")
|
|
|
|
plan = lifecycle.get_plan(plan_id)
|
|
assert plan.processing_state == ProcessingState.ERRORED
|
|
assert "Merge failed" in (plan.error_message or "")
|
|
assert plan.error_details is not None
|
|
assert "merge_conflict" in plan.error_details
|
|
print("merge-failure-ok")
|
|
|
|
|
|
def _diff_json() -> None:
|
|
"""Test diff JSON format."""
|
|
lifecycle, store, apply_svc = _create_services()
|
|
plan_id = _create_plan_with_changeset(lifecycle, store)
|
|
|
|
output = apply_svc.diff(plan_id, fmt="json")
|
|
parsed = json.loads(output)
|
|
assert "entries" in parsed
|
|
assert "summary" in parsed
|
|
assert len(parsed["entries"]) == 2
|
|
print("diff-json-ok")
|
|
|
|
|
|
def _diff_yaml() -> None:
|
|
"""Test diff YAML format."""
|
|
lifecycle, store, apply_svc = _create_services()
|
|
plan_id = _create_plan_with_changeset(lifecycle, store)
|
|
|
|
output = apply_svc.diff(plan_id, fmt="yaml")
|
|
assert "changeset_id" in output
|
|
assert "robot-cs-001" in output
|
|
print("diff-yaml-ok")
|
|
|
|
|
|
COMMANDS: dict[str, object] = {
|
|
"diff-output": _diff_output,
|
|
"artifacts-output": _artifacts_output,
|
|
"empty-guard": _empty_guard,
|
|
"empty-guard-allow": _empty_guard_allow,
|
|
"apply-summary": _apply_summary,
|
|
"merge-failure": _merge_failure,
|
|
"diff-json": _diff_json,
|
|
"diff-yaml": _diff_yaml,
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
|
|
print(f"Available: {', '.join(COMMANDS)}", file=sys.stderr)
|
|
sys.exit(1)
|
|
cmd = sys.argv[1]
|
|
func = COMMANDS.get(cmd)
|
|
if func is None:
|
|
print(f"Unknown command: {cmd}", file=sys.stderr)
|
|
sys.exit(1)
|
|
func() # type: ignore[operator]
|