scripts.test_pipeline_smoke

Reference › scripts › test_pipeline_smoke

scripts/test_pipeline_smoke.py is a 30-line smoke script, version 2, for the pipeline: it builds a PipelineRunner from orchestrata_ultimate.orchestrator.pipeline_runner and runs the review work type (QA then Cp) with the intent "Review the agent module code quality". It takes no arguments and is run by hand; despite its name it is a script, not a pytest module. It prints the overall status, the stage names, one line per stage (status, duration, artifact count and the first 200 characters of any error), the total artifact count and the message-log size, and exits 0 when the result is successful and 1 otherwise. It is the longer sibling of pipeline_quick_test.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/test_pipeline_smoke.py

#!/usr/bin/env python3
"""Smoke test v2: PipelineRunner with review work type (QA → Cp)."""

import asyncio
import sys

from orchestrata_ultimate.orchestrator.pipeline_runner import PipelineRunner


async def main():
    runner = PipelineRunner()
    print("Running review pipeline...")
    result = await runner.run(
        intent="Review the agent module code quality", work_type="review"
    )
    print(f"Status: {result.overall_status.value}")
    print(f"Stages: {list(result.stages.keys())}")
    for alias, sr in result.stages.items():
        print(
            f"  {alias}: {sr.status.value} ({sr.duration_ms:.0f}ms, {len(sr.artifacts)} artifacts)"
        )
        if sr.error:
            print(f"    ERROR: {sr.error[:200]}")
    print(f"Total artifacts: {len(result.artifacts)}")
    print(f"Message log entries: {len(runner.get_message_log())}")
    return 0 if result.successful() else 1


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))