From 118afa4298dc11afedd9cf0e6c5d3832002063d1 Mon Sep 17 00:00:00 2001 From: Forge Date: Thu, 30 Jul 2026 13:56:13 +0000 Subject: [PATCH] [AISOS-2311] Add forge smoke-test CLI command Detailed description: - Implemented an end-to-end smoke-test runner under that verifies diagnostic graph execution, workspace/git/podman setup, and container agent run. - Registered the async subcommand and options in . - Added unit tests covering successes, graph failures, missing container system dependencies, agent failures, and cleanup in . - Updated CLI testing documentation in . Closes: AISOS-2311 --- docs/dev/testing.md | 14 ++ src/forge/cli.py | 16 +++ src/forge/smoke_test.py | 247 ++++++++++++++++++++++++++++++++++ tests/unit/test_smoke_test.py | 201 +++++++++++++++++++++++++++ 4 files changed, 478 insertions(+) create mode 100644 src/forge/smoke_test.py create mode 100644 tests/unit/test_smoke_test.py diff --git a/docs/dev/testing.md b/docs/dev/testing.md index 8598a438..22de55df 100644 --- a/docs/dev/testing.md +++ b/docs/dev/testing.md @@ -97,3 +97,17 @@ uv run forge worker 2>&1 | grep "PROJ-123" ``` See the [Developer Guide](../developer-guide.md#10-debugging-tools) for patching checkpoints directly and other advanced debugging tools. + +## End-to-End Smoke Testing + +Forge provides an automated end-to-end smoke test CLI command to verify that state checkpointing, the container runtime sandbox, model provider API connectivity, and the Deep Agent system are fully functional. + +```bash +uv run forge smoke-test +``` + +This command runs completely offline from Jira/production branches, performing all actions in a disposable workspace: +1. **Diagnostic Graph Execution**: Compiles a 2-node LangGraph and executes it with the configured Redis state checkpointer, verifying workflow resumption. +2. **Workspace Setup & Container Verification**: Launches a rootless Podman environment inside a temporary folder with a mocked Git tree. +3. **Container Agent Execution & Verification**: Spawns the real agent inside the container, instructs it to create a specific verification file, and asserts that the file is successfully populated. +``` diff --git a/src/forge/cli.py b/src/forge/cli.py index ed75a31c..3cd4e31b 100644 --- a/src/forge/cli.py +++ b/src/forge/cli.py @@ -997,6 +997,15 @@ async def cmd_health(_args: argparse.Namespace) -> int: return 0 +async def cmd_smoke_test(_args: argparse.Namespace) -> int: + """Run an end-to-end smoke test to verify Forge runtime connectivity and execution.""" + from forge.config import get_settings + from forge.smoke_test import run_smoke_test + + settings = get_settings() + return await run_smoke_test(settings) + + def main(argv: list[str] | None = None) -> int: """Main CLI entry point.""" parser = argparse.ArgumentParser( @@ -1112,6 +1121,12 @@ def main(argv: list[str] | None = None) -> int: help="Number of log entries to show (default: 50)", ) + # smoke-test command + subparsers.add_parser( + "smoke-test", + help="Run an end-to-end smoke test to verify Forge runtime connectivity and execution", + ) + # skills subparser group skills_parser = subparsers.add_parser( "skills", @@ -1318,6 +1333,7 @@ def main(argv: list[str] | None = None) -> int: "list": cmd_list, "retry": cmd_retry, "logs": cmd_logs, + "smoke-test": cmd_smoke_test, "project-setup": cmd_project_setup, "get-config": cmd_get_config, } diff --git a/src/forge/smoke_test.py b/src/forge/smoke_test.py new file mode 100644 index 00000000..470b98d1 --- /dev/null +++ b/src/forge/smoke_test.py @@ -0,0 +1,247 @@ +"""Smoke test runner for verifying Forge core connectivity and execution.""" + +import contextlib +import logging +import operator +import shutil +import subprocess +import tempfile +import uuid +from pathlib import Path +from typing import Annotated, TypedDict + +from langgraph.graph import END, START, StateGraph + +from forge.config import Settings +from forge.orchestrator.checkpointer import clear_checkpoint, get_checkpointer +from forge.sandbox.runner import ContainerConfig, ContainerRunner + +logger = logging.getLogger(__name__) + + +class SmokeTestState(TypedDict, total=False): + """State for the minimal diagnostic LangGraph workflow.""" + + passed_nodes: Annotated[list[str], operator.add] + status: str + + +async def start_node(_state: SmokeTestState) -> dict: + """Diagnostic graph start node.""" + return {"passed_nodes": ["start"], "status": "running"} + + +async def end_node(_state: SmokeTestState) -> dict: + """Diagnostic graph end node.""" + return {"passed_nodes": ["end"], "status": "success"} + + +async def run_smoke_test(settings: Settings) -> int: + """Run an end-to-end smoke test to verify key Forge runtime capabilities. + + Performs the verification in three sequential stages: + 1. Diagnostic Graph Execution + 2. Workspace Setup & Container Verification + 3. Container Agent Execution & Verification + + Returns: + 0 if all stages passed, or a nonzero exit code on failure. + """ + stage1_status = "FAIL" + stage2_status = "PENDING" + stage3_status = "PENDING" + + stage1_err: Exception | None = None + stage2_err: Exception | None = None + stage3_err: Exception | None = None + + temp_dir: tempfile.TemporaryDirectory | None = None + thread_id = f"smoke-test-graph-{uuid.uuid4().hex[:8]}" + + try: + # --------------------------------------------------------------------- + # Stage 1: Diagnostic Graph Execution + # --------------------------------------------------------------------- + try: + logger.info("Starting Stage 1: Diagnostic Graph Execution") + checkpointer = await get_checkpointer() + + workflow = StateGraph(SmokeTestState) + workflow.add_node("start_node", start_node) + workflow.add_node("end_node", end_node) + workflow.add_edge(START, "start_node") + workflow.add_edge("start_node", "end_node") + workflow.add_edge("end_node", END) + + compiled_graph = workflow.compile(checkpointer=checkpointer) + + config = {"configurable": {"thread_id": thread_id}} + result = await compiled_graph.ainvoke( + {"passed_nodes": [], "status": "initialized"}, config=config + ) + + # Retrieve result and perform assertions + final_status = result.get("status") + final_passed_nodes = result.get("passed_nodes", []) + + if final_status != "success" or final_passed_nodes != ["start", "end"]: + raise RuntimeError( + f"Unexpected diagnostic graph outcome: status={final_status}, " + f"passed_nodes={final_passed_nodes}" + ) + + # Cleanup state checkpoint + await clear_checkpoint(thread_id) + stage1_status = "PASS" + logger.info("Stage 1: Diagnostic Graph Execution [PASS]") + + except Exception as e: + stage1_status = "FAIL" + stage1_err = e + logger.error(f"Stage 1 failed: {e}", exc_info=True) + with contextlib.suppress(Exception): + await clear_checkpoint(thread_id) + return 1 + + # --------------------------------------------------------------------- + # Stage 2: Workspace Setup & Container Verification + # --------------------------------------------------------------------- + try: + stage2_status = "FAIL" + logger.info("Starting Stage 2: Workspace Setup & Container Verification") + + if not shutil.which("podman"): + raise RuntimeError("podman not found in system PATH") + + runner = ContainerRunner(settings=settings) + + # Verify that the configured image exists + image_name = settings.container_image + if not await runner.image_exists(image_name): + raise RuntimeError( + f"Configured container image {image_name!r} does not exist locally" + ) + + # Create the disposable workspace directory + temp_dir = tempfile.TemporaryDirectory() + temp_dir_path = Path(temp_dir.name) + + # Initialize mock Git repository inside temporary workspace + subprocess.run(["git", "init"], cwd=temp_dir_path, check=True, capture_output=True) + subprocess.run( + ["git", "config", "user.name", "Forge Smoke Test"], + cwd=temp_dir_path, + check=True, + capture_output=True, + ) + subprocess.run( + ["git", "config", "user.email", "smoke-test@forge.sdlc"], + cwd=temp_dir_path, + check=True, + capture_output=True, + ) + + # Write initial README.md + readme_path = temp_dir_path / "README.md" + readme_path.write_text( + "# Forge Smoke Test Workspace\n\nThis is a temporary workspace for testing.", + encoding="utf-8", + ) + + # Commit the readme to mock a git repository + subprocess.run( + ["git", "add", "README.md"], cwd=temp_dir_path, check=True, capture_output=True + ) + subprocess.run( + ["git", "commit", "-m", "initial commit"], + cwd=temp_dir_path, + check=True, + capture_output=True, + ) + + stage2_status = "PASS" + logger.info("Stage 2: Workspace Setup & Container Verification [PASS]") + + except Exception as e: + stage2_status = "FAIL" + stage2_err = e + logger.error(f"Stage 2 failed: {e}", exc_info=True) + return 1 + + # --------------------------------------------------------------------- + # Stage 3: Container Agent Execution & Verification + # --------------------------------------------------------------------- + try: + stage3_status = "FAIL" + logger.info("Starting Stage 3: Container Agent Execution & Verification") + + # Configure container running config + container_config = ContainerConfig( + image=settings.container_image, + timeout_seconds=300, + skip_tests=True, + ) + + # Invoke the real container task runner in workspace + result = await runner.run( + workspace_path=temp_dir_path, + task_summary="Smoke Test File Creation", + task_description=( + "Create a text file in the workspace root named 'smoke_test_result.txt' " + "with content 'FORGE_SMOKE_TEST_PASSED'. Keep the change simple." + ), + config=container_config, + ) + + if not result.success: + error_msg = result.error_message or "Container task failed" + if result.stderr: + error_msg += f"\nStderr: {result.stderr}" + raise RuntimeError(f"Container runner execution failed: {error_msg}") + + # Verify that the expected file was created and contains the correct result + verification_file = temp_dir_path / "smoke_test_result.txt" + if not verification_file.exists(): + raise RuntimeError( + "Expected file 'smoke_test_result.txt' was not created in workspace" + ) + + file_content = verification_file.read_text(encoding="utf-8").strip() + if file_content != "FORGE_SMOKE_TEST_PASSED": + raise RuntimeError( + f"Result file content mismatch: expected 'FORGE_SMOKE_TEST_PASSED', " + f"got {file_content!r}" + ) + + stage3_status = "PASS" + logger.info("Stage 3: Container Agent Execution & Verification [PASS]") + + except Exception as e: + stage3_status = "FAIL" + stage3_err = e + logger.error(f"Stage 3 failed: {e}", exc_info=True) + return 1 + + return 0 + + finally: + # Cleanup temporary workspace directories + if temp_dir is not None: + try: + temp_dir.cleanup() + logger.info("Temporary workspace directory successfully cleaned up.") + except Exception as e: + logger.warning(f"Failed to cleanly remove temporary directory: {e}") + + # Print clean checklists report of the run + print("\n=================== Forge Smoke Test Report ===================") + print(f"[{stage1_status}] Stage 1: Diagnostic Graph Execution") + if stage1_err: + print(f" Error: {stage1_err}") + print(f"[{stage2_status}] Stage 2: Workspace Setup & Container Verification") + if stage2_err: + print(f" Error: {stage2_err}") + print(f"[{stage3_status}] Stage 3: Container Agent Execution & Verification") + if stage3_err: + print(f" Error: {stage3_err}") + print("===============================================================\n") diff --git a/tests/unit/test_smoke_test.py b/tests/unit/test_smoke_test.py new file mode 100644 index 00000000..4f989ed6 --- /dev/null +++ b/tests/unit/test_smoke_test.py @@ -0,0 +1,201 @@ +"""Unit tests for the Forge smoke test CLI command runner.""" + +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from forge.config import Settings +from forge.sandbox.runner import ContainerResult +from forge.smoke_test import run_smoke_test + + +@pytest.fixture +def mock_settings() -> Settings: + """Fixture for settings used in tests.""" + return Settings( + jira_base_url="https://mock-company.atlassian.net", + jira_api_token="mock-token", + jira_user_email="mock-email@company.com", + github_token="mock-token", + container_image="localhost/forge-dev:latest", + ) + + +@pytest.mark.asyncio +async def test_smoke_test_success(mock_settings: Settings) -> None: + """Test successful execution of all three stages of the smoke test.""" + # 1. Mock Stage 1: Diagnostic Graph Execution + mock_compiled_graph = MagicMock() + mock_compiled_graph.ainvoke = AsyncMock( + return_value={"passed_nodes": ["start", "end"], "status": "success"} + ) + + # 2. Mock Stage 2: Workspace Setup & Container Verification + async def mock_image_exists(_tag: str) -> bool: + return True + + # 3. Mock Stage 3: Container Agent Execution & Verification + async def mock_runner_run(workspace_path: Path, *_args, **_kwargs) -> ContainerResult: + # Simulate container writing the expected verification file + (workspace_path / "smoke_test_result.txt").write_text( + "FORGE_SMOKE_TEST_PASSED", encoding="utf-8" + ) + return ContainerResult(success=True, exit_code=0, stdout="Done", stderr="") + + with ( + patch("langgraph.graph.StateGraph.compile", return_value=mock_compiled_graph), + patch("forge.smoke_test.get_checkpointer", new_callable=AsyncMock), + patch("forge.smoke_test.clear_checkpoint", new_callable=AsyncMock) as mock_clear, + patch("shutil.which", return_value="/usr/bin/podman"), + patch("forge.sandbox.runner.ContainerRunner.image_exists", side_effect=mock_image_exists), + patch("forge.sandbox.runner.ContainerRunner.run", side_effect=mock_runner_run), + ): + exit_code = await run_smoke_test(mock_settings) + + assert exit_code == 0 + mock_clear.assert_called_once() + + +@pytest.mark.asyncio +async def test_smoke_test_graph_failure(mock_settings: Settings) -> None: + """Test that a failure in Stage 1 aborts early and returns a nonzero exit code.""" + mock_compiled_graph = MagicMock() + mock_compiled_graph.ainvoke = AsyncMock( + side_effect=Exception("Failed to invoke diagnostic graph") + ) + + with ( + patch("langgraph.graph.StateGraph.compile", return_value=mock_compiled_graph), + patch("forge.smoke_test.get_checkpointer", new_callable=AsyncMock), + patch("forge.smoke_test.clear_checkpoint", new_callable=AsyncMock) as mock_clear, + ): + exit_code = await run_smoke_test(mock_settings) + + assert exit_code == 1 + mock_clear.assert_called_once() # clear_checkpoint should be called in finally/except block + + +@pytest.mark.asyncio +async def test_smoke_test_missing_podman_or_image(mock_settings: Settings) -> None: + """Test that Stage 2 fails if podman is missing or the container image doesn't exist.""" + # Mock Stage 1 success + mock_compiled_graph = MagicMock() + mock_compiled_graph.ainvoke = AsyncMock( + return_value={"passed_nodes": ["start", "end"], "status": "success"} + ) + + # Case A: podman is missing + with ( + patch("langgraph.graph.StateGraph.compile", return_value=mock_compiled_graph), + patch("forge.smoke_test.get_checkpointer", new_callable=AsyncMock), + patch("forge.smoke_test.clear_checkpoint", new_callable=AsyncMock), + patch("shutil.which", return_value=None), + ): + exit_code = await run_smoke_test(mock_settings) + assert exit_code == 1 + + # Case B: container image does not exist + async def mock_image_exists_false(_tag: str) -> bool: + return False + + with ( + patch("langgraph.graph.StateGraph.compile", return_value=mock_compiled_graph), + patch("forge.smoke_test.get_checkpointer", new_callable=AsyncMock), + patch("forge.smoke_test.clear_checkpoint", new_callable=AsyncMock), + patch("shutil.which", return_value="/usr/bin/podman"), + patch( + "forge.sandbox.runner.ContainerRunner.image_exists", side_effect=mock_image_exists_false + ), + ): + exit_code = await run_smoke_test(mock_settings) + assert exit_code == 1 + + +@pytest.mark.asyncio +async def test_smoke_test_agent_failure(mock_settings: Settings) -> None: + """Test that Stage 3 fails if the container runner indicates task failure or omits the result.""" + # Mock Stage 1 success + mock_compiled_graph = MagicMock() + mock_compiled_graph.ainvoke = AsyncMock( + return_value={"passed_nodes": ["start", "end"], "status": "success"} + ) + + # Mock Stage 2 success + async def mock_image_exists(_tag: str) -> bool: + return True + + # Case A: container runner returns a failed ContainerResult + async def mock_runner_run_fail(_workspace_path: Path, *_args, **_kwargs) -> ContainerResult: + return ContainerResult(success=False, exit_code=1, stdout="", stderr="Compilation error") + + with ( + patch("langgraph.graph.StateGraph.compile", return_value=mock_compiled_graph), + patch("forge.smoke_test.get_checkpointer", new_callable=AsyncMock), + patch("forge.smoke_test.clear_checkpoint", new_callable=AsyncMock), + patch("shutil.which", return_value="/usr/bin/podman"), + patch("forge.sandbox.runner.ContainerRunner.image_exists", side_effect=mock_image_exists), + patch("forge.sandbox.runner.ContainerRunner.run", side_effect=mock_runner_run_fail), + ): + exit_code = await run_smoke_test(mock_settings) + assert exit_code == 1 + + # Case B: container runner succeeds but file is missing + async def mock_runner_run_no_file(_workspace_path: Path, *_args, **_kwargs) -> ContainerResult: + return ContainerResult(success=True, exit_code=0, stdout="Success", stderr="") + + with ( + patch("langgraph.graph.StateGraph.compile", return_value=mock_compiled_graph), + patch("forge.smoke_test.get_checkpointer", new_callable=AsyncMock), + patch("forge.smoke_test.clear_checkpoint", new_callable=AsyncMock), + patch("shutil.which", return_value="/usr/bin/podman"), + patch("forge.sandbox.runner.ContainerRunner.image_exists", side_effect=mock_image_exists), + patch("forge.sandbox.runner.ContainerRunner.run", side_effect=mock_runner_run_no_file), + ): + exit_code = await run_smoke_test(mock_settings) + assert exit_code == 1 + + +@pytest.mark.asyncio +async def test_smoke_test_timeout_and_cleanup(mock_settings: Settings) -> None: + """Test that temporary workspaces are fully cleaned up even if container runner raises TimeoutError.""" + # Mock Stage 1 success + mock_compiled_graph = MagicMock() + mock_compiled_graph.ainvoke = AsyncMock( + return_value={"passed_nodes": ["start", "end"], "status": "success"} + ) + + # Mock Stage 2 success + async def mock_image_exists(_tag: str) -> bool: + return True + + # Mock Stage 3 raising TimeoutError + async def mock_runner_run_timeout(*_args, **_kwargs) -> ContainerResult: + raise TimeoutError("Execution timed out") + + # Spy on TemporaryDirectory.cleanup using a custom subclass or patching + cleanup_called = False + from tempfile import TemporaryDirectory as RealTemporaryDirectory + + class SpyTemporaryDirectory: + def __init__(self, *_args, **_kwargs): + self._td = RealTemporaryDirectory(*_args, **_kwargs) + self.name = self._td.name + + def cleanup(self): + nonlocal cleanup_called + cleanup_called = True + self._td.cleanup() + + with ( + patch("langgraph.graph.StateGraph.compile", return_value=mock_compiled_graph), + patch("forge.smoke_test.get_checkpointer", new_callable=AsyncMock), + patch("forge.smoke_test.clear_checkpoint", new_callable=AsyncMock), + patch("shutil.which", return_value="/usr/bin/podman"), + patch("forge.sandbox.runner.ContainerRunner.image_exists", side_effect=mock_image_exists), + patch("tempfile.TemporaryDirectory", side_effect=SpyTemporaryDirectory), + patch("forge.sandbox.runner.ContainerRunner.run", side_effect=mock_runner_run_timeout), + ): + exit_code = await run_smoke_test(mock_settings) + assert exit_code == 1 + assert cleanup_called is True