src.asqi.workflow¶

Attributes¶

`oltp_endpoint`
`database_url`
`config`
`console`

Classes¶

TestExecutionResult

Represents the result of a single test execution.

Functions¶

`dbos_check_images_availability`(→ Dict[str, bool])	Check if all required Docker images are available locally.
`dbos_pull_images`(images)	Pull missing Docker images from registries.
`extract_manifest_from_image_step`(...)	Extract and parse manifest.yaml from a Docker image.
`validate_test_plan`(→ List[str])	DBOS step wrapper for comprehensive test plan validation.
`execute_single_test`(→ TestExecutionResult)	Execute a single test in a Docker container.
`evaluate_score_card`(→ List[Dict[str, Any]])	Evaluate score cards against test execution results.
`run_test_suite_workflow`(→ Dict[str, Any])	Execute a test suite with DBOS durability (tests only, no score card evaluation).
`convert_test_results_to_objects`(...)	Convert test results data back to TestExecutionResult objects.
`add_score_cards_to_results`(→ Dict[str, Any])	Add score card evaluation results to test results data.
`evaluate_score_cards_workflow`(→ Dict[str, Any])	Evaluate score cards against existing test results.
`run_end_to_end_workflow`(→ Dict[str, Any])	Execute a complete end-to-end workflow: test execution + score card evaluation.
`save_results_to_file_step`(→ None)	Save execution results to a JSON file.
`start_test_execution`(→ str)	Orchestrate test suite execution workflow.
`start_score_card_evaluation`(→ str)	Orchestrate score card evaluation workflow.

Module Contents¶

src.asqi.workflow.oltp_endpoint¶

src.asqi.workflow.database_url¶

src.asqi.workflow.config: dbos.DBOSConfig¶

src.asqi.workflow.console¶

class src.asqi.workflow.TestExecutionResult(test_name: str, sut_name: str, image: str)¶

Represents the result of a single test execution.

test_name¶

sut_name¶

image¶

start_time: float = 0¶

end_time: float = 0¶

success: bool = False¶

container_id: str = ''¶

exit_code: int = -1¶

container_output: str = ''¶

test_results: Dict[str, Any]¶

error_message: str = ''¶

property execution_time: float¶: Calculate execution time in seconds.

to_dict() → Dict[str, Any]¶: Convert to dictionary for storage/reporting.

src.asqi.workflow.dbos_check_images_availability(images: List[str]) → Dict[str, bool]¶: Check if all required Docker images are available locally.

src.asqi.workflow.dbos_pull_images(images: List[str])¶: Pull missing Docker images from registries.

src.asqi.workflow.extract_manifest_from_image_step(image: str) → asqi.schemas.Manifest | None¶: Extract and parse manifest.yaml from a Docker image.

src.asqi.workflow.validate_test_plan(suite: asqi.schemas.SuiteConfig, systems: asqi.schemas.SystemsConfig, manifests: Dict[str, asqi.schemas.Manifest]) → List[str]¶

DBOS step wrapper for comprehensive test plan validation.

Delegates to validation.py for the actual validation logic. This step exists to provide DBOS durability for validation results.

Args:: suite: Test suite configuration (pre-validated) systems: systems configuration (pre-validated) manifests: Available manifests (pre-validated)
Returns:: List of validation error messages

src.asqi.workflow.execute_single_test(test_name: str, image: str, sut_name: str, systems_params: Dict[str, Any], test_params: Dict[str, Any], container_config: asqi.config.ContainerConfig) → TestExecutionResult¶

Execute a single test in a Docker container.

Focuses solely on test execution. Input validation is handled separately in validation.py to follow single responsibility principle.

Args:: test_name: Name of the test to execute (pre-validated) image: Docker image to run (pre-validated) sut_name: Name of the system under test (pre-validated) systems_params: Dictionary containing system_under_test and other systems (pre-validated) test_params: Parameters for the test (pre-validated) container_config: Container execution configurations
Returns:: TestExecutionResult containing execution metadata and results
Raises:: ValueError: If inputs fail validation or JSON output cannot be parsed RuntimeError: If container execution fails

src.asqi.workflow.evaluate_score_card(test_results: List[TestExecutionResult], score_card_configs: List[Dict[str, Any]]) → List[Dict[str, Any]]¶: Evaluate score cards against test execution results.

src.asqi.workflow.run_test_suite_workflow(suite_config: Dict[str, Any], systems_config: Dict[str, Any], executor_config: Dict[str, Any], container_config: asqi.config.ContainerConfig) → Dict[str, Any]¶

Execute a test suite with DBOS durability (tests only, no score card evaluation).

This workflow: 1. Validates image availability and extracts manifests 2. Performs cross-validation of tests, systems, and manifests 3. Executes tests concurrently with progress tracking 4. Aggregates results with detailed error reporting

Args:: suite_config: Serialized SuiteConfig containing test definitions systems_config: Serialized SystemsConfig containing system configurations executor_config: Execution parameters controlling concurrency and reporting container_config: Container execution configurations
Returns:: Execution summary with metadata and individual test results (no score cards)

src.asqi.workflow.convert_test_results_to_objects(test_results_data: Dict[str, Any]) → List[TestExecutionResult]¶: Convert test results data back to TestExecutionResult objects.

src.asqi.workflow.add_score_cards_to_results(test_results_data: Dict[str, Any], score_card_evaluation: List[Dict[str, Any]]) → Dict[str, Any]¶: Add score card evaluation results to test results data.

src.asqi.workflow.evaluate_score_cards_workflow(test_results_data: Dict[str, Any], score_card_configs: List[Dict[str, Any]]) → Dict[str, Any]¶

Evaluate score cards against existing test results.

Args:: test_results_data: Test execution results containing ‘results’ field score_card_configs: List of score card configurations to evaluate
Returns:: Updated results with score card evaluation data

src.asqi.workflow.run_end_to_end_workflow(suite_config: Dict[str, Any], systems_config: Dict[str, Any], score_card_configs: List[Dict[str, Any]], executor_config: Dict[str, Any], container_config: asqi.config.ContainerConfig) → Dict[str, Any]¶

Execute a complete end-to-end workflow: test execution + score card evaluation.

Args:: suite_config: Serialized SuiteConfig containing test definitions systems_config: Serialized SystemsConfig containing system configurations score_card_configs: List of score card configurations to evaluate executor_config: Execution parameters controlling concurrency and reporting container_config: Container execution configurations
Returns:: Complete execution results with test results and score card evaluations

src.asqi.workflow.save_results_to_file_step(results: Dict[str, Any], output_path: str) → None¶: Save execution results to a JSON file.

src.asqi.workflow.start_test_execution(suite_path: str, systems_path: str, executor_config: Dict[str, Any], container_config: asqi.config.ContainerConfig, output_path: str | None = None, score_card_configs: List[Dict[str, Any]] | None = None, execution_mode: str = 'end_to_end', test_names: List[str] | None = None) → str¶

Orchestrate test suite execution workflow.

Handles input validation, configuration loading, and workflow delegation. Actual execution logic is handled by dedicated workflow functions.

Args:

suite_path: Path to test suite YAML file systems_path: Path to systems YAML file executor_config: Executor configuration dictionary. Expected keys:

“concurrent_tests”: int, number of concurrent tests

“max_failures”: int, max number of failures to display

“progress_interval”: int, interval for progress updates

container_config: Container execution configurations output_path: Optional path to save results JSON file score_card_configs: Optional list of score card configurations to evaluate execution_mode: “tests_only” or “end_to_end” test_names: Optional list of test names to filter from suite

Returns:

Workflow ID for tracking execution

Raises:

ValueError: If inputs are invalid FileNotFoundError: If configuration files don’t exist PermissionError: If configuration files cannot be read

src.asqi.workflow.start_score_card_evaluation(input_path: str, score_card_configs: List[Dict[str, Any]], output_path: str | None = None) → str¶

Orchestrate score card evaluation workflow.

Handles input validation, data loading, and workflow delegation. Actual evaluation logic is handled by dedicated workflow functions.

Args:: input_path: Path to JSON file containing test execution results score_card_configs: List of score card configurations to evaluate output_path: Optional path to save updated results JSON file
Returns:: Workflow ID for tracking execution
Raises:: ValueError: If inputs are invalid FileNotFoundError: If input file doesn’t exist json.JSONDecodeError: If input file contains invalid JSON PermissionError: If input file cannot be read