cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

test(integration): workflow example 16 — devcontainer-driven development (supervised profile) #953

Closed
brent.edwards wants to merge 1 commits from test/int-wf16-devcontainer into master
pull from: test/int-wf16-devcontainer
merge into: cleveragents:master
Conversation 10 Commits 1 Files Changed 3
+458
3 changed files with 458 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+7
View File
@@ -2,6 +2,13 @@
## Unreleased
- Added Robot Framework integration test and Python helper for Specification
Workflow Example 16 — Devcontainer-Driven Development (supervised profile).
Covers devcontainer auto-discovery from `.devcontainer/devcontainer.json`,
lazy container activation on first execution, execution environment routing
to container workspace, and host apply via bind mount using mocked LLM
providers and mocked container operations.
(`robot/wf16_devcontainer.robot`, `robot/helper_wf16_devcontainer.py`) (#780)
- 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
robot/helper_wf16_devcontainer.py
+376
View File
@@ -0,0 +1,376 @@
"""Robot Framework helper for Workflow Example 16: Devcontainer-Driven Development.
Exercises devcontainer auto-discovery, lazy container activation,
execution environment routing, and host apply via bind mount using
mocked LLM providers and mocked container operations.
Each subcommand prints a sentinel on success and exits 0/1.
Usage:
python robot/helper_wf16_devcontainer.py <command>
"""
from __future__ import annotations
import json
import os
import re
import shutil
import subprocess
import sys
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
from typing import 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 ( # noqa: E402
cleanup_workspace,
init_bare_git_repo,
run_cli,
setup_workspace,
write_yaml,
)
from helpers_common import reset_global_state # noqa: E402
from cleveragents.domain.models.core.container_lifecycle import ( # noqa: E402
ContainerLifecycleState,
)
from cleveragents.resource.handlers.devcontainer import ( # noqa: E402
activate_container,
clear_lifecycle_registry,
get_lifecycle_tracker,
stop_container,
)
from cleveragents.resource.handlers.discovery import ( # noqa: E402
discover_devcontainers,
)
_DC_JSON: str = json.dumps(
{"name": "wf16-webapp", "image": "node:20", "workspaceFolder": "/workspaces/webapp"}
)
_ACTION_YAML: str = """\
name: local/add-dark-mode
description: Add dark mode toggle to the webapp
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: Dark mode CSS and toggle component implemented
"""
_ENV: dict[str, str] = {"COLUMNS": "500"}
# -- Mock runner (no real Docker) ------------------------------------------
@dataclass
class _MockResult:
returncode: int = 0
stdout: str = ""
stderr: str = ""
args: list[str] = field(default_factory=list)
class _MockRunner:
def __init__(self) -> None:
self.calls: list[tuple[list[str], dict[str, object]]] = []
self.up_result: _MockResult = _MockResult()
self.stop_result: _MockResult = _MockResult()
def __call__(self, args: list[str], **kwargs: object) -> _MockResult:
self.calls.append((list(args), dict(kwargs)))
if len(args) >= 2 and args[0] == "devcontainer" and args[1] == "up":
return self.up_result
if len(args) >= 2 and args[0] == "docker" and args[1] == "stop":
return self.stop_result
return _MockResult()
def set_up_result(
self,
*,
cid: str = "aabbccddee5566778899",
ws: str = "/workspaces/webapp",
) -> None:
self.up_result = _MockResult(
stdout=json.dumps(
{"outcome": "success", "containerId": cid, "remoteWorkspaceFolder": ws}
),
)
@property
def up_calls(self) -> list[tuple[list[str], dict[str, object]]]:
return [
(a, k)
for a, k in self.calls
if len(a) >= 2 and a[0] == "devcontainer" and a[1] == "up"
]
# -- Helpers ---------------------------------------------------------------
def _fail(msg: str) -> NoReturn:
print(f"FAIL: {msg}", file=sys.stderr)
raise SystemExit(1)
def _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 _make_repo() -> str:
"""Create git repo with .devcontainer/devcontainer.json."""
repo = init_bare_git_repo()
dc = Path(repo) / ".devcontainer"
dc.mkdir()
(dc / "devcontainer.json").write_text(_DC_JSON, encoding="utf-8")
subprocess.run(["git", "add", "."], cwd=repo, capture_output=True, check=True)
subprocess.run(
["git", "commit", "-m", "Add devcontainer"],
cwd=repo,
capture_output=True,
check=True,
)
return repo
def _cli(ws: str, *args: str) -> subprocess.CompletedProcess[str]:
"""Run CLI command, fail on non-zero exit."""
r = run_cli(*args, workspace=ws, env_extra=_ENV)
if r.returncode != 0:
_fail(f"{' '.join(args[:2])}: {r.stderr}")
return r
def _setup_full(ws: str, repo: str, yaml: str) -> str:
"""Register resource, project, action, plan use. Return plan_id."""
_cli(
ws,
"resource",
"add",
"git-checkout",
"local/webapp",
"--path",
repo,
"--branch",
"main",
)
_cli(ws, "project", "create", "local/webapp-project", "--resource", "local/webapp")
_cli(ws, "action", "create", "--config", yaml)
r = _cli(
ws,
"plan",
"use",
"local/add-dark-mode",
"local/webapp-project",
"--automation-profile",
"supervised",
"--format",
"plain",
)
pid = _plan_id(r.stdout)
if not pid:
_fail(f"no plan_id:\n{r.stdout}")
return pid
# -- Test commands ---------------------------------------------------------
def wf16_resource_auto_detection() -> None:
"""Step 1: resource add git-checkout, verify devcontainer discovery."""
ws = setup_workspace(prefix="wf16_detect_")
repo = _make_repo()
try:
_cli(
ws,
"resource",
"add",
"git-checkout",
"local/webapp",
"--path",
repo,
"--branch",
"main",
"--format",
"plain",
)
r = _cli(ws, "resource", "show", "local/webapp", "--format", "plain")
if "git-checkout" not in r.stdout:
_fail(f"not git-checkout:\n{r.stdout}")
results = discover_devcontainers(repo, "git-checkout")
if len(results) != 1:
_fail(f"expected 1 result, got {len(results)}")
if not str(results[0].config_path).endswith("devcontainer.json"):
_fail(f"bad config_path: {results[0].config_path}")
if results[0].config_data.get("name") != "wf16-webapp":
_fail(f"bad config name: {results[0].config_data.get('name')}")
print("wf16-detect-ok")
finally:
shutil.rmtree(repo, ignore_errors=True)
cleanup_workspace(ws)
def wf16_project_and_link() -> None:
"""Step 2: project create with --resource linking."""
ws = setup_workspace(prefix="wf16_proj_")
repo = _make_repo()
try:
_cli(
ws,
"resource",
"add",
"git-checkout",
"local/webapp",
"--path",
repo,
"--branch",
"main",
)
_cli(
ws,
"project",
"create",
"local/webapp-project",
"--description",
"WF16 webapp",
"--resource",
"local/webapp",
"--format",
"plain",
)
r = _cli(ws, "project", "show", "local/webapp-project", "--format", "plain")
if "local/webapp-project" not in r.stdout:
_fail(f"project missing:\n{r.stdout}")
print("wf16-project-ok")
finally:
shutil.rmtree(repo, ignore_errors=True)
cleanup_workspace(ws)
def wf16_plan_with_devcontainer() -> None:
"""Step 3: plan use with --automation-profile supervised."""
ws = setup_workspace(prefix="wf16_plan_")
repo = _make_repo()
yaml = write_yaml(_ACTION_YAML)
try:
pid = _setup_full(ws, repo, yaml)
r = _cli(ws, "plan", "status", pid, "--format", "plain")
if pid not in r.stdout:
_fail(f"plan_id missing:\n{r.stdout}")
print("wf16-plan-ok")
finally:
os.unlink(yaml)
shutil.rmtree(repo, ignore_errors=True)
cleanup_workspace(ws)
def wf16_execute_in_container() -> None:
"""Step 4: plan execute + domain-level lazy activation verification."""
ws = setup_workspace(prefix="wf16_exec_")
repo = _make_repo()
yaml = write_yaml(_ACTION_YAML)
try:
pid = _setup_full(ws, repo, yaml)
# CLI plan execute — controlled rejection (no LLM)
r = run_cli("plan", "execute", pid, workspace=ws, env_extra=_ENV)
out = r.stdout + r.stderr
if "INTERNAL" in out or "Traceback" in out:
_fail(f"plan execute crashed:\n{out}")
# Domain-level lazy activation with mock runner
clear_lifecycle_registry()
runner = _MockRunner()
runner.set_up_result(cid="aa16c0a1a1e200112233", ws="/workspaces/webapp")
rid = "01WF16TEST0000000000010"
tracker = activate_container(rid, repo, run_command=runner, session_id=pid)
if tracker.current_state != ContainerLifecycleState.RUNNING:
_fail(f"expected RUNNING: {tracker.current_state}")
if tracker.container_id != "aa16c0a1a1e200112233":
_fail(f"container_id: {tracker.container_id}")
if tracker.workspace_path != "/workspaces/webapp":
_fail(f"workspace_path: {tracker.workspace_path}")
if tracker.host_workspace_path != os.path.realpath(repo):
_fail(f"host_workspace_path: {tracker.host_workspace_path}")
if len(runner.up_calls) != 1:
_fail(f"up_calls: {len(runner.up_calls)}")
if "--workspace-folder" not in runner.up_calls[0][0]:
_fail(f"missing --workspace-folder: {runner.up_calls[0][0]}")
if get_lifecycle_tracker(rid).session_id != pid:
_fail(f"session_id: {get_lifecycle_tracker(rid).session_id}")
print("wf16-execute-ok")
finally:
clear_lifecycle_registry()
os.unlink(yaml)
shutil.rmtree(repo, ignore_errors=True)
cleanup_workspace(ws)
def wf16_apply_to_host() -> None:
"""Step 5: plan apply + domain-level bind-mount path verification."""
ws = setup_workspace(prefix="wf16_apply_")
repo = _make_repo()
yaml = write_yaml(_ACTION_YAML)
try:
pid = _setup_full(ws, repo, yaml)
# CLI lifecycle-apply — controlled rejection
r = run_cli("plan", "lifecycle-apply", pid, workspace=ws, env_extra=_ENV)
out = r.stdout + r.stderr
if "INTERNAL" in out or "Traceback" in out:
_fail(f"lifecycle-apply crashed:\n{out}")
# Domain-level bind-mount tracking
clear_lifecycle_registry()
runner = _MockRunner()
runner.set_up_result(cid="bb16a00112233445566ff", ws="/workspaces/webapp")
rid = "01WF16TEST0000000000020"
activate_container(rid, repo, run_command=runner, session_id=pid)
t = get_lifecycle_tracker(rid)
if t.host_workspace_path != os.path.realpath(repo):
_fail(f"host_workspace_path: {t.host_workspace_path}")
if t.workspace_path != "/workspaces/webapp":
_fail(f"workspace_path: {t.workspace_path}")
stop_container(rid, run_command=runner)
final = get_lifecycle_tracker(rid)
if final.current_state != ContainerLifecycleState.STOPPED:
_fail(f"expected STOPPED: {final.current_state}")
print("wf16-apply-ok")
finally:
clear_lifecycle_registry()
os.unlink(yaml)
shutil.rmtree(repo, ignore_errors=True)
cleanup_workspace(ws)
# -- Dispatcher ------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"wf16-resource-auto-detection": wf16_resource_auto_detection,
"wf16-project-and-link": wf16_project_and_link,
"wf16-plan-with-devcontainer": wf16_plan_with_devcontainer,
"wf16-execute-in-container": wf16_execute_in_container,
"wf16-apply-to-host": wf16_apply_to_host,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(f"Usage: helper_wf16_devcontainer.py <{'|'.join(_COMMANDS)}>")
return 1
cmd = sys.argv[1]
handler = _COMMANDS.get(cmd)
if handler is None:
print(f"Unknown command: {cmd}")
return 1
reset_global_state()
handler()
return 0
if __name__ == "__main__":
sys.exit(main())
robot/wf16_devcontainer.robot
+75
View File
@@ -0,0 +1,75 @@
*** Settings ***
Documentation Integration test for Specification Workflow Example 16:
... Devcontainer-Driven Development (supervised profile).
...
... Exercises devcontainer auto-discovery from
... ``.devcontainer/devcontainer.json``, lazy container
... activation on first execution, execution environment
... routing to container workspace, and host apply via
... bind mount using mocked LLM providers and mocked
... container operations.
Force Tags wf16 integration devcontainer v3.7.0 tdd_issue tdd_issue_780
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment With Database Isolation
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_wf16_devcontainer.py
*** Test Cases ***
WF16 Step 1 Resource Auto Detection
[Documentation] Register git-checkout resource with
... ``.devcontainer/`` directory and verify
... auto-detection of devcontainer-instance child
... resource via ``discover_devcontainers()``.
[Timeout] 120s
${result}= Run Process ${PYTHON} ${HELPER} wf16-resource-auto-detection 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} wf16-detect-ok
WF16 Step 2 Project And Link
[Documentation] Create project and link the git-checkout resource
... that contains a devcontainer configuration.
[Timeout] 120s
${result}= Run Process ${PYTHON} ${HELPER} wf16-project-and-link 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} wf16-project-ok
WF16 Step 3 Plan With Devcontainer
[Documentation] Create action and ``plan use`` with supervised
... automation profile targeting the devcontainer
... project. Verifies plan is created in strategize
... phase.
[Timeout] 120s
${result}= Run Process ${PYTHON} ${HELPER} wf16-plan-with-devcontainer 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} wf16-plan-ok
WF16 Step 4 Execute In Container
[Documentation] ``plan execute`` triggers devcontainer lazy build
... and routes tool invocations to the container
... workspace. Verifies lazy activation via mocked
... container operations and session ID propagation.
[Timeout] 120s
${result}= Run Process ${PYTHON} ${HELPER} wf16-execute-in-container 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} wf16-execute-ok
WF16 Step 5 Apply To Host
[Documentation] ``plan apply --yes`` writes changes back to host
... via bind mount. Verifies host_workspace_path
... tracking and container stop after apply.
[Timeout] 120s
${result}= Run Process ${PYTHON} ${HELPER} wf16-apply-to-host 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} wf16-apply-ok