forked from cleveragents/cleveragents-core
test(robot): make wf15 rollback helper use real git sandbox
This commit is contained in:
@@ -2,6 +2,10 @@
|
||||
|
||||
## Unreleased
|
||||
|
||||
- Added Robot Framework integration test and Python helper for Specification
|
||||
Workflow Example 15 — disaster recovery, rollback a failed apply. Covers
|
||||
checkpoint creation, intentional apply failure, rollback to checkpoint,
|
||||
and verify restore. (#779)
|
||||
- Added TDD bug-capture tests for bug #1076 — `use_action()` does not
|
||||
propagate `automation_profile` to Plan. Three Behave BDD scenarios
|
||||
(`@tdd_bug @tdd_bug_1076 @tdd_expected_fail`) verify the full precedence
|
||||
|
||||
@@ -0,0 +1,494 @@
|
||||
"""Robot helper for Workflow Example 15 — disaster recovery, rollback a failed apply."""
|
||||
|
||||
# ruff: noqa: E402
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from collections.abc import Callable, Iterator
|
||||
from pathlib import Path
|
||||
from typing import Any, NoReturn
|
||||
|
||||
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
||||
if _SRC not in sys.path:
|
||||
sys.path.insert(0, _SRC)
|
||||
_ROBOT = str(Path(__file__).resolve().parent)
|
||||
if _ROBOT not in sys.path:
|
||||
sys.path.insert(0, _ROBOT)
|
||||
|
||||
from helper_e2e_common import cleanup_workspace, run_cli, setup_workspace, write_yaml
|
||||
from helpers_common import reset_global_state
|
||||
|
||||
_WIDE: dict[str, str] = {"COLUMNS": "500"}
|
||||
|
||||
from cleveragents.application.services.checkpoint_service import CheckpointService
|
||||
from cleveragents.application.services.decision_service import DecisionService
|
||||
from cleveragents.config.settings import Settings
|
||||
from cleveragents.domain.models.core.decision import (
|
||||
ContextSnapshot,
|
||||
Decision,
|
||||
DecisionType,
|
||||
ResourceRef,
|
||||
)
|
||||
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
||||
|
||||
_PROJECT = "local/api-service"
|
||||
_RES_DB = "01HXM8D2ZK4Q7C2B3F2R4VYV6K"
|
||||
_RES_POOL = "01HXM8E2ZK4Q7C2B3F2R4VYV6M"
|
||||
|
||||
_ACTION_YAML = """\
|
||||
name: local/optimize-db-connections
|
||||
description: Optimize database connection pooling for PostgreSQL replicas
|
||||
definition_of_done: Connection pool tuned without exceeding max_connections
|
||||
strategy_actor: openai/gpt-4
|
||||
execution_actor: openai/gpt-4
|
||||
arguments:
|
||||
- name: target_module
|
||||
type: string
|
||||
required: false
|
||||
description: Module path to optimise
|
||||
default: src/db
|
||||
"""
|
||||
|
||||
|
||||
def _fail(msg: str) -> NoReturn:
|
||||
print(f"FAIL: {msg}", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
|
||||
|
||||
def _extract_plan_id(output: str) -> str | None:
|
||||
m = re.search(r"\b([0-9A-Z]{26})\b", output)
|
||||
return m.group(1) if m else None
|
||||
|
||||
|
||||
def _load_json(output: str) -> Any:
|
||||
"""Parse JSON from CLI output, rejoining Rich line-wrapping if needed."""
|
||||
text = output.strip()
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
dec = json.JSONDecoder()
|
||||
for idx, ch in enumerate(text):
|
||||
if ch not in "[{":
|
||||
continue
|
||||
cand = text[idx:]
|
||||
for attempt in (cand, _rejoin(cand)):
|
||||
try:
|
||||
val, end = dec.raw_decode(attempt)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
if not attempt[end:].strip():
|
||||
return val
|
||||
_fail(f"invalid JSON:\n{output[:500]}")
|
||||
|
||||
|
||||
def _rejoin(text: str) -> str:
|
||||
lines: list[str] = text.split("\n")
|
||||
out: list[str] = []
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
ln = lines[i]
|
||||
qc = sum(
|
||||
1 for j, c in enumerate(ln) if c == '"' and (j == 0 or ln[j - 1] != "\\")
|
||||
)
|
||||
if qc % 2 == 1 and i + 1 < len(lines):
|
||||
lines[i + 1] = ln + lines[i + 1].lstrip()
|
||||
i += 1
|
||||
continue
|
||||
out.append(ln)
|
||||
i += 1
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _ws(prefix: str, *, yaml: bool = True) -> Iterator[tuple[str, str | None]]:
|
||||
"""Workspace + optional action YAML context manager."""
|
||||
ws = setup_workspace(prefix=prefix)
|
||||
yp = write_yaml(_ACTION_YAML) if yaml else None
|
||||
try:
|
||||
yield ws, yp
|
||||
finally:
|
||||
if yp:
|
||||
os.unlink(yp)
|
||||
cleanup_workspace(ws)
|
||||
|
||||
|
||||
def _ok(result: Any, label: str) -> None:
|
||||
c = result.stdout + result.stderr
|
||||
if "INTERNAL" in c or "Traceback" in c:
|
||||
_fail(f"{label} crashed:\n{c}")
|
||||
|
||||
|
||||
def _cli(*args: str, ws: str) -> Any:
|
||||
r = run_cli(*args, workspace=ws, env_extra=_WIDE)
|
||||
_ok(r, args[0] if args else "cli")
|
||||
return r
|
||||
|
||||
|
||||
def _git(*args: str, cwd: str) -> subprocess.CompletedProcess[str]:
|
||||
"""Run git command and fail fast with stderr context."""
|
||||
try:
|
||||
return subprocess.run(
|
||||
["git", *args],
|
||||
cwd=cwd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
except subprocess.CalledProcessError as exc:
|
||||
_fail(f"git {' '.join(args)} failed:\nstdout={exc.stdout}\nstderr={exc.stderr}")
|
||||
|
||||
|
||||
def _prepare_checkpoint_sandbox(ws: str) -> tuple[str, str]:
|
||||
"""Create a real git sandbox and return (sandbox_path, checkpoint_ref)."""
|
||||
sandbox = Path(ws) / "wf15-sandbox"
|
||||
sandbox.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
_git("init", cwd=str(sandbox))
|
||||
_git("config", "user.email", "wf15@example.invalid", cwd=str(sandbox))
|
||||
_git("config", "user.name", "WF15 Test", cwd=str(sandbox))
|
||||
|
||||
tracked = sandbox / "db_pool.conf"
|
||||
tracked.write_text("pool_size=20\n", encoding="utf-8")
|
||||
_git("add", "db_pool.conf", cwd=str(sandbox))
|
||||
_git("commit", "-m", "baseline checkpoint", cwd=str(sandbox))
|
||||
checkpoint_ref = _git("rev-parse", "HEAD", cwd=str(sandbox)).stdout.strip()
|
||||
if not checkpoint_ref:
|
||||
_fail("failed to resolve baseline checkpoint commit")
|
||||
|
||||
# mutate tracked + create untracked so rollback has something to restore
|
||||
tracked.write_text("pool_size=100\n", encoding="utf-8")
|
||||
(sandbox / "scratch.tmp").write_text("temporary\n", encoding="utf-8")
|
||||
|
||||
return str(sandbox), checkpoint_ref
|
||||
|
||||
|
||||
def _make_settings(database_url: str = "sqlite:///:memory:") -> Settings:
|
||||
prev = os.environ.get("CLEVERAGENTS_DATABASE_URL")
|
||||
os.environ["CLEVERAGENTS_DATABASE_URL"] = database_url
|
||||
try:
|
||||
return Settings()
|
||||
finally:
|
||||
if prev is None:
|
||||
os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
|
||||
else:
|
||||
os.environ["CLEVERAGENTS_DATABASE_URL"] = prev
|
||||
|
||||
|
||||
def _create_plan(ws: str, yp: str) -> str:
|
||||
"""Create action and plan, return plan_id."""
|
||||
r = run_cli("action", "create", "--config", yp, workspace=ws, env_extra=_WIDE)
|
||||
if r.returncode != 0:
|
||||
_fail(f"action create: {r.stderr}")
|
||||
r2 = run_cli(
|
||||
"plan",
|
||||
"use",
|
||||
"local/optimize-db-connections",
|
||||
_PROJECT,
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
if r2.returncode != 0:
|
||||
_fail(f"plan use: {r2.stderr}")
|
||||
pd = _load_json(r2.stdout)
|
||||
pid = pd.get("plan_id") if isinstance(pd, dict) else _extract_plan_id(r2.stdout)
|
||||
if not isinstance(pid, str) or not pid:
|
||||
_fail(f"missing plan_id: {pd}")
|
||||
return pid
|
||||
|
||||
|
||||
def _snap(prefix: str, res_id: str, path: str) -> ContextSnapshot:
|
||||
return ContextSnapshot(
|
||||
hot_context_hash=f"sha256:wf15_{prefix}",
|
||||
hot_context_ref=f"store://snapshots/wf15_{prefix}",
|
||||
relevant_resources=[ResourceRef(resource_id=res_id, path=path)],
|
||||
actor_state_ref=f"checkpoint://actor/wf15_{prefix}",
|
||||
)
|
||||
|
||||
|
||||
def _seed_decisions(
|
||||
svc: DecisionService,
|
||||
plan_id: str,
|
||||
) -> tuple[Decision, Decision, Decision]:
|
||||
"""Seed 3-node tree: root -> strategy -> implementation_choice (root cause)."""
|
||||
kw: dict[str, Any] = {"plan_id": plan_id}
|
||||
root = svc.record_decision(
|
||||
**kw,
|
||||
decision_type=DecisionType.PROMPT_DEFINITION,
|
||||
question="How should we optimize the DB connection pool?",
|
||||
chosen_option="Increase pool_size from 20 to 100 across all replicas",
|
||||
alternatives_considered=["Add connection health checks", "Use PgBouncer"],
|
||||
confidence_score=0.85,
|
||||
rationale="Higher pool size reduces wait time under load.",
|
||||
context_snapshot=_snap("root", _RES_DB, "src/db/config.py"),
|
||||
)
|
||||
strategy = svc.record_decision(
|
||||
**kw,
|
||||
decision_type=DecisionType.STRATEGY_CHOICE,
|
||||
question="What pool sizing strategy to use?",
|
||||
chosen_option="Set pool_size=100 per replica, 3 replicas total",
|
||||
parent_decision_id=root.decision_id,
|
||||
alternatives_considered=["pool_size=50 with overflow"],
|
||||
confidence_score=0.78,
|
||||
rationale="Maximise throughput with dedicated connections.",
|
||||
context_snapshot=_snap("strategy", _RES_POOL, "src/db/pool.py"),
|
||||
)
|
||||
impl = svc.record_decision(
|
||||
**kw,
|
||||
decision_type=DecisionType.STRATEGY_CHOICE,
|
||||
question="How to apply pool_size across replicas?",
|
||||
chosen_option=(
|
||||
"pool_size=100 on each of 3 replicas (300 total vs max_connections=150)"
|
||||
),
|
||||
parent_decision_id=strategy.decision_id,
|
||||
alternatives_considered=["Keep pool_size=20 and add health checks"],
|
||||
confidence_score=0.60,
|
||||
rationale="Uniform config is simpler but exceeds max_connections.",
|
||||
context_snapshot=_snap("impl", _RES_DB, "src/db/pool_config.py"),
|
||||
)
|
||||
return root, strategy, impl
|
||||
|
||||
|
||||
def _get_dsvc(db_url: str) -> DecisionService:
|
||||
return DecisionService(
|
||||
settings=_make_settings(db_url),
|
||||
unit_of_work=UnitOfWork(db_url),
|
||||
)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Subcommands
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
|
||||
def wf15_plan_status_errored() -> None:
|
||||
"""Create a plan, check status shows plan lifecycle info."""
|
||||
with _ws("wf15_status_") as (ws, yp):
|
||||
assert yp is not None
|
||||
pid = _create_plan(ws, yp)
|
||||
r = run_cli(
|
||||
"plan",
|
||||
"status",
|
||||
pid,
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
_ok(r, "plan status")
|
||||
if r.returncode != 0:
|
||||
_fail(f"plan status rc={r.returncode}\n{r.stderr}")
|
||||
sd = _load_json(r.stdout)
|
||||
if not isinstance(sd, dict):
|
||||
_fail(f"status not dict: {sd}")
|
||||
if sd.get("plan_id") != pid:
|
||||
_fail(f"plan_id mismatch: {sd.get('plan_id')} != {pid}")
|
||||
if not sd.get("phase"):
|
||||
_fail(f"missing phase: {sd}")
|
||||
if not sd.get("processing_state"):
|
||||
_fail(f"missing processing_state: {sd}")
|
||||
print("wf15-status-ok")
|
||||
|
||||
|
||||
def wf15_tree_and_root_cause() -> None:
|
||||
"""Seed decisions, call plan tree, verify 3-level tree."""
|
||||
with _ws("wf15_tree_") as (ws, yp):
|
||||
assert yp is not None
|
||||
pid = _create_plan(ws, yp)
|
||||
db_url = os.environ["CLEVERAGENTS_DATABASE_URL"]
|
||||
dsvc = _get_dsvc(db_url)
|
||||
root, strategy, impl = _seed_decisions(dsvc, pid)
|
||||
r = run_cli(
|
||||
"plan",
|
||||
"tree",
|
||||
pid,
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
_ok(r, "plan tree")
|
||||
if r.returncode != 0:
|
||||
_fail(f"plan tree rc={r.returncode}\n{r.stderr}")
|
||||
tree = _load_json(r.stdout)
|
||||
if not isinstance(tree, list) or len(tree) != 1:
|
||||
_fail(f"expected 1 root node, got: {tree}")
|
||||
rn = tree[0]
|
||||
if rn.get("decision_id") != root.decision_id:
|
||||
_fail(f"root mismatch: {rn.get('decision_id')}")
|
||||
cn = rn.get("children")
|
||||
if not isinstance(cn, list) or len(cn) != 1:
|
||||
_fail(f"expected 1 strategy child: {cn}")
|
||||
if cn[0].get("decision_id") != strategy.decision_id:
|
||||
_fail(f"strategy mismatch: {cn[0].get('decision_id')}")
|
||||
gn = cn[0].get("children")
|
||||
if not isinstance(gn, list) or len(gn) != 1:
|
||||
_fail(f"expected 1 impl grandchild: {gn}")
|
||||
if gn[0].get("decision_id") != impl.decision_id:
|
||||
_fail(f"impl mismatch: {gn[0].get('decision_id')}")
|
||||
print("wf15-tree-ok")
|
||||
|
||||
|
||||
def wf15_explain_root_cause() -> None:
|
||||
"""Explain the root cause decision, verify context and rationale."""
|
||||
with _ws("wf15_explain_") as (ws, yp):
|
||||
assert yp is not None
|
||||
pid = _create_plan(ws, yp)
|
||||
db_url = os.environ["CLEVERAGENTS_DATABASE_URL"]
|
||||
dsvc = _get_dsvc(db_url)
|
||||
_, _, impl = _seed_decisions(dsvc, pid)
|
||||
r = run_cli(
|
||||
"plan",
|
||||
"explain",
|
||||
impl.decision_id,
|
||||
"--show-context",
|
||||
"--show-reasoning",
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
if r.returncode != 0:
|
||||
_fail(f"plan explain rc={r.returncode}\n{r.stderr}")
|
||||
d = _load_json(r.stdout)
|
||||
if not isinstance(d, dict):
|
||||
_fail(f"explain not dict: {d}")
|
||||
if d.get("decision_id") != impl.decision_id:
|
||||
_fail(f"id mismatch: {d.get('decision_id')}")
|
||||
if "pool" not in d.get("question", "").lower():
|
||||
_fail(f"question missing pool ref: {d.get('question')}")
|
||||
if not d.get("rationale"):
|
||||
_fail("rationale missing")
|
||||
snap = d.get("context_snapshot")
|
||||
if not isinstance(snap, dict) or not snap.get("hot_context_hash"):
|
||||
_fail(f"bad snapshot: {snap}")
|
||||
res_list = snap.get("relevant_resources")
|
||||
if not isinstance(res_list, list) or not res_list:
|
||||
_fail(f"no resources: {snap}")
|
||||
if not any("pool" in entry.get("path", "") for entry in res_list):
|
||||
_fail(f"no pool resource: {[e.get('path') for e in res_list]}")
|
||||
print("wf15-explain-ok")
|
||||
|
||||
|
||||
def wf15_rollback_to_checkpoint() -> None:
|
||||
"""Create checkpoint, rollback via CLI, verify response envelope."""
|
||||
with _ws("wf15_rollback_") as (ws, yp):
|
||||
assert yp is not None
|
||||
pid = _create_plan(ws, yp)
|
||||
sandbox_path, checkpoint_ref = _prepare_checkpoint_sandbox(ws)
|
||||
|
||||
# Create checkpoint via CheckpointService and rollback against a real git sandbox
|
||||
cp_svc = CheckpointService()
|
||||
cp_svc.register_sandbox(pid, sandbox_path)
|
||||
cp = cp_svc.create_checkpoint(
|
||||
pid,
|
||||
checkpoint_ref,
|
||||
reason="pre-apply",
|
||||
phase="execute",
|
||||
)
|
||||
result = cp_svc.rollback_to_checkpoint(pid, cp.checkpoint_id)
|
||||
if result.from_checkpoint_id != cp.checkpoint_id:
|
||||
_fail(f"checkpoint mismatch: {result.from_checkpoint_id}")
|
||||
if result.restored_files_count < 1:
|
||||
_fail(f"no files restored: {result.restored_files_count}")
|
||||
|
||||
# Validate rollback side effects on filesystem
|
||||
tracked = Path(sandbox_path) / "db_pool.conf"
|
||||
if tracked.read_text(encoding="utf-8") != "pool_size=20\n":
|
||||
_fail("tracked file was not restored to checkpoint content")
|
||||
if (Path(sandbox_path) / "scratch.tmp").exists():
|
||||
_fail("untracked file should be removed by rollback clean")
|
||||
|
||||
# Also verify plan status still works after rollback
|
||||
r = run_cli(
|
||||
"plan",
|
||||
"status",
|
||||
pid,
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
_ok(r, "plan status after rollback")
|
||||
print("wf15-rollback-ok")
|
||||
|
||||
|
||||
def wf15_correct_and_reapply() -> None:
|
||||
"""Correct root cause via plan correct --mode revert, verify response."""
|
||||
with _ws("wf15_correct_") as (ws, yp):
|
||||
assert yp is not None
|
||||
pid = _create_plan(ws, yp)
|
||||
db_url = os.environ["CLEVERAGENTS_DATABASE_URL"]
|
||||
dsvc = _get_dsvc(db_url)
|
||||
_, _, impl = _seed_decisions(dsvc, pid)
|
||||
r = run_cli(
|
||||
"plan",
|
||||
"correct",
|
||||
impl.decision_id,
|
||||
"--mode",
|
||||
"revert",
|
||||
"--guidance",
|
||||
"Keep pool_size at 20 and add connection health checks instead",
|
||||
"--plan",
|
||||
pid,
|
||||
"--yes",
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
_ok(r, "plan correct")
|
||||
if r.returncode == 0:
|
||||
cd = _load_json(r.stdout)
|
||||
if not isinstance(cd, dict):
|
||||
_fail(f"correct not dict: {cd}")
|
||||
if cd.get("status") != "applied":
|
||||
_fail(f"correct status mismatch: {cd}")
|
||||
rev = cd.get("reverted_decisions")
|
||||
if not isinstance(rev, list) or impl.decision_id not in rev:
|
||||
_fail(f"reverted mismatch: {cd}")
|
||||
# Verify plan is still accessible after correction
|
||||
r2 = run_cli(
|
||||
"plan",
|
||||
"status",
|
||||
pid,
|
||||
"--format",
|
||||
"json",
|
||||
workspace=ws,
|
||||
env_extra=_WIDE,
|
||||
)
|
||||
_ok(r2, "plan status after correct")
|
||||
if r2.returncode != 0:
|
||||
_fail(f"plan status rc={r2.returncode}\n{r2.stderr}")
|
||||
sd = _load_json(r2.stdout)
|
||||
if not isinstance(sd, dict) or sd.get("plan_id") != pid:
|
||||
_fail(f"status mismatch after correct: {sd}")
|
||||
print("wf15-correct-ok")
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Dispatcher
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
_COMMANDS: dict[str, Callable[[], None]] = {
|
||||
"plan-status-errored": wf15_plan_status_errored,
|
||||
"tree-and-root-cause": wf15_tree_and_root_cause,
|
||||
"explain-root-cause": wf15_explain_root_cause,
|
||||
"rollback-to-checkpoint": wf15_rollback_to_checkpoint,
|
||||
"correct-and-reapply": wf15_correct_and_reapply,
|
||||
}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
|
||||
print(f"Usage: {Path(__file__).name} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
reset_global_state()
|
||||
_COMMANDS[sys.argv[1]]()
|
||||
raise SystemExit(0)
|
||||
@@ -0,0 +1,76 @@
|
||||
*** Settings ***
|
||||
Documentation Integration test for Specification Workflow Example 15:
|
||||
... Disaster Recovery — Rollback a Failed Apply.
|
||||
...
|
||||
... Exercises post-apply failure detection via ``plan status``,
|
||||
... decision tree ROOT CAUSE investigation via ``plan tree``,
|
||||
... ``plan explain --show-context --show-reasoning``,
|
||||
... ``plan rollback --yes`` to checkpoint,
|
||||
... ``plan correct --mode revert --guidance`` to fix the root
|
||||
... cause decision, and re-apply using mocked LLM providers.
|
||||
Resource ${CURDIR}/common.resource
|
||||
Suite Setup Setup Test Environment
|
||||
Suite Teardown Cleanup Test Environment
|
||||
Force Tags wf15 integration disaster-recovery
|
||||
|
||||
*** Variables ***
|
||||
${HELPER} ${CURDIR}/helper_wf15_disaster_recovery.py
|
||||
|
||||
*** Test Cases ***
|
||||
WF15 Plan Status Shows Plan Lifecycle After Failed Apply
|
||||
[Documentation] Create a plan and verify ``plan status`` returns
|
||||
... lifecycle fields (plan_id, phase, processing_state).
|
||||
[Tags] plan_status lifecycle
|
||||
${result}= Run Process ${PYTHON} ${HELPER} plan-status-errored cwd=${WORKSPACE} timeout=120s on_timeout=kill
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} wf15-status-ok
|
||||
|
||||
WF15 Decision Tree Shows Root Cause Investigation
|
||||
[Documentation] Create a plan with seeded decisions and call
|
||||
... ``plan tree --format json`` to verify the full
|
||||
... three-level decision tree structure including the
|
||||
... root cause node.
|
||||
[Tags] plan_tree root_cause core_assertion
|
||||
${result}= Run Process ${PYTHON} ${HELPER} tree-and-root-cause cwd=${WORKSPACE} timeout=120s on_timeout=kill
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} wf15-tree-ok
|
||||
|
||||
WF15 Explain Root Cause Decision With Context And Reasoning
|
||||
[Documentation] Call ``plan explain <decision_id> --show-context
|
||||
... --show-reasoning --format json`` on the root cause
|
||||
... decision and verify rationale, context snapshot, and
|
||||
... relevant resource references are present.
|
||||
[Tags] plan_explain decision_investigation
|
||||
${result}= Run Process ${PYTHON} ${HELPER} explain-root-cause cwd=${WORKSPACE} timeout=120s on_timeout=kill
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} wf15-explain-ok
|
||||
|
||||
WF15 Rollback Plan To Pre Apply Checkpoint
|
||||
[Documentation] Create a checkpoint via CheckpointService, then call
|
||||
... ``plan rollback --yes <plan_id> <checkpoint_id>``
|
||||
... and verify the rollback response envelope contains
|
||||
... the correct plan_id and checkpoint_id.
|
||||
[Tags] plan_rollback checkpoint core_assertion
|
||||
${result}= Run Process ${PYTHON} ${HELPER} rollback-to-checkpoint cwd=${WORKSPACE} timeout=120s on_timeout=kill
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} wf15-rollback-ok
|
||||
|
||||
WF15 Correct Root Cause Decision And Reapply
|
||||
[Documentation] Call ``plan correct <decision_id> --mode revert
|
||||
... --guidance "..." --yes --format json`` on the root
|
||||
... cause decision and verify the correction response
|
||||
... contains reverted decisions.
|
||||
[Tags] plan_correct revert reapply
|
||||
${result}= Run Process ${PYTHON} ${HELPER} correct-and-reapply cwd=${WORKSPACE} timeout=120s on_timeout=kill
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} wf15-correct-ok
|
||||
Reference in New Issue
Block a user