Orcraft Test Flow

A flow is an ordered sequence of steps with a shared result. It exists so a multi-step procedure can be named, reused across tests, and reported as a unit rather than as a run of unrelated actions.

from __future__ import annotations

from orcraft import (
	OrcraftTestFlow,
	OrcraftTestFlowConfig,
	OrcraftTestFlowResult,
	OrcraftConfigField,
	OrcraftResultField,
)


class MyFlowConfig(OrcraftTestFlowConfig):
	name = OrcraftConfigField.Text(default="My Flow")
	app_folder = OrcraftConfigField.PackageFolder(default="", label="App Folder")
	payload_size = OrcraftConfigField.Numeric(default=64, label="Payload Size")


class MyFlowResult(OrcraftTestFlowResult):
	output_file = OrcraftResultField.Text(default="", label="Output File")
	measured_value = OrcraftResultField.Numeric(default=0, label="Measured Value")


class MyFlow(OrcraftTestFlow):
	result_type = MyFlowResult

	def __init__(self, config: MyFlowConfig) -> None:
		super().__init__(config)

	def _run_flow(self) -> None:
		first_result = self.run_step(
			MyFirstStep,
			app_folder=self.config.app_folder,
		)

		if self.handle_errors(first_result, self.result):
			return

		second_result = self.run_step(
			MySecondStep,
			input_file=first_result.output_file,
			payload_size=self.config.payload_size,
		)

		if self.handle_errors(second_result, self.result):
			return

		self.result.output_file = first_result.output_file
		self.result.measured_value = second_result.measured_value

		self.checkpoint(
			second_result.measured_value > 0,
			f"Measured {second_result.measured_value}.",
			"The system under test reported no measurement.",
		)

Result class declaration

A test or a step builds its result inside the body. A flow names its result class up front instead, through the result_type class attribute, as MyFlow does above. The framework instantiates that class when the flow starts, so self.result exists and is ready to be filled in from the first line of _run_flow - which is also why _run_flow returns nothing: it populates self.result as it goes, and the caller reads that object back from flow.run().

The attribute is optional. Omit it and the flow reports a plain OrcraftTestFlowResult - status, descriptions and timing, with no fields of its own - which is all a flow needs when it only sequences steps and has nothing of its own to measure.

Error reporting

When a flow body raises, it is the framework that finishes the result. It sets the status to ERROR, records the error description, stamps the finish and elapsed times, and emits the result the controller renders. All of that lands on whatever self.result is at that moment. Building the declared class before the body runs guarantees that an early failure still reports through the flow's own result class: the log table and the controller show the flow's declared fields alongside the error, and its result columns do not vanish from the run.

The alternative is to do what a step does and assign the result inside the body:

def _run_flow(self) -> None:
	self.result = MyFlowResult()
	...

On the happy path the two are identical - same class, same values, same result columns. They differ only when the body raises before that assignment line runs. The framework then finishes the base OrcraftTestFlowResult it created on its own: status and error text are still correct, but the flow's declared fields are absent and its result columns are lost for that run.

Put plainly, result_type turns "the result always carries the flow's declared shape" into a framework guarantee. Assigning in the body turns it into an authoring rule - make the assignment the first statement - that costs nothing when followed and quietly degrades error output when it is not.

Config fields

A flow config adds nothing of its own to the base set - its inputs are whatever fields you declare, such as app_folder and payload_size on MyFlowConfig above. What it starts with:

Overridable in code

Set in your config class, fixed for the run.

Reserved

Written by the framework while a run is in progress.

Result fields

Whether the framework builds it from result_type or the body assigns it, a flow result starts with fields it did not declare. Beyond anything you add, OrcraftTestFlowResult carries:

overridable by user (optional)  ·  written by the framework, may still be set directly  ·  reserved

Running steps

run_step takes a step class, keyword values for that step's config fields, and optionally the endpoint to run it on:

result = self.run_step(
	MySecondStep,
	endpoint=self.config.endpoint_2,
	input_file=first_result.output_file,
)

Passing a keyword the step config does not declare raises immediately, naming the config and the field. A renamed field is caught at the call site rather than silently ignored, which is the failure mode a dictionary of parameters would have given you instead.

Failure handling

Steps in a flow are ordered because each one depends on the last. Two mechanisms keep a flow from running past a broken precondition.

The first is explicit: handle_errors after each step, copying the step result into the flow result and returning True when the flow should stop.

step_result = self.run_step(MyFirstStep, app_folder=self.config.app_folder)

if self.handle_errors(step_result, self.result):
	return

The second is automatic: once the flow result is in ERROR, further run_step calls short-circuit instead of executing. A flow that loses its first step does not go on to run the rest against a state that never got set up.

Note that this applies to ERROR, not to FAIL. A failing checkpoint is a finding, and a flow keeps going so it can report the rest of them; an error means the flow cannot meaningfully continue.

Flows in a case

A case body builds a flow, runs it, and reads its result:

def _run_case(self, case: OrcraftTestCase) -> OrcraftTestResult:
	case.result = MyTestResult()

	flow = MyFlow(
		MyFlowConfig(
			app_folder=self.config.app_folder,
			payload_size=case.params["payload_size"],
		)
	)
	flow_result = flow.run()

	if self.handle_errors(flow_result, case.result):
		return case.result

	case.result.measured_value = flow_result.measured_value
	return case.result

Values move down through the flow config and back up through the flow result. A case that runs three flows reads as three named phases, and the log nests accordingly - test, case, flow, step - so a failure is located rather than searched for.

Endpoint helpers

Flows also carry the helpers for driving endpoints, since the flow is the level that knows which endpoint each step is going to:

All three are covered in Orcraft Endpoints.