src.asqi.output

Functions

parse_container_json_output(→ Dict[str, Any])	Extract JSON from container output with robust parsing.
extract_container_json_output_fields(...)	Parse and validate container output into a ContainerOutput object.
create_test_execution_progress(→ rich.progress.Progress)	Create a progress bar for test execution tracking.
format_execution_summary(→ tuple[str, str])	Format execution summary with appropriate styling.
format_failure_summary(→ None)	Display summary of failed tests.
create_workflow_summary(→ Dict[str, Any])	Create standardized workflow summary dictionary.
translate_report_paths(...)	Translate the test container report path to the host path for each report.
translate_dataset_paths(...)	Translate the container dataset path to the host path for each dataset.
display_score_card_reports(→ None)	Display information about all generated reports referenced in score card evaluations.
display_generated_datasets(→ None)	Display information about all generated datasets from test/generation job results.

Module Contents

src.asqi.output.parse_container_json_output(output: str) → Dict[str, Any]

Extract JSON from container output with robust parsing.

Args:

output: Raw container output string

Returns:

Parsed JSON dictionary

Raises:

ValueError: If no valid JSON found in output or output is empty

src.asqi.output.extract_container_json_output_fields(container_json_output: Dict[str, Any]) → asqi.response_schemas.ContainerOutput

Parse and validate container output into a ContainerOutput object.

Uses Pydantic schema validation for type safety. Supports both ‘results’ (recommended) and ‘test_results’ (legacy) field names, preferring ‘results’ if both are present.

Args:

container_json_output: Parsed JSON dictionary from container output

Returns:

ContainerOutput: Validated Pydantic object with type-safe access to:

• results (via get_results())

• generated_reports (List[GeneratedReport])

• generated_datasets (List[GeneratedDataset])

Notes:

• Validation warnings are logged but don’t break execution

• Backward compatible with flat container outputs (no structured fields)

• Falls back gracefully on validation errors, extracting what it can

src.asqi.output.create_test_execution_progress(console: rich.console.Console) → rich.progress.Progress

Create a progress bar for test execution tracking.

Args:

test_count: Total number of tests to execute console: Rich console instance

Returns:

Configured Progress instance

src.asqi.output.format_execution_summary(total_tests: int, successful_tests: int, failed_tests: int, execution_time: float) → tuple[str, str]

Format execution summary with appropriate styling.

Args:

total_tests: Total number of tests executed successful_tests: Number of successful tests failed_tests: Number of failed tests execution_time: Total execution time in seconds

Returns:

Tuple of (status_color, formatted_message)

src.asqi.output.format_failure_summary(failed_results: List, console: rich.console.Console, max_displayed: int = 3) → None

Display summary of failed tests.

Args:

failed_results: List of failed test results console: Rich console instance max_displayed: Maximum number of failures to display

src.asqi.output.create_workflow_summary(suite_name: str, workflow_id: str, status: str, total_tests: int, successful_tests: int, failed_tests: int, execution_time: float, **kwargs) → Dict[str, Any]

Create standardized workflow summary dictionary.

Args:

suite_name: Name of the test suite workflow_id: DBOS workflow ID status: Execution status total_tests: Total number of tests successful_tests: Number of successful tests failed_tests: Number of failed tests execution_time: Total execution time **kwargs: Additional summary fields

Returns:

Standardized summary dictionary

src.asqi.output.translate_report_paths(generated_reports: List[asqi.response_schemas.GeneratedReport], host_output_volume: str) → List[asqi.response_schemas.GeneratedReport]

Translate the test container report path to the host path for each report.

Args:

generated_reports: List of GeneratedReport Pydantic objects host_output_volume: String path to the host output volume

Returns:

List of GeneratedReport objects with translated paths

src.asqi.output.translate_dataset_paths(generated_datasets: List[asqi.response_schemas.GeneratedDataset], host_output_volume: str) → List[asqi.response_schemas.GeneratedDataset]

Translate the container dataset path to the host path for each dataset.

Args:

generated_datasets: List of GeneratedDataset Pydantic objects host_output_volume: String path to the host output volume

Returns:

List of GeneratedDataset objects with translated paths

src.asqi.output.display_score_card_reports(all_evaluations: List[Dict[str, Any]]) → None

Display information about all generated reports referenced in score card evaluations.

Args:

all_evaluations: List of score card evaluation results

src.asqi.output.display_generated_datasets(all_results: List[Dict[str, Any]]) → None

Display information about all generated datasets from test/generation job results.

Args:

all_results: List of test execution or generation job results