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.

パラメータ:

context (ProceduralContext)

close()[ソース]

Stop and validate every created instance in reverse order.

戻り値の型:

None

create(name=None, **process_options)[ソース]

Create a named Traffic Server with fixture-owned cleanup.

パラメータ:
  • name (str | None) -- Unique process name, or None for an automatic name.

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

戻り値の型:

ATS

has_feature(feature)[ソース]

Return whether ATS was built with a feature.

パラメータ:

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

戻り値の型:

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.

戻り値の型:

str

class tools.uranium.services.ATS(context, name='ats', **process_options)[ソース]

A pytest-owned Traffic Server process with staged configuration.

パラメータ:
add_default_ssl_files()[ソース]

Enable TLS using Uranium's default certificate and key.

戻り値の型:

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.

パラメータ:

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

戻り値の型:

None

allow_private_connect(methods=('CONNECT',))[ソース]

Allow outbound test traffic to loopback addresses.

パラメータ:

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

戻り値の型:

None

append_records_document(values)[ソース]

Append another records.yaml document after the initial configuration.

パラメータ:

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

戻り値の型:

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.

パラメータ:

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

戻り値の型:

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.

戻り値の型:

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.

パラメータ:

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

戻り値の型:

None

copy_to_config(*paths)[ソース]

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

パラメータ:

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

戻り値の型:

None

copy_to_ssl(*paths)[ソース]

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

パラメータ:

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

戻り値の型:

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.

パラメータ:

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

戻り値の型:

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.

パラメータ:
  • expression (str) -- Regular expression required in startup output.

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

戻り値の型:

None

has_feature(feature)[ソース]

Return whether ATS was built with a feature.

パラメータ:

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

戻り値の型:

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.

戻り値の型:

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.

パラメータ:

filename (str) -- Configuration filename to omit.

戻り値の型:

None

plugin_exists(name)[ソース]

Return whether an installed ATS plugin is available.

パラメータ:

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

戻り値の型:

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.

パラメータ:
  • request (object) -- JSON string or JSON-serializable request value.

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

戻り値の型:

CommandResult

run(*arguments, timeout=30)[ソース]

Run a command with this ATS instance's environment.

パラメータ:
  • arguments (str | Path) -- Executable followed by its command-line arguments.

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

戻り値の型:

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.

パラメータ:
  • script (str) -- Bash source to execute.

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

戻り値の型:

CommandResult

property runtime_directory: pathlib.Path

Return the ATS local-state directory.

send_signal(signal_number)[ソース]

Send a signal to the Traffic Server process.

パラメータ:

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

戻り値の型:

None

set_environment(name, value)[ソース]

Set one environment variable for Traffic Server and its tools.

パラメータ:
  • name (str) -- Environment variable name.

  • value (str) -- Environment variable value.

戻り値の型:

None

set_logging_yaml(document)[ソース]

Write a structured logging.yaml configuration when ATS starts.

パラメータ:

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

戻り値の型:

None

set_ssl_multicert_yaml(document)[ソース]

Write structured TLS certificate configuration without defaults.

パラメータ:

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

戻り値の型:

None

set_startup_timeout(timeout)[ソース]

Set the maximum time allowed for ATS startup.

パラメータ:

timeout (float) -- Startup timeout in seconds.

戻り値の型:

None

property ssl_directory: pathlib.Path

Return the ATS TLS material directory.

start()[ソース]

Materialize configuration and wait until Traffic Server is ready.

戻り値の型:

None

stop()[ソース]

Stop Traffic Server without relinquishing fixture ownership.

戻り値の型:

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.

戻り値の型:

str

traffic_ctl(*arguments, timeout=30)[ソース]

Run traffic_ctl against this ATS run root.

パラメータ:
  • arguments (str) -- Arguments passed to traffic_ctl.

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

戻り値の型:

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.

パラメータ:

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

戻り値の型:

None

wait(timeout=60)[ソース]

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

パラメータ:

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

戻り値の型:

None

write_body_factory_file(relative_path, content)[ソース]

Materialize a custom body-factory template on start.

パラメータ:
  • relative_path (str) -- Destination path below the body-factory root.

  • content (str) -- Complete template contents.

戻り値の型:

None

write_config_file(filename, content)[ソース]

Materialize a test-owned ATS configuration file on start.

パラメータ:
  • filename (str) -- Destination filename under the config directory.

  • content (str) -- Complete file contents.

戻り値の型:

None

write_runtime_file(filename, content)[ソース]

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

パラメータ:
  • filename (str) -- Destination filename under the runtime directory.

  • content (str) -- Complete file contents.

戻り値の型:

None

class tools.uranium.services.RecordsConfig(values)[ソース]

Stage records.yaml updates until ATS starts.

パラメータ:

values (dict[str, object])

update(values)[ソース]

Merge flat ATS record names into the staged values.

パラメータ:

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

戻り値の型:

None

class tools.uranium.services.ConfigFile(lines, path)[ソース]

Stage one line-oriented ATS configuration file.

パラメータ:
add_line(line)[ソース]

Append one line to the staged file.

パラメータ:

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

戻り値の型:

None

add_lines(lines)[ソース]

Append lines from a string or sequence.

パラメータ:

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

戻り値の型:

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.

パラメータ:
  • 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.

戻り値の型:

bool

get(ats, path='/', *, headers=None, options='', timeout=30)[ソース]

Send a GET request to an ATS instance.

パラメータ:
  • 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.

戻り値の型:

CommandResult

run(arguments, timeout=30)[ソース]

Run curl without applying an ATS transport.

パラメータ:
  • 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.

戻り値の型:

CommandResult

run_for(ats, arguments, timeout=30)[ソース]

Run curl arguments using the selected ATS transport.

パラメータ:
  • 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.

戻り値の型:

CommandResult

run_script(ats, script, timeout=30)[ソース]

Run a Bash script containing curl command placeholders.

パラメータ:
  • 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.

戻り値の型:

CommandResult

static supports(feature)[ソース]

Return whether the installed curl advertises a feature.

パラメータ:

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

戻り値の型:

bool

class tools.uranium.services.ServiceFactory(context)[ソース]

Create pytest-owned support processes for procedural tests.

パラメータ:

context (ProceduralContext)

allocate_port(socket_type=SocketKind.SOCK_STREAM)[ソース]

Reserve a listener port for a bespoke support process.

パラメータ:

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

戻り値の型:

int

dns(name, **options)[ソース]

Create a microDNS server.

パラメータ:
  • name (str) -- Unique support-process name within the scenario.

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

戻り値の型:

DNSServer

httpbin(name, **options)[ソース]

Create a go-httpbin server.

パラメータ:
  • name (str) -- Unique support-process name within the scenario.

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

戻り値の型:

HttpBinServer

origin(name, **options)[ソース]

Create a microserver origin.

パラメータ:
  • name (str) -- Unique support-process name within the scenario.

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

戻り値の型:

OriginServer

process(name, command, *, environment=None, ready_port=0, ready_address='127.0.0.1')[ソース]

Create an arbitrary managed process for a bespoke scenario.

パラメータ:
  • 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.

戻り値の型:

ProcessService

proxy_verifier_at_least(version)[ソース]

Return whether Proxy Verifier meets a minimum version.

パラメータ:

version (str) -- Minimum dotted version string.

戻り値の型:

bool

resolve_path(value)[ソース]

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

パラメータ:

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

戻り値の型:

Path

verifier_client(name, replay_path, **options)[ソース]

Create a Proxy Verifier client.

パラメータ:
  • 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.

戻り値の型:

ProcessService

verifier_server(name, replay_path, **options)[ソース]

Create a Proxy Verifier server.

パラメータ:
  • 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.

戻り値の型:

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.

パラメータ:
close()[ソース]

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

戻り値の型:

None

expect_return_codes(*codes)[ソース]

Set the acceptable process exit statuses.

パラメータ:

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

戻り値の型:

None

property return_codes: tuple[int, ...]

Return the acceptable process exit statuses.

run(timeout=60)[ソース]

Start the process and wait for completion.

パラメータ:

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

戻り値の型:

CommandResult

start()[ソース]

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

戻り値の型:

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.

戻り値の型:

None

wait(timeout=60)[ソース]

Wait for the process and validate its captured output.

パラメータ:

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

戻り値の型:

CommandResult

wait_for_output(expression, count=1, timeout=10)[ソース]

Wait until captured process output contains enough matching lines.

パラメータ:
  • 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.

戻り値の型:

str

class tools.uranium.services.OriginServer(process, port, https_port, data_directory, address)[ソース]

A pytest-owned microserver origin.

パラメータ:
add_response(request, response, filename='sessionlog.json')[ソース]

Append a request-response transaction to a microserver session.

パラメータ:
  • 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.

戻り値の型:

None

start()[ソース]

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

戻り値の型:

None

class tools.uranium.services.DNSServer(process, port, zone_file)[ソース]

A pytest-owned microDNS server.

パラメータ:
add_records(records)[ソース]

Append hostname mappings to the zone document.

パラメータ:

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

戻り値の型:

None

start()[ソース]

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

戻り値の型:

None

class tools.uranium.services.HttpBinServer(process, port)[ソース]

A pytest-owned go-httpbin server.

パラメータ:
start()[ソース]

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

戻り値の型:

None

class tools.uranium.services.VerifierServer(process, http_port, https_port)[ソース]

A pytest-owned Proxy Verifier server.

パラメータ:
start()[ソース]

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

戻り値の型:

None

Results and context

class tools.uranium.services.CommandResult(command, returncode, stdout, stderr)[ソース]

The observable result of a client command.

パラメータ:
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.

パラメータ:
classmethod create(runtime, node_name, test_path, *, use_uds=False)[ソース]

Create and empty the sandbox for one pytest test.

パラメータ:
  • 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.

戻り値の型:

ProceduralContext

resolve_path(value)[ソース]

Resolve a source or build artifact used by this test.

パラメータ:

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

戻り値の型:

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.

パラメータ:
allocate_port(socket_type=SocketKind.SOCK_STREAM)[ソース]

Allocate a unique available port across all xdist workers.

パラメータ:

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

戻り値の型:

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.

パラメータ:
戻り値の型:

TestRuntime

execution_lock(is_exclusive)[ソース]

Coordinate tests that must run without any parallel peers.

パラメータ:

is_exclusive (bool)

戻り値の型:

Iterator[None]

item_sandbox(node_name)[ソース]

Return the named sandbox for one replay or procedural item.

パラメータ:

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

戻り値の型:

pathlib.Path

prepare_sandbox(path, *, parent=None)[ソース]

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

パラメータ:
  • path (Path) -- Directory for this item or embedded replay.

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

戻り値の型:

None

requirement_failure(requirements)[ソース]

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

パラメータ:

requirements (dict[str, Any])

戻り値の型:

str | None

resolve_artifact(test_directory, value)[ソース]

Resolve a source or build artifact referenced by replay metadata.

パラメータ:
戻り値の型:

pathlib.Path

static sandbox_name(node_name)[ソース]

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

パラメータ:

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

戻り値の型:

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.

パラメータ:
  • process_name (str)

  • stream_name (str)

  • path (Path)

  • test_directory (Path)

contains(expression, explanation, *, reflags=0)[ソース]

Require this stream to contain a regular expression.

パラメータ:
  • 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().

戻り値の型:

None

excludes(expression, explanation, *, reflags=0)[ソース]

Require this stream to exclude a regular expression.

パラメータ:
  • 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().

戻り値の型:

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.

パラメータ:
  • path (str | Path) -- Gold-file path, relative to the test source directory.

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

戻り値の型:

None

property path: pathlib.Path

Return the path containing this stream's captured output.

reset()[ソース]

Remove all registered expectations without changing stream identity.

戻り値の型:

None

validate(content)[ソース]

Apply every registered expectation to captured stream content.

パラメータ:

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

戻り値の型:

None

class tools.uranium.expectations.StreamExpectation(kind, value, explanation='', reflags=0)[ソース]

One assertion applied to a captured process stream.

パラメータ:
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.

パラメータ:
  • 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.

パラメータ:

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

戻り値の型:

None

output()[ソース]

Return the captured stdout and stderr for diagnostics.

戻り値の型:

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.

パラメータ:

signal_number (int)

戻り値の型:

None

start()[ソース]

Start the process in a new process group.

戻り値の型:

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.

パラメータ:

timeout (float)

戻り値の型:

None

validate_output()[ソース]

Apply registered stdout and stderr expectations.

戻り値の型:

None

wait(timeout)[ソース]

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

パラメータ:

timeout (float)

戻り値の型:

None

wait_until(ready, timeout, description)[ソース]

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

パラメータ:
戻り値の型:

None

Assertion helpers

tools.uranium.services.assert_matches_gold(actual, expected)[ソース]

Assert that output matches a Uranium wildcard gold file.

パラメータ:
  • actual (str | Path) -- Captured output or its file path.

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

戻り値の型:

None

tools.uranium.services.wait_for_file_lines(path, expression, count, timeout=10)[ソース]

Wait until a file contains enough matching lines.

パラメータ:
  • 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.

戻り値の型:

str

tools.uranium.services.wait_for_metric(ats, name, expected, timeout=10)[ソース]

Wait until an ATS metric reaches an expected value.

パラメータ:
  • 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.

戻り値の型:

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.

パラメータ:
  • 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.

戻り値の型:

str