nat.test.utils

Functions

locate_repo_root(→ pathlib.Path)
locate_example_src_dir(→ pathlib.Path)	Locate the example src directory for an example's config class.
locate_example_dir(→ pathlib.Path)	Locate the example directory for an example's config class.
locate_example_config(→ pathlib.Path)	Locate the example config file for an example's config class, assumes the example contains a 'configs' directory
run_workflow(→ str)	Test specific wrapper for nat.utils.run_workflow to run a workflow with a question and validate the expected
serve_workflow(→ dict)	Execute a workflow using nat serve, and issue a POST request to the /generate endpoint with the given question.
build_nat_client(→ AsyncIterator[AsyncClient])	Build a NAT client for testing purposes.
validate_workflow_output(→ None)	Validate the contents of the workflow output file.

Module Contents

locate_repo_root() → pathlib.Path

locate_example_src_dir(example_config_class: type) → pathlib.Path

Locate the example src directory for an example’s config class.

locate_example_dir(example_config_class: type) → pathlib.Path

Locate the example directory for an example’s config class.

locate_example_config(

example_config_class: type,

config_file: str = 'config.yml',

assert_exists: bool = True,

) → pathlib.Path

Locate the example config file for an example’s config class, assumes the example contains a ‘configs’ directory

async run_workflow(

*,

config: Config | None = None,

config_file: StrPath | None = None,

question: str,

expected_answer: str | None = None,

assert_expected_answer: bool = True,

\*\*kwargs,

) → str

Test specific wrapper for nat.utils.run_workflow to run a workflow with a question and validate the expected answer. This variant always sets the result type to str.

async serve_workflow(

*,

config_path: pathlib.Path,

question: str,

expected_answer: str | None = None,

assert_expected_answer: bool = True,

port: int = 8000,

pipeline_timeout: int = 60,

request_timeout: int = 30,

) → dict

Execute a workflow using nat serve, and issue a POST request to the /generate endpoint with the given question.

Intended to be analogous to run_workflow but for the REST API serving mode.

async build_nat_client(

config: nat.data_models.config.Config,

worker_class: type[FastApiFrontEndPluginWorker] | None = None,

) → AsyncIterator[AsyncClient]

Build a NAT client for testing purposes.

Creates a test client with an ASGI transport for the specified configuration. The client is backed by a FastAPI application built from the provided worker class.

Args:

config: The NAT configuration to use for building the client. worker_class: Optional worker class to use. Defaults to FastApiFrontEndPluginWorker.

Yields:

An AsyncClient instance configured for testing.

validate_workflow_output(workflow_output_file: pathlib.Path) → None

Validate the contents of the workflow output file. WIP: output format should be published as a schema and this validation should be done against that schema.