Add Cli class to docker.py (2586a564) · Commits · Dom Sekotill / Behave Utils

behave_utils/docker.py

+104 −0

Original line number	Diff line number	Diff line
		@@ -17,6 +17,7 @@ from pathlib import Path
		from secrets import token_hex
		from subprocess import DEVNULL
		from subprocess import PIPE
		from subprocess import CalledProcessError
		from subprocess import CompletedProcess
		from subprocess import Popen
		from subprocess import run
		@@ -25,17 +26,22 @@ from typing import IO
		from typing import Any
		from typing import Iterable
		from typing import Iterator
		from typing import Literal
		from typing import SupportsBytes
		from typing import Tuple
		from typing import TypeVar
		from typing import Union
		from typing import overload

		from .json import JSONObject
		from .proc import Arguments
		from .proc import Deserialiser
		from .proc import Environ
		from .proc import MutableArguments
		from .proc import PathArg
		from .proc import PathLike
		from .proc import coerce_args
		from .proc import exec_io

		HostMount = tuple[PathLike, PathLike]
		NamedMount = tuple[str, PathLike]
		@@ -387,3 +393,101 @@ class Network(Item):
		Remove the network
		"""
		docker_quiet("network", "rm", self._name)


		class Cli:
		"""
		Manage calling executables in a container

		Any arguments passed to the constructor will prefix the arguments passed when the object
		is called.
		"""

		T = TypeVar("T")

		def __init__(self, container: Container, *cmd: PathArg):
		self.container = container
		self.cmd = cmd

		@overload
		def __call__(
		self,
		*args: PathArg,
		input: str\|bytes\|SupportsBytes\|None = ...,
		deserialiser: Deserialiser[T],
		query: Literal[False] = False,
		**kwargs: Any,
		) -> T: ...

		@overload
		def __call__(
		self,
		*args: PathArg,
		input: str\|bytes\|SupportsBytes\|None = ...,
		deserialiser: None = None,
		query: Literal[True],
		**kwargs: Any,
		) -> int: ...

		@overload
		def __call__(
		self,
		*args: PathArg,
		input: str\|bytes\|SupportsBytes\|None = ...,
		deserialiser: None = None,
		query: Literal[False] = False,
		**kwargs: Any,
		) -> None: ...

		def __call__(
		self,
		*args: PathArg,
		input: str\|bytes\|SupportsBytes\|None = None,
		deserialiser: Deserialiser[Any]\|None = None,
		query: bool = False,
		**kwargs: Any,
		) -> Any:
		"""
		Run the container executable with the given arguments

		Input:
		Any bytes passed as "input" will be fed into the process' stdin pipe.

		Output:
		If "deserialiser" is provided it will be called with a memoryview of a buffer
		containing any bytes from the process' stdout; whatever is returned by
		"deserialiser" will be returned.

		If "query" is true the return code of the process will be returned.

		Otherwise nothing is returned.

		Note that "deserialiser" and "query" are mutually exclusive; if debugging is
		enabled an AssertionError will be raised if both are non-None/non-False, otherwise
		"query" is ignored.

		Errors:
		If "query" is not true any non-zero return code will cause CalledProcessError to
		be raised.
		"""
		# deserialiser = kwargs.pop('deserialiser', None)
		assert not deserialiser or not query

		data = (
		b"" if input is None else
		input.encode() if isinstance(input, str) else
		bytes(input)
		)
		cmd = self.container.get_exec_args([self.cmd, args], interactive=bool(data))

		if deserialiser:
		return exec_io(cmd, data, deserialiser=deserialiser, **kwargs)

		rcode = exec_io(cmd, data, **kwargs)
		if query:
		return rcode
		if not isinstance(rcode, int):
		raise TypeError(f"got rcode {rcode!r}")
		if 0 != rcode:
		raise CalledProcessError(rcode, ' '.join(coerce_args(cmd)))
		return None