scripts.pipeline_quick_test

Reference › scripts › pipeline_quick_test

scripts/pipeline_quick_test.py is a 26-line smoke script 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 "Check code quality". It takes no arguments. It prints the overall status, one line per stage (status, duration, artifact count, and the first 200 characters of any error) and the number of logged messages, and exits 0 when the result is successful and 1 otherwise. A human runs it by hand; nothing imports it.

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/pipeline_quick_test.py

#!/usr/bin/env python3
"""Quick pipeline test — review work type."""

import asyncio
import sys

from orchestrata_ultimate.orchestrator.pipeline_runner import PipelineRunner


async def main():
    runner = PipelineRunner()
    print("Running review pipeline (QA → Cp)...")
    result = await runner.run(intent="Check code quality", work_type="review")
    print(f"Status: {result.overall_status.value}")
    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"Messages: {len(runner.get_message_log())}")
    return 0 if result.successful() else 1


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