Make docker.Cli a subclass of new, more general proc.Executor class (3e561110) · Commits · Dom Sekotill / Behave Utils

behave_utils/docker.py

+14 −88

Original line number	Diff line number	Diff line
		# Copyright 2021 Dominik Sekotill <dom.sekotill@kodo.org.uk>
		# Copyright 2021-2022 Dominik Sekotill <dom.sekotill@kodo.org.uk>
		#
		# This Source Code Form is subject to the terms of the Mozilla Public
		# License, v. 2.0. If a copy of the MPL was not distributed with this
		@@ -30,13 +30,11 @@ 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 MutableMapping
		from typing import Tuple
		from typing import TypeVar
		from typing import Union
		from typing import cast
		from typing import overload

		from .binaries import DownloadableDocker
		from .json import JSONArray
		@@ -45,8 +43,8 @@ from .proc import Argument
		from .proc import Arguments
		from .proc import Deserialiser
		from .proc import Environ
		from .proc import Executor
		from .proc import MutableArguments
		from .proc import coerce_args
		from .proc import exec_io

		MountPath = Union[PathLike[bytes], PathLike[str]]
		@@ -537,7 +535,7 @@ class Network(Item):
		raise LookupError(f"No free addresses found in subnet {net}")


		class Cli:
		class Cli(Executor):
		"""
		Manage calling executables in a container

		@@ -545,91 +543,19 @@ class Cli:
		is called.
		"""

		T = TypeVar("T")

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

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

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

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

		def __call__(
		self,
		*args: Argument,
		input: str\|bytes\|SupportsBytes\|None = None,
		deserialiser: Deserialiser[Any]\|None = None,
		query: bool = False,
		**kwargs: Any,
		) -> Any:
		cmd: Arguments,
		kwargs: MutableMapping[str, Any],
		has_input: bool,
		is_query: bool,
		deserialiser: Deserialiser[Any]\|None,
		) -> Arguments:
		"""
		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.
		Prefix the command arguments with a command necessary for executing in a container
		"""
		# 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=data, deserialiser=deserialiser, **kwargs)

		rcode = exec_io(cmd, data=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
		return self.container.get_exec_args(cmd, interactive=has_input)

behave_utils/proc.py

+122 −1

Original line number	Diff line number	Diff line
		# Copyright 2021 Dominik Sekotill <dom.sekotill@kodo.org.uk>
		# Copyright 2021-2022 Dominik Sekotill <dom.sekotill@kodo.org.uk>
		#
		# This Source Code Form is subject to the terms of the Mozilla Public
		# License, v. 2.0. If a copy of the MPL was not distributed with this
		@@ -17,13 +17,17 @@ from os import fspath
		from os import write as fdwrite
		from subprocess import DEVNULL
		from subprocess import PIPE
		from subprocess import CalledProcessError
		from typing import IO
		from typing import Any
		from typing import Callable
		from typing import Iterator
		from typing import Literal
		from typing import Mapping
		from typing import MutableMapping
		from typing import MutableSequence
		from typing import Sequence
		from typing import SupportsBytes
		from typing import TypeVar
		from typing import Union
		from typing import overload
		@@ -156,3 +160,120 @@ async def _passthru(in_stream: trio.abc.ReceiveStream, out_stream: IO[str]\|IO[by
		if not data:
		return
		await write(data)


		class Executor(list[Argument]):
		"""
		Manage calling executables with composable argument lists

		Subclasses may add or amend the argument list just prior to execution by implementing
		`get_arguments`.

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

		T = TypeVar("T")

		def __init__(self, *cmd: Argument):
		self[:] = cmd

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

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

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

		def __call__(
		self,
		*args: Argument,
		input: str\|bytes\|SupportsBytes\|None = None,
		deserialiser: Deserialiser[Any]\|None = None,
		query: bool = False,
		**kwargs: Any,
		) -> Any:
		"""
		Execute the configure command 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.
		"""
		assert not deserialiser or not query

		data = (
		b"" if input is None else
		input.encode() if isinstance(input, str) else
		bytes(input)
		)
		cmd = self.get_arguments(
		[self, args], kwargs,
		has_input=bool(data),
		is_query=query,
		deserialiser=deserialiser,
		)

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

		rcode = exec_io(cmd, data=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

		def get_arguments(
		self,
		cmd: Arguments,
		kwargs: MutableMapping[str, Any],
		has_input: bool,
		is_query: bool,
		deserialiser: Deserialiser[Any]\|None,
		) -> Arguments:
		"""
		Override to amend command arguments and kwargs for exec_io() prior to execution
		"""
		return cmd