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 |
|---|---|---|
|
One unstarted ATS instance named ats. |
|
|
Create multiple named ATS instances in one test. |
|
|
Create origins, DNS, verifier servers, and custom processes. |
|
|
Run curl respecting the selected TCP/UDS transport. |
|
|
Own the test's paths, runtime, and sandbox. |
|
|
|
Execute a replay manifest and all variants inside a native test. |
|
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 the isolated HTTP cache. |
|
|
Add a TLS listener and default test certificate. |
|
|
Add an HTTP/3 listener; normally used with TLS enabled. |
|
|
Add dedicated PROXY protocol listeners. |
|
|
Keep logs at predictable paths. Set false to test installed defaults;
an explicit |
|
|
Additional argument strings passed to |
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)
- create(name=None, **process_options)[ソース]
Create a named Traffic Server with fixture-owned cleanup.
- property run_directory: pathlib.Path
Return the test's isolated sandbox directory.
- class tools.uranium.services.ATS(context, name='ats', **process_options)[ソース]
A pytest-owned Traffic Server process with staged configuration.
- パラメータ:
context (ProceduralContext)
name (str)
process_options (Any)
- 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.
- append_records_document(values)[ソース]
Append another records.yaml document after the initial configuration.
- 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 byallow_errors()and unrecognized configuration values, unless the instance was created withdisable_log_checks=True.- 戻り値の型:
None
- property config_directory: pathlib.Path
Return the ATS configuration directory.
- 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 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.
- property log_directory: pathlib.Path
Return the ATS log directory.
- omit_config_file(filename)[ソース]
Leave a file absent from the materialized ATS configuration.
- パラメータ:
filename (str) -- Configuration filename to omit.
- 戻り値の型:
None
- 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.
- パラメータ:
- 戻り値の型:
- 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_ssl_multicert_yaml(document)[ソース]
Write structured TLS certificate configuration without defaults.
- 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.
- 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.
- 戻り値の型:
- property traffic_out: pathlib.Path
Return the path to traffic.out.
- 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.
- class tools.uranium.services.RecordsConfig(values)[ソース]
Stage records.yaml updates until ATS starts.
- 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
- 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.
- 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.
- 戻り値の型:
- 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.
- 戻り値の型:
- 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.
- 戻り値の型:
- 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.
- 戻り値の型:
- 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.
- 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.
- 戻り値の型:
- 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.
- パラメータ:
process (ManagedProcess)
reject_expression (str)
ready_port (int)
ready_address (str)
- expect_return_codes(*codes)[ソース]
Set the acceptable process exit statuses.
- パラメータ:
codes (int) -- One or more integer exit statuses treated as success.
- 戻り値の型:
None
- run(timeout=60)[ソース]
Start the process and wait for completion.
- パラメータ:
timeout (float) -- Maximum number of seconds to wait after startup.
- 戻り値の型:
- property stderr: StreamExpectations
Return the read-only standard-error expectation API.
- property stdout: StreamExpectations
Return the read-only standard-output expectation API.
- wait(timeout=60)[ソース]
Wait for the process and validate its captured output.
- パラメータ:
timeout (float) -- Maximum number of seconds to wait for completion.
- 戻り値の型:
- class tools.uranium.services.OriginServer(process, port, https_port, data_directory, address)[ソース]
A pytest-owned microserver origin.
- パラメータ:
process (ManagedProcess)
port (int)
https_port (int)
data_directory (Path)
address (str)
- class tools.uranium.services.DNSServer(process, port, zone_file)[ソース]
A pytest-owned microDNS server.
- パラメータ:
process (ManagedProcess)
port (int)
zone_file (Path)
- class tools.uranium.services.HttpBinServer(process, port)[ソース]
A pytest-owned go-httpbin server.
- パラメータ:
process (ManagedProcess)
port (int)
Results and context
- class tools.uranium.services.CommandResult(command, returncode, stdout, stderr)[ソース]
The observable result of a client command.
- 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.
- パラメータ:
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.
- パラメータ:
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.
- 戻り値の型:
- 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.
- パラメータ:
repository_root (pathlib.Path)
build_root (pathlib.Path)
ats_bin (pathlib.Path)
verifier_bin (pathlib.Path)
sandbox_root (pathlib.Path)
- allocate_port(socket_type=SocketKind.SOCK_STREAM)[ソース]
Allocate a unique available port across all xdist workers.
- 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.
- パラメータ:
repository_root (pathlib.Path)
build_root (pathlib.Path)
ats_bin (pathlib.Path)
verifier_bin (pathlib.Path)
sandbox_root (pathlib.Path)
- 戻り値の型:
- item_sandbox(node_name)[ソース]
Return the named sandbox for one replay or procedural item.
- パラメータ:
node_name (str) -- Pytest node identifier or replay name.
- 戻り値の型:
- prepare_sandbox(path, *, parent=None)[ソース]
Create an empty sandbox below its owning test or the session root.
- requirement_failure(requirements)[ソース]
Return why a replay should be skipped, or None when it can run.
- resolve_artifact(test_directory, value)[ソース]
Resolve a source or build artifact referenced by replay metadata.
- パラメータ:
test_directory (pathlib.Path)
value (str)
- 戻り値の型:
- static sandbox_name(node_name)[ソース]
Keep the full test name, making it safe as one directory component.
- 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.
- 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.
- property path: pathlib.Path
Return the path containing this stream's captured output.
- 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.
- パラメータ:
- expect_return_codes(*codes)[ソース]
Set the acceptable process exit statuses.
- パラメータ:
codes (int) -- One or more integer exit statuses treated as success.
- 戻り値の型:
None
- send_signal(signal_number)[ソース]
Send @a signal_number to the process group leader.
- パラメータ:
signal_number (int)
- 戻り値の型:
None
- property stderr: StreamExpectations
Return the read-only standard-error expectation API.
- property stdout: StreamExpectations
Return the read-only standard-output expectation API.
- stop(timeout=5.0)[ソース]
Stop the process group, escalating to SIGKILL if necessary.
- パラメータ:
timeout (float)
- 戻り値の型:
None
Assertion helpers
- tools.uranium.services.assert_matches_gold(actual, expected)[ソース]
Assert that output matches a Uranium wildcard gold file.
- 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.
- 戻り値の型:
- tools.uranium.services.wait_for_metric(ats, name, expected, timeout=10)[ソース]
Wait until an ATS metric reaches an expected value.