cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity
Files
7aa50ac4895f91b75e39a36cf1dca5bd0d21bb69
cleveragents-core/robot/helper_m1_e2e_verification.py
T
HAL9000 7aa50ac489 fix(e2e): restore M5/WF14 tests broken by JSON envelope and stale config 
- Add NO_COLOR=1 to E2E Suite Setup and Test Environment to prevent
  Rich from injecting ANSI escape codes into JSON output, which breaks
  Extract JSON From Stdout and JSON assertions in M5/WF14 acceptance tests.

- Add reset_global_state() calls to all M1/M2/M4/M5 E2E verification
  helpers to clear Settings singleton, DI container, provider registry,
  and SQLAlchemy engine cache between parallel pabot worker runs.

- Propagate NO_COLOR=1 at suite-setup level in E2E and integration
  test resource files so all Run Process subprocesses inherit clean
  terminal output mode.

Root causes:
1. Missing NO_COLOR=1 caused Rich ANSI codes in CLI JSON output,
   making json.loads() failures in robot Extract JSON From Stdout
2. Stale Settings singleton carried provider keys and DB paths between
   pabot workers, causing UNIQUE constraint violations and stale config
   lookups in context policy and server mode E2E tests.

Related: PR #9912
2026-05-01 00:59:02 +00:00

681 lines
22 KiB
Python
Raw Blame History

"""Robot helper for M1 E2E verification suite.
Exercises the complete M1 success-criteria sequence:
1. Action creation from YAML config
2. Git-checkout resource registration
3. Project creation and resource linking
4. Plan use / execute / diff / apply lifecycle
5. SQLite persistence assertions (Plan + Action records)
6. ChangeSet built from tool invocations (not parsed output)
7. Git worktree sandbox creates isolated working directory
8. Sandbox changes do not affect original until Apply
9. Post-apply commit exists in the target repo
Each subcommand prints a sentinel on success.
Exit code 0 = pass, 1 = failure.
CLI-facing tests (1-4) invoke the real ``agents`` CLI via subprocess
without mocking the service layer. Domain-level tests (5-9) exercise
real application code directly.
"""
from __future__ import annotations
import os
import shutil
import subprocess
import sys
from collections.abc import Callable
from datetime import datetime
from pathlib import Path
from typing import NoReturn
# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
# Ensure robot/ is on the import path for helper_e2e_common.
_ROBOT = str(Path(__file__).resolve().parent)
if _ROBOT not in sys.path:
sys.path.insert(0, _ROBOT)
from helper_e2e_common import ( # noqa: E402
cleanup_workspace,
init_bare_git_repo,
is_expected_provider_unavailable,
run_cli,
setup_workspace,
write_yaml,
)
from cleveragents.domain.models.core.action import Action, ActionState # noqa: E402
from cleveragents.domain.models.core.change import ( # noqa: E402
ChangeEntry,
ChangeOperation,
InMemoryChangeSetStore,
ToolInvocation,
)
from cleveragents.domain.models.core.plan import ( # noqa: E402
from helpers_common import reset_global_state # noqa: E402
AutomationProfileProvenance,
AutomationProfileRef,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
ProjectLink,
)
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8M"
_VALID_ACTION_YAML = """\
name: local/m1-verify-action
description: M1 verification action
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: All M1 criteria pass
"""
def _extract_plan_id(output: str) -> str | None:
"""Extract a ULID plan_id from plain CLI output.
Searches for a 26-character ULID pattern (uppercase alphanumeric)
which is the format used by ``python-ulid``.
"""
import re
match = re.search(r"\b([0-9A-Z]{26})\b", output)
return match.group(1) if match else None
def _mock_action(name: str = "local/m1-verify-action") -> Action:
"""Create a minimal valid Action for testing."""
return Action(
namespaced_name=NamespacedName.parse(name),
description="M1 verification action",
long_description=None,
definition_of_done="All M1 criteria pass",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
state=ActionState.AVAILABLE,
reusable=True,
read_only=False,
created_at=datetime.now(),
updated_at=datetime.now(),
created_by=None,
)
def _mock_plan(
phase: PlanPhase = PlanPhase.STRATEGIZE,
state: ProcessingState = ProcessingState.QUEUED,
plan_id: str = _PLAN_ULID,
) -> Plan:
"""Create a minimal valid Plan for testing."""
now = datetime.now()
return Plan(
identity=PlanIdentity(plan_id=plan_id),
namespaced_name=NamespacedName.parse("local/m1-verify-plan"),
description="M1 verification plan",
definition_of_done="All M1 criteria pass",
action_name="local/m1-verify-action",
phase=phase,
processing_state=state,
project_links=[ProjectLink(project_name="local/m1-project")],
arguments={"target_coverage": 97},
arguments_order=["target_coverage"],
automation_profile=AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
),
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
reusable=True,
read_only=False,
created_by=None,
timestamps=PlanTimestamps(created_at=now, updated_at=now),
)
def _make_store() -> tuple[InMemoryChangeSetStore, str]:
"""Create an InMemoryChangeSetStore with one changeset started."""
store = InMemoryChangeSetStore()
cid = store.start(_PLAN_ULID)
return store, cid
def _fail(msg: str) -> NoReturn:
"""Print failure message and exit."""
print(f"FAIL: {msg}", file=sys.stderr)
raise SystemExit(1)
# ---------------------------------------------------------------------------
# Subcommand: action-create
# ---------------------------------------------------------------------------
def action_create_from_yaml() -> None:
"""Verify action create from YAML config via real CLI subprocess."""
reset_global_state()
workspace = setup_workspace(prefix="m1_action_")
yaml_path = write_yaml(_VALID_ACTION_YAML)
try:
result = run_cli("action", "create", "--config", yaml_path, workspace=workspace)
if result.returncode != 0:
_fail(f"action create rc={result.returncode}\n{result.stderr}")
# Verify persistence: the action should be retrievable
show = run_cli(
"action",
"show",
"local/m1-verify-action",
"--format",
"plain",
workspace=workspace,
)
if show.returncode != 0:
_fail(f"action show rc={show.returncode}\n{show.stderr}")
if "local/m1-verify-action" not in show.stdout:
_fail(f"action not found in show output:\n{show.stdout}")
print("m1-action-create-ok")
finally:
os.unlink(yaml_path)
cleanup_workspace(workspace)
# ---------------------------------------------------------------------------
# Subcommand: resource-register
# ---------------------------------------------------------------------------
def resource_register_git_checkout() -> None:
"""Register a git-checkout resource via real CLI subprocess."""
reset_global_state()
workspace = setup_workspace(prefix="m1_resource_")
repo_dir = init_bare_git_repo()
try:
result = run_cli(
"resource",
"add",
"git-checkout",
"local/m1-repo",
"--path",
repo_dir,
"--branch",
"main",
"--format",
"plain",
workspace=workspace,
)
if result.returncode != 0:
_fail(f"resource add rc={result.returncode}\n{result.stderr}")
# Verify persistence: the resource should be retrievable
show = run_cli(
"resource",
"show",
"local/m1-repo",
"--format",
"plain",
workspace=workspace,
)
if show.returncode != 0:
_fail(f"resource show rc={show.returncode}\n{show.stderr}")
if "git-checkout" not in show.stdout:
_fail(f"resource type not git-checkout:\n{show.stdout}")
print("m1-resource-register-ok")
finally:
shutil.rmtree(repo_dir, ignore_errors=True)
cleanup_workspace(workspace)
# ---------------------------------------------------------------------------
# Subcommand: project-create-link
# ---------------------------------------------------------------------------
def project_create_and_link() -> None:
"""Create a project and link a resource via real CLI subprocess."""
reset_global_state()
workspace = setup_workspace(prefix="m1_project_")
repo_dir = init_bare_git_repo()
try:
# First register a resource (dependency for --resource flag)
r1 = run_cli(
"resource",
"add",
"git-checkout",
"local/m1-repo",
"--path",
repo_dir,
"--branch",
"main",
"--format",
"plain",
workspace=workspace,
)
if r1.returncode != 0:
_fail(f"resource add rc={r1.returncode}\n{r1.stderr}")
# Create project with resource link
r2 = run_cli(
"project",
"create",
"local/m1-project",
"--description",
"M1 test project",
"--resource",
"local/m1-repo",
"--format",
"plain",
workspace=workspace,
)
if r2.returncode != 0:
_fail(f"project create rc={r2.returncode}\n{r2.stderr}")
# Verify persistence
r3 = run_cli(
"project",
"show",
"local/m1-project",
"--format",
"plain",
workspace=workspace,
)
if r3.returncode != 0:
_fail(f"project show rc={r3.returncode}\n{r3.stderr}")
if "local/m1-project" not in r3.stdout:
_fail(f"project not found in show output:\n{r3.stdout}")
print("m1-project-create-link-ok")
finally:
shutil.rmtree(repo_dir, ignore_errors=True)
cleanup_workspace(workspace)
# ---------------------------------------------------------------------------
# Subcommand: plan-lifecycle
# ---------------------------------------------------------------------------
def plan_full_lifecycle() -> None:
"""Run the plan lifecycle via real CLI subprocesses.
Invokes action create -> plan use -> plan show -> plan execute ->
plan diff -> plan apply using the real ``agents`` CLI
binary. Without an AI backend the plan stays in ``strategize /
queued``, so ``execute``, ``diff``, and ``apply``
return graceful "not ready" messages rather than crashing. The
test verifies the full CLI wiring (DI, persistence, env vars)
end-to-end. ChangeSet verification uses the domain-level tests
(``changeset-invocations``), not this CLI lifecycle test.
"""
reset_global_state()
workspace = setup_workspace(prefix="m1_plan_")
repo_dir = init_bare_git_repo()
yaml_path = write_yaml(_VALID_ACTION_YAML)
try:
# Setup: register resource + create project
r_res = run_cli(
"resource",
"add",
"git-checkout",
"local/m1-repo",
"--path",
repo_dir,
"--branch",
"main",
workspace=workspace,
)
if r_res.returncode != 0:
_fail(f"resource add: {r_res.stderr}")
r_proj = run_cli(
"project",
"create",
"local/m1-project",
"--resource",
"local/m1-repo",
workspace=workspace,
)
if r_proj.returncode != 0:
_fail(f"project create: {r_proj.stderr}")
# Step 1: Create action
r1 = run_cli("action", "create", "--config", yaml_path, workspace=workspace)
if r1.returncode != 0:
_fail(f"action create: {r1.stderr}")
# Step 2: Plan use (--format plain to extract plan_id)
r2 = run_cli(
"plan",
"use",
"local/m1-verify-action",
"local/m1-project",
"--format",
"plain",
workspace=workspace,
)
if r2.returncode != 0:
_fail(f"plan use: {r2.stderr}")
# Extract plan_id from plain output
plan_id = _extract_plan_id(r2.stdout)
if not plan_id:
_fail(f"could not extract plan_id from plan use output:\n{r2.stdout}")
# Step 3: Verify plan persisted — status must find it
r3 = run_cli(
"plan",
"status",
plan_id,
"--format",
"plain",
workspace=workspace,
)
if r3.returncode != 0:
_fail(f"plan status: {r3.stderr}")
if plan_id not in r3.stdout:
_fail(f"plan_id not in plan status output:\n{r3.stdout}")
# Step 4: Plan execute — plan is strategize/queued without
# an AI backend so it correctly rejects with "not ready".
# The test verifies no internal server error (500); a
# controlled abort with a user-facing message is expected.
r4 = run_cli("plan", "execute", plan_id, workspace=workspace)
combined4 = r4.stdout + r4.stderr
if "Traceback" in combined4:
_fail(f"plan execute crashed:\n{combined4}")
if "INTERNAL" in combined4 and not is_expected_provider_unavailable(combined4):
_fail(f"plan execute crashed:\n{combined4}")
# Step 5: Plan diff — similar: plan not in execute phase
r5 = run_cli("plan", "diff", plan_id, workspace=workspace)
combined5 = r5.stdout + r5.stderr
if "Traceback" in combined5:
_fail(f"plan diff crashed:\n{combined5}")
if "INTERNAL" in combined5 and not is_expected_provider_unavailable(combined5):
_fail(f"plan diff crashed:\n{combined5}")
# Step 6: Plan apply — similar: plan not ready
r6 = run_cli("plan", "apply", "--yes", plan_id, workspace=workspace)
combined6 = r6.stdout + r6.stderr
if "Traceback" in combined6:
_fail(f"plan apply crashed:\n{combined6}")
if "INTERNAL" in combined6 and not is_expected_provider_unavailable(combined6):
_fail(f"plan apply crashed:\n{combined6}")
print("m1-plan-lifecycle-ok")
finally:
os.unlink(yaml_path)
shutil.rmtree(repo_dir, ignore_errors=True)
cleanup_workspace(workspace)
# ---------------------------------------------------------------------------
# Subcommand: sqlite-persistence
# ---------------------------------------------------------------------------
def sqlite_persistence_check() -> None:
"""Verify Plan and Action records persist to SQLite via domain models.
Uses the in-memory ChangeSetStore and domain model constructors
to verify data survives serialisation round-trips.
"""
reset_global_state()
# Create Action and verify serialisation
action = _mock_action()
action_dict = {
"namespaced_name": str(action.namespaced_name),
"state": action.state.value,
"description": action.description,
"definition_of_done": action.definition_of_done,
"strategy_actor": action.strategy_actor,
"execution_actor": action.execution_actor,
}
if action_dict["state"] != "available":
_fail(f"action state={action_dict['state']}")
# Create Plan and verify serialisation
plan = _mock_plan()
plan_dict = {
"plan_id": plan.identity.plan_id,
"namespaced_name": str(plan.namespaced_name),
"phase": plan.phase.value,
"processing_state": plan.processing_state.value,
"action_name": plan.action_name,
}
if plan_dict["phase"] != "strategize":
_fail(f"plan phase={plan_dict['phase']}")
if plan_dict["plan_id"] != _PLAN_ULID:
_fail(f"plan_id mismatch: {plan_dict['plan_id']}")
# Verify ChangeSet round-trip
store = InMemoryChangeSetStore()
cid = store.start(_PLAN_ULID)
store.record(
cid,
ChangeEntry(
plan_id=_PLAN_ULID,
resource_id="res-persist",
tool_name="builtin/file-write",
operation=ChangeOperation.CREATE,
path="src/persisted.py",
),
)
cs = store.get(cid)
if cs is None:
_fail("changeset not found")
if cs.plan_id != _PLAN_ULID:
_fail(f"changeset plan_id mismatch: {cs.plan_id}")
if len(cs.entries) != 1:
_fail(f"expected 1 entry, got {len(cs.entries)}")
# Verify changesets retrievable by plan_id
plan_sets = store.get_for_plan(_PLAN_ULID)
if len(plan_sets) != 1:
_fail(f"expected 1 changeset for plan, got {len(plan_sets)}")
print("m1-sqlite-persistence-ok")
# ---------------------------------------------------------------------------
# Subcommand: changeset-from-invocations
# ---------------------------------------------------------------------------
def changeset_from_tool_invocations() -> None:
"""Verify ChangeSet is built from tool invocations, not parsed output.
Creates ToolInvocation records and verifies ChangeEntry objects
are linked via tool_name, not extracted from CLI output text.
"""
reset_global_state()
store, cid = _make_store()
# Simulate tool invocations producing ChangeEntry records
inv = ToolInvocation(
plan_id=_PLAN_ULID,
tool_name="builtin/file-write",
arguments={"path": "src/tool_generated.py", "content": "# new"},
success=True,
duration_ms=42.0,
sequence_number=0,
)
# Record the change entry that this invocation produced
entry = ChangeEntry(
plan_id=_PLAN_ULID,
resource_id="res-inv",
tool_name=inv.tool_name,
operation=ChangeOperation.CREATE,
path="src/tool_generated.py",
)
store.record(cid, entry)
# Link invocation to change
inv.change_ids.append(entry.entry_id)
# Verify: entry tool_name matches invocation tool_name
cs = store.get(cid)
if cs is None:
_fail("changeset not found")
if cs.entries[0].tool_name != "builtin/file-write":
_fail(f"tool_name mismatch: {cs.entries[0].tool_name}")
if entry.entry_id not in inv.change_ids:
_fail("entry not linked to invocation")
print("m1-changeset-invocations-ok")
# ---------------------------------------------------------------------------
# Subcommand: sandbox-isolation
# ---------------------------------------------------------------------------
def sandbox_isolation_check() -> None:
"""Verify git worktree sandbox creates isolated working directory.
Also verifies sandbox changes do not affect the original repo
until Apply (commit + merge).
"""
reset_global_state()
from cleveragents.infrastructure.sandbox.git_worktree import (
GitWorktreeSandbox,
)
from cleveragents.infrastructure.sandbox.protocol import SandboxStatus
repo_dir = init_bare_git_repo()
try:
# Create sandbox
sandbox = GitWorktreeSandbox(
resource_id="res-sandbox",
original_path=repo_dir,
)
ctx = sandbox.create(plan_id=_PLAN_ULID)
# Verify sandbox created in isolated directory
if not os.path.isdir(ctx.sandbox_path):
_fail("sandbox path does not exist")
if ctx.sandbox_path == repo_dir:
_fail("sandbox path same as original")
if sandbox.status != SandboxStatus.CREATED:
_fail(f"expected CREATED, got {sandbox.status}")
# Write a file in the sandbox
sandbox_file = sandbox.get_path("src/sandbox_only.py")
os.makedirs(os.path.dirname(sandbox_file), exist_ok=True)
with open(sandbox_file, "w") as f:
f.write("# created in sandbox\n")
# Verify original repo does NOT have the sandbox file
original_file = os.path.join(repo_dir, "src", "sandbox_only.py")
if os.path.exists(original_file):
_fail("sandbox change leaked to original before apply")
# Commit and verify the merge applies to original
result = sandbox.commit("sandbox: apply changes")
if not result.success:
_fail(f"commit failed: {result.error}")
if result.commit_ref is None:
_fail("no commit ref after commit")
# Now original repo should have the file
if not os.path.exists(original_file):
_fail("file not in original after commit/merge")
sandbox.cleanup()
print("m1-sandbox-isolation-ok")
finally:
shutil.rmtree(repo_dir, ignore_errors=True)
# ---------------------------------------------------------------------------
# Subcommand: post-apply-commit
# ---------------------------------------------------------------------------
def post_apply_commit_check() -> None:
"""Verify post-apply commit exists in the target repo.
Creates a git worktree sandbox, writes a file, commits, and
verifies the commit is visible in git log on the original branch.
"""
reset_global_state()
from cleveragents.infrastructure.sandbox.git_worktree import (
GitWorktreeSandbox,
)
repo_dir = init_bare_git_repo()
try:
sandbox = GitWorktreeSandbox(
resource_id="res-commit-check",
original_path=repo_dir,
)
sandbox.create(plan_id=_PLAN_ULID)
# Write and commit
new_file = sandbox.get_path("post_apply_test.txt")
with open(new_file, "w") as f:
f.write("post-apply verification content\n")
commit_result = sandbox.commit("sandbox: post-apply test")
if not commit_result.success:
_fail(f"commit failed: {commit_result.error}")
sandbox.cleanup()
# Verify commit is in git log of original repo
log_output = subprocess.run(
["git", "log", "--oneline", "-5"],
cwd=repo_dir,
capture_output=True,
text=True,
check=True,
)
if "post-apply test" not in log_output.stdout:
_fail(f"commit not found in git log:\n{log_output.stdout}")
# Verify the file exists on disk in original
if not os.path.exists(os.path.join(repo_dir, "post_apply_test.txt")):
_fail("post-apply file not on disk")
print("m1-post-apply-commit-ok")
finally:
shutil.rmtree(repo_dir, ignore_errors=True)
# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"action-create": action_create_from_yaml,
"resource-register": resource_register_git_checkout,
"project-create-link": project_create_and_link,
"plan-lifecycle": plan_full_lifecycle,
"sqlite-persistence": sqlite_persistence_check,
"changeset-invocations": changeset_from_tool_invocations,
"sandbox-isolation": sandbox_isolation_check,
"post-apply-commit": post_apply_commit_check,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(
f"Usage: helper_m1_e2e_verification.py <{'|'.join(_COMMANDS)}>",
)
return 1
command = sys.argv[1]
handler = _COMMANDS.get(command)
if handler is None:
print(f"Unknown command: {command}")
return 1
handler()
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink