pytest_xdocker package

Submodules

pytest_xdocker.cache module

Cache providers.

class pytest_xdocker.cache.Cache

Bases: object

Base class for cache providers.

abstract get(key, default)

Return cached value for the given key or the default.

abstract set(key, value)

Save value for the given key.

exception pytest_xdocker.cache.CacheError

Bases: Exception

Raised with an unexpected cache error occurs.

class pytest_xdocker.cache.FileCache(cachedir, encode=<function cache_encode>, decode=<function cache_decode>)

Bases: Cache

Lightweight implementation of pytest.cache.

Parameters:

  • path – Base path to cache directory.

  • encode – Encoding function, defaults to cache_encode

  • decode – Decoding function, defaults to cache_decode

decode
encode
get(key, default)

Read from file.

set(key, value)

Write to file.

class pytest_xdocker.cache.MemoryCache(memory=_Nothing.NOTHING)

Bases: Cache

Memory cache.

get(key, default)

Read from dict.

set(key, value)

Write the value to dict.

class pytest_xdocker.cache.NullCache

Bases: Cache

Null cache.

This cache never sets a value and always gets the default value.

get(key, default)

Noop.

set(key, value)

Noop.

pytest_xdocker.cache.cache_decode(payload)

Deserialize cache payload.

pytest_xdocker.cache.cache_encode(data)

Serialize cache payload.

pytest_xdocker.command module

Module to build shell commands declaratively.

A Command instance can be used to build a shell command:

>>> command = Command('whoami')

The command can then be executed later:

>>> lines = command.execute().splitlines()
>>> len(lines)
1
class pytest_xdocker.command.Command(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Iterable

Manages a shell command.

execute(**kwargs)

Run the command.

reparent(parent=None)

Add a wrapping command.

to_string(escape=None)

Stringify the command.

with_optionals(*optionals)

Add optional args.

with_positionals(*positionals)

Add positional args.

class pytest_xdocker.command.OptionalArg(name, type=<function empty_type>, **kwargs)

Bases: object

Descriptor for optional arguments.

Parameters:

  • name – Name of the option.

  • type – Optional argument type, defaults to empty_type.

  • kwargs – Optional keyword arguments passed to the type.

class pytest_xdocker.command.PositionalArg(type=<function arg_type>, **kwargs)

Bases: object

Descriptor for positional arguments.

Parameters:

  • type – Optional argument type, defaults to arg_type.

  • kwargs – Optional keyword arguments passed to the type.

pytest_xdocker.command.arg_type(arg, **kwargs)

Option type for a single args.

pytest_xdocker.command.args_type(*args, **kwargs)

Option type for multiple args.

pytest_xdocker.command.const_type(const)

Option arg for a constant string.

pytest_xdocker.command.empty_type()

Option arg for an undefined optional arg.

pytest_xdocker.command.script_to_command(script, cls=<class 'pytest_xdocker.command.Command'>)

On Windows, console scripts created as .exe should be called directly and those created as .cmd with Python.

pytest_xdocker.docker module

Docker tools.

class pytest_xdocker.docker.DockerBuildCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker build”.

with_build_arg()

Configure a build ARG of the Dockerfile.

Parameters:

  • key – ARG key.

  • value – Optional value, no value will carry from envionment variable.

with_file()

Name of the Dockerfile, defaults to PATH/Dockerfile.

with_pull()

Always attempt to pull a newer version of the image.

with_tag()

Name and optionally a tag in the ‘name:tag’ format.

class pytest_xdocker.docker.DockerCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker”.

build(path)

Return a build command.

command(command)

Return the base command.

compose()

Return a compose command.

exec_(name)

Return an exec command.

logs(name)

Return a logs command.

port(name)

Return a port command.

pull(image)

Return a pull command.

remove(name)

Return a rm command.

run(image)

Return a run command.

with_debug()

Enable debug mode.

with_version()

Print version information and quit.

class pytest_xdocker.docker.DockerComposeBuildCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker compose build”.

with_no_cache()

Do not use cache when building the image.

with_pull()

Always attempt to pull a newer version of the image.

class pytest_xdocker.docker.DockerComposeCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker compose”.

build(*services)

Return a build command.

run(service)

Return a run command.

up(*services)

Return an up command.

with_env_file()

Specify an alternate environment file.

Parameters:

file – Path to env file.

with_file()

Compose configuration files.

Parameters:

file – Path to configuration file.

with_project_name()

Assign a project name to the compose configuration.

Parameters:

project_name – Project name.

class pytest_xdocker.docker.DockerComposeRunCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: DockerRunCommand

Shortcut for “docker compose run”.

with_build()

Build image before starting container.

class pytest_xdocker.docker.DockerComposeUpCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker compose up”.

with_build()

Build images before starting containers.

with_force_recreate()

Recreate containers even if their configuration and image haven’t changed.

class pytest_xdocker.docker.DockerContainer(name, inspect=None)

Bases: object

Manager a docker container.

property env

Return the environment of the container as a dictionary.

property exposed_ip

Return the IP exposed to other containers.

property exposed_port

Return the only port exposed to other containers.

property exposed_ports

Return all ports exposed to other containers.

host_ip(port=None)

Return the host IP accessible from the host.

Parameters:

port – Port exposed to the host, defaults to port_binding.

host_port(port=None)

Return the host port accessible from the host.

Parameters:

port – Port exposed to the host, defaults to port_binding.

property isrunning

Check if has running state.

property port_binding

Return the only port binding.

property port_bindings

Return all port bindings.

remove()

Remove the container.

start()

Start the container.

property status

Get status from inspect.

stop(wait=False)

Stop the container.

class pytest_xdocker.docker.DockerExecCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker exec”.

with_command()

Add command to execute in the docker container.

Parameters:

command – List ofcommands passed to docker.

with_detach()

Execute the command in the background.

with_env()

Set environment variables.

Parameters:

  • key – Environment key.

  • value – Optional environment value.

with_interactive()

Keep STDIN open even if not attached.

class pytest_xdocker.docker.DockerImage

Bases: object

Representation of a docker image.

classmethod from_string(string)

Make a DockerImage based on the given string.

class pytest_xdocker.docker.DockerImageDigest(name, digest)

Bases: DockerImage

Representation of a image:tag@digest.

digest
name
class pytest_xdocker.docker.DockerImageId(identifier)

Bases: DockerImage

Representation of a docker image id.

identifier
class pytest_xdocker.docker.DockerImageTag(name, tag='latest')

Bases: DockerImage

Representation of a image:tag.

name
tag
class pytest_xdocker.docker.DockerInspect(name, data=None)

Bases: UserDict

Reader for a docker inspect call.

property command

Return the base command.

property data

Inspect data as a dictionary.

get(*keys)

Read the keys from the docker inspect result.

refresh()

Refresh the inspect data.

class pytest_xdocker.docker.DockerLogsCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker logs”.

with_follow()

Follow log output.

with_since(timestamp)

Show logs since timestamp.

class pytest_xdocker.docker.DockerNetworkInspect(name, data=None)

Bases: DockerInspect

Shortcut for “docker network inspect”.

property command

Return the base command.

class pytest_xdocker.docker.DockerPortCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker port”.

with_private_port()
class pytest_xdocker.docker.DockerPullCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker pull”.

execute(**kwargs)

Run the docker pull command and output the progress.

Retries when failing to pull because this is usually caused by a recoverable network failure.

Parameters:

kwargs – See Command.execute.

class pytest_xdocker.docker.DockerRemoveCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker remove”.

with_force()

Force the removal of a running container (uses SIGKILL).

with_volumes()

Remove the volumes associated with the container.

class pytest_xdocker.docker.DockerRunCommand(command, parent=<tuple_iterator object>, positionals=_Nothing.NOTHING, optionals=_Nothing.NOTHING)

Bases: Command

Shortcut for “docker run”.

execute(**kwargs)

Run the docker command and output the progress.

Parameters:

kwargs – Optional keyword arguments passed to run.

with_command()

Add command to run in the docker container.

Parameters:

command – List of commands passed to docker.

with_detach()

Run the container in the background.

with_env()

Set environment variables.

Parameters:

  • key – Environment key.

  • value – Optional environment value.

with_env_file()

Read in a file of environment variables.

Parameters:

file – Path to env file.

with_interactive()

Keep STDIN open even if not attached.

Link to another container.

Parameters:

name – Name of the other container.

with_name()

Assign a name to the container.

Parameters:

name – Container name.

with_publish(container_ports, host_ports=None, host_ip=None)

Publish ports from the docker container.

Parameters:

  • container_ports – Container ports, ie 1234 or 1234-1238/tcp.

  • host_ports – Optional host ports, defaults to container ports.

  • host_ip – Optional host IP, defaults to all interfaces.

with_remove()

Automatically remove the container when it exits.

with_volume(host_src, container_dest=None, options=None)

Mount volumes from the host to the docker container.

Parameters:

  • host_src – Source path from the host.

  • container_dest – Destination path in the container, defaults to host_src.

  • options – Comma separated list of options like [rw|ro], [z|Z], [[r]shared|[r]slave|[r]private], [nocopy], etc.

with_workdir()

Set the working directory in the container.

Parameters:

workdir – Working directory.

class pytest_xdocker.docker.DockerText

Bases: Iterable

Tool to write a text file.

with_comment(comment)

Write a comment beginning with a #.

Parameters:

comment – String to comment.

with_line(line)

Add line to the test.

write(path)

Write the text to the given path.

Parameters:

path – Path where to write the docker text.

class pytest_xdocker.docker.Dockerfile(image)

Bases: DockerText

Tool to write a Dockerfile.

classmethod from_lines(lines)

Load lines of a dockerfile.

classmethod from_path(path)

Read a dockerfile content.

classmethod from_string(string)

Read a dockerfile content.

image
with_add(src, dest)

Add an Add instruction.

The ADD instruction copies new files, directories or remote file URLs from <src> and adds them to the filesystem of the image at the path <dest>.

with_copy(src, dest)

Add a COPY instruction.

The COPY instruction copies new files or directories from <src> and adds them to the filesystem of the container at the path <dest>.

with_env(key, value)

Add an ENV instruction.

The ENV instruction sets the environment variable <key> to the value <value>.

with_expose(*ports)

Add an EXPOSE instruction.

The EXPOSE instruction informs Docker that the container listens on the specified network ports at runtime.

with_instruction(instruction, *args)

Add instruction to the Dockerfile.

Parameters:

  • instruction – Name of the instruction, eg ADD.

  • args – Argument list for the instruction.

with_run(command)

Add a RUN instruction.

The RUN instruction will execute any commands in a new layer on top of the current image and commit the results. The resulting committed image will be used for the next step in the Dockerfile.

with_workdir(workdir)

Add a WORKDIR instruction.

The WORKDIR instruction sets the working directory for any RUN, CMD, ENTRYPOINT, COPY and ADD instructions that follow it in the Dockerfile.

class pytest_xdocker.docker.Dockerignore

Bases: DockerText

Tool to write a .dockerignore file.

with_pattern(pattern)

Add a CLI instruction.

The CLI interprets the .dockerignore file as a newline-separated list of patterns similar to the file globs of Unix shells.

pytest_xdocker.docker.docker_env_type(key, value=None)

Docker environment variable type.

Parameters:

  • key – Environment key.

  • value – Optional environment value.

pytest_xdocker.fixtures module

XProcess fixtures.

pytest_xdocker.fixtures.process(request)

Initiliaze XProcess.

pytest_xdocker.fixtures.pytest_addoption(parser)

Add pytest options.

pytest_xdocker.fixtures.pytest_runtest_makereport(item, call)

Workaround pyest-xprocess trying to read from closed logfiles.

pytest_xdocker.lock module

Locks.

exception pytest_xdocker.lock.AlreadyLockedError

Bases: Exception

Raised when a lock was already acquired.

class pytest_xdocker.lock.BaseLock

Bases: object

Base class for locks.

abstract property is_locked

Return True if lock acquired, False otherwise.

abstract lock()

Acquire the lock.

Raises:

AlreadyLockedError – If the lock was already acquired.

abstract unlock()

Release the lock.

Raises:

NotLockedError – If the lock was never acquired or already released.

class pytest_xdocker.lock.FileLock(lockfile, lockfd=None)

Bases: BaseLock

Advisory file locking.

Parameters:

lockfile – Path to the lock file.

property is_locked

Check if the file is locked, based on the file descriptor.

lock()

See BaseLock.lock.

unlock()

See BaseLock.unlock.

class pytest_xdocker.lock.MemoryLock(is_locked=False)

Bases: BaseLock

In-memory locking.

property is_locked

Accessor for read-only private field.

lock()

See BaseLock.lock.

unlock()

See BaseLock.unlock.

exception pytest_xdocker.lock.NotLockedError

Bases: Exception

Raised when the lock was never acquired or already released.

class pytest_xdocker.lock.NullLock

Bases: BaseLock

Null pattern implementation.

property is_locked

Return False.

lock()

Do nothing.

unlock()

Do nothing.

pytest_xdocker.lock.lock(fd)
pytest_xdocker.lock.unlock(fd)

pytest_xdocker.network module

Networking tools.

pytest_xdocker.network.get_host_ip()

Get an IP on this host.

The function tries to use the IP bound to the docker0 interface which should be available on Linux and, if that fails, it defaults to the default en0-en9 interfaces which should be available on MacOS.

pytest_xdocker.network.get_open_port()

Get an unused port.

There is a race condition where the port could be taken after this method closes the socket but before the consumer opens it. Since this is just for test fixtures I’m not worrying about that.

pytest_xdocker.process module

XProcess management.

class pytest_xdocker.process.Process(config=None, root_dir=None, log=None)

Bases: XProcess

XProcess with restarting capability and extra logging.

ensure(name, prepare_func, restart=None)

Ensure the container is running or restarted if requested.

getinfo(name)

Get the process info based on the name.

property root_dir

Return the root dir, but as a snake_case property.

class pytest_xdocker.process.ProcessConfig(root_dir=None, cache=None)

Bases: object

Lightweight process config.

option = None
trace(string)

Write a trace log.

class pytest_xdocker.process.ProcessData(pattern, args, env=None, timeout=120)

Bases: ProcessData

Representation of a process’ data.

change(**changes)

Access for namedtuple _replace so that it doesn’t look private.

class pytest_xdocker.process.ProcessInfo(path, name)

Bases: XProcessInfo

XProcessInfo with better kill/running methods.

isrunning()

Check if process is running.

kill()

Kill the process and wait for children to exit.

class pytest_xdocker.process.ProcessServer(process=None)

Bases: object

Base class for a container process.

get_cache_publish(controldir, container_ports)

Read from cache or define published ports.

abstract prepare_func(controldir)

Prepare function passed to Process.ensure.

Parameters:

controldir – py.path instance of the control directory.

Returns:

ProcessData used to ensure the server is running.

run(name, restart=None)

Run the server by name.

Parameters:

  • name – Name of the process.

  • restart – True to restart, False to keep the process, None to look in the process config.

pytest_xdocker.process.get_process_dir(config, root_dir=None)

Get the process directory under the root directory.

Parameters:

root_dir – Root directory, defaults to the root of the project.

pytest_xdocker.process.get_root_dir(config)

Get the root directory of the project.

The root is assumed to contain a .git subdirectory, or config.ini.

Parameters:

config – Pytest or Process config instance.

pytest_xdocker.process.pytest_cache_config_compat

alias of ProcessConfig

pytest_xdocker.retry module

Retry functions until a condition is met.

Here is an example of using the retry function:

>>> from itertools import count
>>> from hamcrest import greater_than
>>> retry(calling(next, count())).until(greater_than(2), delay=0)
3
class pytest_xdocker.retry.Calling(func, args=_Nothing.NOTHING, kwargs=_Nothing.NOTHING, returned=<object object>)

Bases: object

Function call configuration that can be repeated.

args
func
kwargs
returned
class pytest_xdocker.retry.CatchingProbe(func, exception, pattern='')

Bases: Probe

Probe to ignore an exception while retrying.

exception
func
pattern
class pytest_xdocker.retry.Poller(tries, delay, sleeper=<built-in function sleep>)

Bases: object

Poller for retrying an operation.

check(probe)

Poll until the probe succeeds.

delay
sleeper
tries
class pytest_xdocker.retry.Probe

Bases: object

Base probe class.

class pytest_xdocker.retry.ProbeResult(success, returned=None, raised=None)

Bases: object

Result of a probe.

raised
returned
success
class pytest_xdocker.retry.Retry(func)

Bases: object

Retry the given function until the expected result.

catching(exception, pattern='', tries=30, delay=1)

Return a poller with a catching probe.

func
until(value, tries=30, delay=1)

Return a poller with a value check probe.

class pytest_xdocker.retry.UntilProbe(func, value)

Bases: Probe

Probe to expect a value from a retry.

func
value
pytest_xdocker.retry.calling(func, *args, **kwargs)

Defer calling a function with the given arguments.

Parameters:

  • func – The function or method to be called.

  • args – Optional positional arguments.

  • kwargs – Optional keyword arguments.

Returns:

A Calling instance which can be called later.

pytest_xdocker.retry.retry(func, *args, **kwargs)

Retry calling a function with the given arguments.

Returns:

A Retry instance on which to specify the retry condition.

pytest_xdocker.retry.retry_catching(*args, **kwargs)

Retry decorator for catching exceptions.

pytest_xdocker.validators module

Attr validators.

pytest_xdocker.validators.matches(matcher)

Use as field(validator=…) to matcher based validation.

pytest_xdocker.xdocker module

Run a docker container with pytest xprocess.

The xprocess plugin extends pytest with options like –xkill which sends SIGKILL to fixtures processes. The intention is for the process to stop, so this script ensures the docker container is removed. The script is called with the same arguments passed to docker run:

xdocker run alpine:3.14 sleep 600

pytest_xdocker.xdocker.docker_call(*args, command=DockerCommand(['docker']))

Call Docker to run a container and return the container name.

pytest_xdocker.xdocker.docker_remove(name)

Remove a Docker container forcefully and ignore errors.

pytest_xdocker.xdocker.docker_run(*args, command=Command(['docker', 'run']))

Run a Docker container detached and return the container name.

pytest_xdocker.xdocker.main(argv=None)

Launch and monitor a container.

pytest_xdocker.xdocker.monitor_container(name, interval=1)

Monitor that a Docker container exists.

If the container is running, follow the logs. If it is stopped, inspect the status every interval seconds.

Parameters:

  • name – Name of the docker container to monitor.

  • interval – Check the container status every interval seconds.

pytest_xdocker.xdocker.monitor_ppid(name, ppid=None, interval=1)

Monitor a parent PID associated with a Docker container.

Wait for the parent PID to exit and then remove the associated container.

Parameters:

  • name – Name of the docker container to remove.

  • ppid – The parent PID to monitor. If None, uses os.getppid().

  • interval – Wait for the parent PID every interval seconds.

pytest_xdocker.xdocker.wait_ppid(ppid=None, interval=1)

Wait for a parent PID to exit (become a zombie).

Parameters:

  • ppid – The parent PID to monitor. If None, uses os.getppid().

  • interval – Check the parent PID status every interval seconds.

Module contents