Uranium Python API

Request fixtures by name in the arguments of a test_* function. Do not import and call fixture functions yourself. Import returned types and assertion helpers from tools.uranium.services.

def test_traffic_server_starts(ats: ATS, curl: Curl) -> None:
    """Verify that Traffic Server starts with its default test configuration.

    :param ats: Unstarted proxy with fixture-owned teardown.
    :param curl: Client honoring the selected test transport.
    """
    ats.start()
    result = curl.get(ats)
    assert result.returncode == 0, result.output

Prefer direct replay tests when Proxy Verifier can express the scenario. Native tests use ordinary pytest functions for custom clients or runtime changes. Share setup through helper functions or local @pytest.fixture functions; a scenario class and a separate run() entry point are not required.

Fixtures

All fixtures are function-scoped except uranium_test_runtime, which is session-scoped. Configuration is staged until start(): requesting a fixture does not start ATS or a support process.

Fixture	Returned value	Responsibility
ats	ATS	One unstarted ATS instance named ats.
ats_factory	ATSFactory	Create multiple named ATS instances in one test.
services	ServiceFactory	Create origins, DNS, verifier servers, and custom processes.
curl	Curl	Run curl respecting the selected TCP/UDS transport.
procedural_context	ProceduralContext	Own the test’s paths, runtime, and sandbox.
uranium_replay	Callable[[Path], None]	Execute a replay manifest and all variants inside a native test.
uranium_test_runtime	TestRuntime	Share installed-tool paths, capabilities, and port allocation.

ats depends on ats_factory. Both factories and curl share the test’s procedural_context, so they can safely be requested together. Names passed to each factory must be unique within a test. Pass options such as enable_tls=True or enable_cache=True to ats_factory.create().

Common ATS creation options are:

Option	Default	Effect
enable_cache	True	Enable the isolated HTTP cache.
enable_tls	False	Add a TLS listener and default test certificate.
enable_quic	False	Add an HTTP/3 listener; normally used with TLS enabled.
enable_proxy_protocol	False	Add dedicated PROXY protocol listeners.
disable_log_rolling	True	Keep logs at predictable paths. Set false to test installed defaults; an explicit records.update() value takes precedence.
server_args	[]	Additional argument strings passed to traffic_server.

Configure records with ats.records.update({"proxy.config.http.cache.http": 0}) and line-oriented files with ats.remap_config.add_line(...). Use the set_*_yaml() methods for structured configuration and write_config_file() for other files. Modify live files explicitly before requesting a reload; changing a staged object alone does not reload a running server.

Start origins and DNS before ATS when the scenario requires them. Factories stop their processes during teardown, even after an assertion fails. Unexpected ATS exits, missing diagnostic output, and FATAL:, ERROR: or Unrecognized configuration value lines in diags.log fail teardown. ATS.allow_errors(expression) accepts the ERROR: lines a test produces deliberately, and disable_log_checks=True accepts every error and unrecognized value. Use ATS.wait() for intentional one-shot exits and ATS.expect_start_failure() for an expected startup failure, which skips the diagnostic checks.

A service’s run() waits for a one-shot command; start() launches a long-lived process. Check command results explicitly, or register stream expectations before running the service. Negative log assertions should inspect complete output after stopping the process, not an early positive-marker snapshot.

uranium_replay(path) accepts a pathlib.Path to a manifest. Each call and variant uses a distinct child of the native test’s sandbox, preserving sibling services and earlier results. Replay-only scenarios should normally use direct *.test.yaml collection.

Paths, cleanup, and parallel execution

Sandboxes use <sandbox-root>/<test-name>/. A rerun clears the previous directory for that test. Failures retain artifacts; passing tests retain them only with --keep-sandboxes. A session lock prevents concurrent invocations from sharing a root; give simultaneous runs different --sandbox paths.

Use services.allocate_port() for custom listeners instead of fixed ports. Ownership and allocation work across pytest-xdist workers. Tests that cannot overlap others are listed in tests/serial_tests.txt.

Copy helpers resolve relative paths against the test’s source directory. Use procedural_context.run_directory for generated inputs. ATS exposes config_directory, log_directory, and storage_directory. Unix-socket paths can live outside the sandbox to satisfy OS path limits.

Clients and expectations

curl.get(ats, ...) and curl.run_for(ats, arguments, ...) honor --curl-uds. run_for accepts one shell-style argument string, parsed without executing a shell. Use curl.run when choosing the endpoint and transport yourself. ATS.run takes an argument vector; ATS.run_shell explicitly executes a shell script. Timeout values are in seconds.

CommandResult exposes returncode, stdout, stderr, and combined output. Managed service streams support contains(), excludes(), and matches_gold(). Register these before running the service; do not use += or replace the stream expectation object.

API reference

These signatures and docstrings are generated from the implementation by Sphinx autodoc. No ATS installation or running services are needed to build the reference.

Traffic Server and configuration

class tools.uranium.services.ATSFactory(context)

Create Traffic Server instances owned by one pytest scenario.

Parameters:

context (ProceduralContext)

close()

Stop and validate every created instance in reverse order.

Return type:

None

create(name=None, **process_options)

Create a named Traffic Server with fixture-owned cleanup.

Parameters:

• name (str | None) – Unique process name, or None for an automatic name.

• process_options (Any) – Replay-runner options controlling ATS startup.

Return type:

ATS

has_feature(feature)

Return whether ATS was built with a feature.

Parameters:

feature (str) – Build-feature name to query.

Return type:

bool

property run_directory: pathlib.Path

Return the test’s isolated sandbox directory.

tls_library()

Return the version string of the TLS library ATS was built with.

Test clients such as openssl and Python’s ssl module may use a different library, e.g. system OpenSSL against a BoringSSL ATS.

Return type:

str

class tools.uranium.services.ATS(context, name='ats', **process_options)

A pytest-owned Traffic Server process with staged configuration.

Parameters:

• context (ProceduralContext)

• name (str)

• process_options (Any)

add_default_ssl_files()

Enable TLS using Uranium’s default certificate and key.

Return type:

None

allow_errors(expression)

Accept diags.log ERROR: lines matching a regular expression.

Errors that do not match any allowed expression still fail the test.

Parameters:

expression (str) – Regular expression searched for in each error line.

Return type:

None

allow_private_connect(methods=('CONNECT',))

Allow outbound test traffic to loopback addresses.

Parameters:

methods (Sequence[str]) – HTTP methods allowed to connect to loopback origins.

Return type:

None

append_records_document(values)

Append another records.yaml document after the initial configuration.

Parameters:

values (Mapping[str, object]) – Flat record names and values for the appended document.

Return type:

None

chown_for_ats(path)

Give files a test created to the user Traffic Server runs as.

This does nothing unless the test runs as root, in which case Traffic Server switches to an unprivileged user that cannot read root-owned 0600 files such as generated keys.

Parameters:

path (pathlib.Path) – Directory whose contents are handed to the ATS user.

Return type:

None

close()

Stop Traffic Server and validate its diagnostics.

Once expect_start_failure() is called, no diagnostics are checked, including whether diags.log exists. Otherwise fatal diagnostics fail the test, as do errors not accepted by allow_errors() and unrecognized configuration values, unless the instance was created with disable_log_checks=True.

Return type:

None

property config_directory: pathlib.Path

Return the ATS configuration directory.

copy_custom_plugin(path)

Copy a test plugin into ATS’s plugin directory on start.

Parameters:

path (str | Path) – Test plugin path to copy.

Return type:

None

copy_to_config(*paths)

Copy source files into ATS’s configuration directory on start.

Parameters:

paths (str | Path) – Source paths copied when ATS starts.

Return type:

None

copy_to_ssl(*paths)

Copy certificate material into ATS’s SSL directory on start.

Parameters:

paths (str | Path) – Certificate or key paths copied when ATS starts.

Return type:

None

property diags_log: pathlib.Path

Return the configured diagnostic file or captured diagnostic stream.

drain_and_stop(delay=1.0)

Drain Traffic Server and stop it cleanly.

Parameters:

delay (float) – Number of seconds between draining and stopping.

Return type:

None

property environment: dict[str, str]

Return the environment used by ATS and its tools.

property error_log: pathlib.Path

Return the path to error.log.

expect_start_failure(expression, return_code=33)

Expect startup to fail with a matching diagnostic.

Parameters:

• expression (str) – Regular expression required in startup output.

• return_code (int | Sequence[int]) – Accepted process return code or codes.

Return type:

None

has_feature(feature)

Return whether ATS was built with a feature.

Parameters:

feature (str) – Build-feature name to query.

Return type:

bool

property http_port: int

Return the selected IPv4 HTTP port.

property https_port: int

Return the selected IPv4 TLS port.

property ipv6_https_port: int

Return a port reserved for an explicit IPv6 TLS listener.

property ipv6_port: int

Return a port reserved for an explicit IPv6 listener.

property is_running: bool

Return whether Traffic Server is running.

kill()

Kill Traffic Server without running its shutdown handlers.

Return type:

None

property log_directory: pathlib.Path

Return the ATS log directory.

property name: str

Return this Traffic Server instance’s process name.

omit_config_file(filename)

Leave a file absent from the materialized ATS configuration.

Parameters:

filename (str) – Configuration filename to omit.

Return type:

None

plugin_exists(name)

Return whether an installed ATS plugin is available.

Parameters:

name (str) – Plugin basename, with or without the .so suffix.

Return type:

bool

property process_output: str

Return stdout and stderr captured outside ATS’s bound traffic.out.

property proxy_protocol_https_port: int

Return the selected TLS Proxy Protocol port.

property proxy_protocol_port: int

Return the selected clear-text Proxy Protocol port.

rpc(request, timeout=30)

Send one JSON-RPC request through traffic_ctl.

Parameters:

• request (object) – JSON string or JSON-serializable request value.

• timeout (float) – Maximum number of seconds to wait for completion.

Return type:

CommandResult

run(*arguments, timeout=30)

Run a command with this ATS instance’s environment.

Parameters:

• arguments (str | Path) – Executable followed by its command-line arguments.

• timeout (float) – Maximum number of seconds to wait for completion.

Return type:

CommandResult

property run_directory: pathlib.Path

Return this process’s isolated run directory.

run_shell(script, timeout=30)

Run a Bash script with this ATS instance’s environment.

Parameters:

• script (str) – Bash source to execute.

• timeout (float) – Maximum number of seconds to wait for completion.

Return type:

CommandResult

property runtime_directory: pathlib.Path

Return the ATS local-state directory.

send_signal(signal_number)

Send a signal to the Traffic Server process.

Parameters:

signal_number (int) – Operating-system signal number to send.

Return type:

None

set_environment(name, value)

Set one environment variable for Traffic Server and its tools.

Parameters:

• name (str) – Environment variable name.

• value (str) – Environment variable value.

Return type:

None

set_logging_yaml(document)

Write a structured logging.yaml configuration when ATS starts.

Parameters:

document (Mapping[str, object]) – Parsed logging.yaml document to serialize.

Return type:

None

set_ssl_multicert_yaml(document)

Write structured TLS certificate configuration without defaults.

Parameters:

document (Mapping[str, object]) – Parsed ssl_multicert.yaml document to serialize.

Return type:

None

set_startup_timeout(timeout)

Set the maximum time allowed for ATS startup.

Parameters:

timeout (float) – Startup timeout in seconds.

Return type:

None

property ssl_directory: pathlib.Path

Return the ATS TLS material directory.

start()

Materialize configuration and wait until Traffic Server is ready.

Return type:

None

stop()

Stop Traffic Server without relinquishing fixture ownership.

Return type:

None

property storage_directory: pathlib.Path

Return the ATS storage directory.

tls_library()

Return the version string of the TLS library ATS was built with.

Test clients such as openssl and Python’s ssl module may use a different library, e.g. system OpenSSL against a BoringSSL ATS.

Return type:

str

traffic_ctl(*arguments, timeout=30)

Run traffic_ctl against this ATS run root.

Parameters:

• arguments (str) – Arguments passed to traffic_ctl.

• timeout (float) – Maximum number of seconds to wait for completion.

Return type:

CommandResult

property traffic_out: pathlib.Path

Return the path to traffic.out.

property uds_path: str

Return this instance’s Unix-domain listener path.

unset_environment(*names)

Remove environment variables before ATS starts.

Parameters:

names (str) – Inherited or harness-provided variable names to remove.

Return type:

None

wait(timeout=60)

Wait for Traffic Server to exit and validate its return code.

Parameters:

timeout (float) – Maximum number of seconds to wait.

Return type:

None

write_body_factory_file(relative_path, content)

Materialize a custom body-factory template on start.

Parameters:

• relative_path (str) – Destination path below the body-factory root.

• content (str) – Complete template contents.

Return type:

None

write_config_file(filename, content)

Materialize a test-owned ATS configuration file on start.

Parameters:

• filename (str) – Destination filename under the config directory.

• content (str) – Complete file contents.

Return type:

None

write_runtime_file(filename, content)

Materialize a test-owned ATS local-state file on start.

Parameters:

• filename (str) – Destination filename under the runtime directory.

• content (str) – Complete file contents.

Return type:

None

class tools.uranium.services.RecordsConfig(values)

Stage records.yaml updates until ATS starts.

Parameters:

values (dict[str, object])

update(values)

Merge flat ATS record names into the staged values.

Parameters:

values (Mapping[str, object]) – Record names and values to merge.

Return type:

None

class tools.uranium.services.ConfigFile(lines, path)

Stage one line-oriented ATS configuration file.

Parameters:

• lines (list[str])

• path (Path)

add_line(line)

Append one line to the staged file.

Parameters:

line (object) – Value converted to one configuration line.

Return type:

None

add_lines(lines)

Append lines from a string or sequence.

Parameters:

lines (Sequence[object] | str) – Newline-delimited text or values for separate lines.

Return type:

None

property path: pathlib.Path

Return the configuration file’s materialized path.

Clients and support services

Origin, DNS, HTTP-bin, and verifier services share the lifecycle and output expectation methods documented on ProcessService below. services.origin accepts microserver options such as ssl, both, lookup_key, and delay (seconds). services.dns accepts port and a default response. Use services.process for a custom executable, with ready_port when TCP listener readiness is sufficient; otherwise wait for a protocol response or a specific output marker before sending test traffic.

class tools.uranium.services.Curl(working_directory, *, use_uds=False)

Run curl requests and return ordinary Python result objects.

Parameters:

• working_directory (Path)

• use_uds (bool)

static accepts_seclevel()

Return whether curl’s TLS library accepts an @SECLEVEL directive in --ciphers.

OpenSSL 1.1.0 and later, and its quictls fork, accept the directive. BoringSSL rejects the whole cipher list, so curl fails before the TLS handshake. A curl built with several TLS backends lists the inactive ones in parentheses, so only the active backend counts.

Return type:

bool

get(ats, path='/', *, headers=None, options='', timeout=30)

Send a GET request to an ATS instance.

Parameters:

• ats (ATS) – ATS instance that receives the request.

• path (str) – Request path, with or without a leading slash.

• headers (Mapping[str, str] | None) – HTTP headers to add to the request.

• options (str) – Shell-style curl arguments. Quoting is parsed with shlex.split(); no shell is invoked.

• timeout (float) – Maximum number of seconds to wait for curl.

Return type:

CommandResult

run(arguments, timeout=30)

Run curl without applying an ATS transport.

Parameters:

• arguments (str) – Shell-style curl arguments. Quoting is parsed with shlex.split(); no shell is invoked.

• timeout (float) – Maximum number of seconds to wait for curl.

Return type:

CommandResult

run_for(ats, arguments, timeout=30)

Run curl arguments using the selected ATS transport.

Parameters:

• ats (ATS) – ATS instance whose Unix socket is used when UDS mode is enabled.

• arguments (str) – Shell-style curl arguments. Quoting is parsed with shlex.split(); no shell is invoked.

• timeout (float) – Maximum number of seconds to wait for curl.

Return type:

CommandResult

run_script(ats, script, timeout=30)

Run a Bash script containing curl command placeholders.

Parameters:

• ats (ATS) – ATS instance whose Unix socket is substituted for {curl} when UDS mode is enabled.

• script (str) – Bash script in which {curl} is the transport-aware curl command and {curl_base} is plain curl.

• timeout (float) – Maximum number of seconds to wait for the script.

Return type:

CommandResult

static supports(feature)

Return whether the installed curl advertises a feature.

Parameters:

feature (str) – Case-insensitive feature name to find in curl --version output.

Return type:

bool

class tools.uranium.services.ServiceFactory(context)

Create pytest-owned support processes for procedural tests.

Parameters:

context (ProceduralContext)

allocate_port(socket_type=SocketKind.SOCK_STREAM)

Reserve a listener port for a bespoke support process.

Parameters:

socket_type (int) – Socket type, normally SOCK_STREAM or SOCK_DGRAM.

Return type:

int

dns(name, **options)

Create a microDNS server.

Parameters:

• name (str) – Unique support-process name within the scenario.

• options (Any) – Listener port and default DNS response options.

Return type:

DNSServer

httpbin(name, **options)

Create a go-httpbin server.

Parameters:

• name (str) – Unique support-process name within the scenario.

• options (Any) – Listener address, port, and command options.

Return type:

HttpBinServer

origin(name, **options)

Create a microserver origin.

Parameters:

• name (str) – Unique support-process name within the scenario.

• options (Any) – Microserver listener, TLS, delay, and command options.

Return type:

OriginServer

process(name, command, *, environment=None, ready_port=0, ready_address='127.0.0.1')

Create an arbitrary managed process for a bespoke scenario.

Parameters:

• name (str) – Unique support-process name within the scenario.

• command (Sequence[str | Path]) – Executable followed by its command-line arguments.

• environment (dict[str, str] | None) – Complete or augmented process environment.

• ready_port (int) – TCP port that must accept connections after start, or zero to skip listener readiness checks.

• ready_address (str) – Address used for listener readiness checks.

Return type:

ProcessService

proxy_verifier_at_least(version)

Return whether Proxy Verifier meets a minimum version.

Parameters:

version (str) – Minimum dotted version string.

Return type:

bool

resolve_path(value)

Resolve a source path, placeholder, or generated test artifact.

Parameters:

value (str | Path) – Absolute path or path relative to the test directory.

Return type:

Path

verifier_client(name, replay_path, **options)

Create a Proxy Verifier client.

Parameters:

• name (str) – Unique support-process name within the scenario.

• replay_path (str | Path) – Replay file sent by verifier-client.

• options (Any) – Target ports, TLS, keys, and command options.

Return type:

ProcessService

verifier_server(name, replay_path, **options)

Create a Proxy Verifier server.

Parameters:

• name (str) – Unique support-process name within the scenario.

• replay_path (str | Path) – Replay file served by verifier-server.

• options (Any) – Listener, TLS, verbosity, and command options.

Return type:

VerifierServer

class tools.uranium.services.ProcessService(process, *, reject_expression='', ready_port=0, ready_address='127.0.0.1')

A pytest-owned non-ATS process used by a procedural test.

Parameters:

• process (ManagedProcess)

• reject_expression (str)

• ready_port (int)

• ready_address (str)

close()

Stop the service and validate expectations not already checked by wait.

Return type:

None

expect_return_codes(*codes)

Set the acceptable process exit statuses.

Parameters:

codes (int) – One or more integer exit statuses treated as success.

Return type:

None

property return_codes: tuple[int, ...]

Return the acceptable process exit statuses.

run(timeout=60)

Start the process and wait for completion.

Parameters:

timeout (float) – Maximum number of seconds to wait after startup.

Return type:

CommandResult

start()

Launch the service and wait for its configured TCP listener, if any.

Return type:

None

property stderr: StreamExpectations

Return the read-only standard-error expectation API.

property stderr_text: str

Return captured standard error.

property stdout: StreamExpectations

Return the read-only standard-output expectation API.

property stdout_text: str

Return captured standard output.

stop()

Stop a running service and finish capturing its output.

Return type:

None

wait(timeout=60)

Wait for the process and validate its captured output.

Parameters:

timeout (float) – Maximum number of seconds to wait for completion.

Return type:

CommandResult

wait_for_output(expression, count=1, timeout=10)

Wait until captured process output contains enough matching lines.

Parameters:

• expression (str) – Regular expression matched against combined standard output and standard error.

• count (int) – Minimum number of required matches.

• timeout (float) – Maximum number of seconds to wait.

Return type:

str

class tools.uranium.services.OriginServer(process, port, https_port, data_directory, address)

A pytest-owned microserver origin.

Parameters:

• process (ManagedProcess)

• port (int)

• https_port (int)

• data_directory (Path)

• address (str)

add_response(request, response, filename='sessionlog.json')

Append a request-response transaction to a microserver session.

Parameters:

• request (Mapping[str, Any]) – Request headers, body, and optional replay options.

• response (Mapping[str, Any]) – Response headers, body, and optional replay options.

• filename (str) – Session filename within the microserver data directory.

Return type:

None

start()

Launch the service and wait for its configured TCP listener, if any.

Return type:

None

class tools.uranium.services.DNSServer(process, port, zone_file)

A pytest-owned microDNS server.

Parameters:

• process (ManagedProcess)

• port (int)

• zone_file (Path)

add_records(records)

Append hostname mappings to the zone document.

Parameters:

records (Mapping[str, Sequence[str]]) – Hostnames mapped to one or more addresses.

Return type:

None

start()

Launch the service and wait for its configured TCP listener, if any.

Return type:

None

class tools.uranium.services.HttpBinServer(process, port)

A pytest-owned go-httpbin server.

Parameters:

• process (ManagedProcess)

• port (int)

start()

Launch the service and wait for its configured TCP listener, if any.

Return type:

None

class tools.uranium.services.VerifierServer(process, http_port, https_port)

A pytest-owned Proxy Verifier server.

Parameters:

• process (ManagedProcess)

• http_port (int)

• https_port (int)

start()

Launch the service and wait for its configured TCP listener, if any.

Return type:

None

Results and context

class tools.uranium.services.CommandResult(command, returncode, stdout, stderr)

The observable result of a client command.

Parameters:

• command (tuple[str, ...])

• returncode (int)

• stdout (str)

• stderr (str)

property output: str

Return stdout and stderr together for assertion diagnostics.

class tools.uranium.services.ProceduralContext(runtime, node_name, test_path, run_directory, use_uds=False)

Runtime and sandbox state shared by one procedural pytest item.

Parameters:

• runtime (TestRuntime)

• node_name (str)

• test_path (pathlib.Path)

• run_directory (pathlib.Path)

• use_uds (bool)

classmethod create(runtime, node_name, test_path, *, use_uds=False)

Create and empty the sandbox for one pytest test.

Parameters:

• runtime (TestRuntime) – Shared Uranium runtime configuration.

• node_name (str) – Pytest node identifier used to name the sandbox.

• test_path (pathlib.Path) – Path to the procedural test module.

• use_uds (bool) – Whether curl clients should use ATS Unix sockets.

Return type:

ProceduralContext

resolve_path(value)

Resolve a source or build artifact used by this test.

Parameters:

value (str | Path) – Absolute path or path relative to the test directory.

Return type:

Path

property test_directory: pathlib.Path

Return the source directory containing the procedural test.

class tools.uranium.runtime.TestRuntime(repository_root, build_root, ats_bin, verifier_bin, sandbox_root, layout, features, versions=<factory>)

Paths and build features shared by replay test items.

Parameters:

• repository_root (pathlib.Path)

• build_root (pathlib.Path)

• ats_bin (pathlib.Path)

• verifier_bin (pathlib.Path)

• sandbox_root (pathlib.Path)

• layout (dict[str, str])

• features (dict[str, Any])

• versions (dict[str, str])

allocate_port(socket_type=SocketKind.SOCK_STREAM)

Allocate a unique available port across all xdist workers.

Parameters:

socket_type (int) – Socket kind to probe, normally TCP or UDP.

Return type:

int

property build_uranium_tests: pathlib.Path

Return the build-tree mirror of uranium_tests.

classmethod create(repository_root, build_root, ats_bin, verifier_bin, sandbox_root)

Validate paths and query the installed ATS layout.

Parameters:

• repository_root (pathlib.Path)

• build_root (pathlib.Path)

• ats_bin (pathlib.Path)

• verifier_bin (pathlib.Path)

• sandbox_root (pathlib.Path)

Return type:

TestRuntime

execution_lock(is_exclusive)

Coordinate tests that must run without any parallel peers.

Parameters:

is_exclusive (bool)

Return type:

Iterator[None]

item_sandbox(node_name)

Return the named sandbox for one replay or procedural item.

Parameters:

node_name (str) – Pytest node identifier or replay name.

Return type:

pathlib.Path

prepare_sandbox(path, *, parent=None)

Create an empty sandbox below its owning test or the session root.

Parameters:

• path (Path) – Directory for this item or embedded replay.

• parent (Path | None) – Owning native test directory for an embedded replay.

Return type:

None

requirement_failure(requirements)

Return why a replay should be skipped, or None when it can run.

Parameters:

requirements (dict[str, Any])

Return type:

str | None

resolve_artifact(test_directory, value)

Resolve a source or build artifact referenced by replay metadata.

Parameters:

• test_directory (pathlib.Path)

• value (str)

Return type:

pathlib.Path

static sandbox_name(node_name)

Keep the full test name, making it safe as one directory component.

Parameters:

node_name (str) – Pytest node identifier or replay name.

Return type:

str

property test_plugins: pathlib.Path

Return the directory containing test-only plugins.

property test_tools: pathlib.Path

Return the source test-tools directory.

class tools.uranium.expectations.StreamExpectations(process_name, stream_name, path, test_directory)

Explicit assertions and the captured path for one process stream.

Parameters:

• process_name (str)

• stream_name (str)

• path (Path)

• test_directory (Path)

contains(expression, explanation, *, reflags=0)

Require this stream to contain a regular expression.

Parameters:

• expression (str) – Regular expression that must match the stream.

• explanation (str) – Non-empty description of the expected behavior.

• reflags (int) – Flags passed to re.search().

Return type:

None

excludes(expression, explanation, *, reflags=0)

Require this stream to exclude a regular expression.

Parameters:

• expression (str) – Regular expression that must not match the stream.

• explanation (str) – Non-empty description of the expected behavior.

• reflags (int) – Flags passed to re.search().

Return type:

None

property expectations: tuple[StreamExpectation, ...]

Return the registered expectations as an immutable tuple.

matches_gold(path, explanation='')

Require this stream to match a wildcard-aware gold file.

Parameters:

• path (str | Path) – Gold-file path, relative to the test source directory.

• explanation (str) – Optional description of the expected behavior.

Return type:

None

property path: pathlib.Path

Return the path containing this stream’s captured output.

reset()

Remove all registered expectations without changing stream identity.

Return type:

None

validate(content)

Apply every registered expectation to captured stream content.

Parameters:

content (str) – Complete captured text for this stream.

Return type:

None

class tools.uranium.expectations.StreamExpectation(kind, value, explanation='', reflags=0)

One assertion applied to a captured process stream.

Parameters:

• kind (str)

• value (str | Path)

• explanation (str)

• reflags (int)

class tools.uranium.process.ManagedProcess(name, command, run_directory, environment=None, expected_return_codes=(0,), test_directory=None)

Own a subprocess and its captured output files.

Parameters:

• name (str)

• command (Sequence[str])

• run_directory (Path)

• environment (dict[str, str] | None)

• expected_return_codes (Iterable[int])

• test_directory (Path | None)

expect_return_codes(*codes)

Set the acceptable process exit statuses.

Parameters:

codes (int) – One or more integer exit statuses treated as success.

Return type:

None

output()

Return the captured stdout and stderr for diagnostics.

Return type:

str

property return_code: int | None

Return the process exit status, or None while it is running.

property return_codes: tuple[int, ...]

Return the acceptable process exit statuses.

send_signal(signal_number)

Send @a signal_number to the process group leader.

Parameters:

signal_number (int)

Return type:

None

start()

Start the process in a new process group.

Return type:

None

property stderr: StreamExpectations

Return the read-only standard-error expectation API.

property stderr_text: str

Return captured standard error.

property stdout: StreamExpectations

Return the read-only standard-output expectation API.

property stdout_text: str

Return captured standard output.

stop(timeout=5.0)

Stop the process group, escalating to SIGKILL if necessary.

Parameters:

timeout (float)

Return type:

None

validate_output()

Apply registered stdout and stderr expectations.

Return type:

None

wait(timeout)

Wait for a one-shot process and validate its return code.

Parameters:

timeout (float)

Return type:

None

wait_until(ready, timeout, description)

Wait until @a ready succeeds while ensuring the process remains alive.

Parameters:

• ready (Callable[[], bool])

• timeout (float)

• description (str)

Return type:

None

Assertion helpers

tools.uranium.services.assert_matches_gold(actual, expected)

Assert that output matches a Uranium wildcard gold file.

Parameters:

• actual (str | Path) – Captured output or its file path.

• expected (Path) – Gold-file path containing literal text and wildcards.

Return type:

None

tools.uranium.services.wait_for_file_lines(path, expression, count, timeout=10)

Wait until a file contains enough matching lines.

Parameters:

• path (pathlib.Path) – File path to poll.

• expression (str) – Regular expression matched against the file contents.

• count (int) – Minimum number of required matches.

• timeout (float) – Maximum number of seconds to wait.

Return type:

str

tools.uranium.services.wait_for_metric(ats, name, expected, timeout=10)

Wait until an ATS metric reaches an expected value.

Parameters:

• ats (ATS) – Traffic Server instance queried through traffic_ctl.

• name (str) – Fully qualified metric name.

• expected (int) – Metric value required for success.

• timeout (float) – Maximum number of seconds to wait.

Return type:

int

tools.uranium.services.send_tcp(port, data, *, address='127.0.0.1', timeout=10)

Send bytes to a TCP listener and return its response.

Parameters:

• port (int) – Destination TCP port.

• data (str | bytes) – Text or bytes to send before closing the write side.

• address (str) – Destination IP address or hostname.

• timeout (float) – Connection and receive timeout in seconds.

Return type:

str