python-useful-helpers / exec-helpers / 423

pending completion

Build # 423

Build Type

Pull #128

travis-ci

Committed by

web-flow

Commit Message

Update docs: expected is not optional

(cherry picked from commit 31c6a013a)
(cherry picked from commit df8a43853)
(cherry picked from commit f4908a594)

On subprocess kill write exit code to result to prevent changes (#120)

* fix missed docstring

(cherry picked from commit fb0ef72af)

Add timestamp of command start to execute_async for future processing

(cherry picked from commit b3dc16231)

More info in result: start timestamp (#124)

(cherry picked from commit 8b7a8e78e)

Always set timestamp on kill, provide spent time in str()

(cherry picked from commit 06f1650e0)

Pull Request Pull Request #128: Update docs: expected is not optional

Run Details

40 of 40 new or added lines in 4 files covered. (100.0%)

1100 of 1109 relevant lines covered (99.19%)

0.99 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.47

/exec_helpers/exec_result.py

#    Copyright 2018 Alexey Stepanov aka penguinolog.

#    Copyright 2016 Mirantis, Inc.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

"""Execution result."""

import datetime
import json
import logging
import threading
import typing

import yaml

from exec_helpers import exceptions  # pylint: disable=cyclic-import
from exec_helpers import proc_enums

__all__ = ("ExecResult",)

logger = logging.getLogger(__name__)


class ExecResult:
    """Execution result."""

    __slots__ = [
        "__cmd",
        "__stdin",
        "_stdout",
        "_stderr",
        "__exit_code",
        "__timestamp",
        "_stdout_str",
        "_stderr_str",
        "_stdout_brief",
        "_stderr_brief",
        "__stdout_lock",
        "__stderr_lock",
        "__started",
    ]

    def __init__(
        self,
        cmd: str,
        stdin: typing.Union[bytes, str, bytearray, None] = None,
        stdout: typing.Optional[typing.Iterable[bytes]] = None,
        stderr: typing.Optional[typing.Iterable[bytes]] = None,
        exit_code: typing.Union[int, proc_enums.ExitCodes] = proc_enums.INVALID,
        *,
        started: typing.Optional[datetime.datetime] = None
    ) -> None:
        """Command execution result.

        :param cmd: command
        :type cmd: str
        :param stdin: string STDIN
        :type stdin: typing.Union[bytes, str, bytearray, None]
        :param stdout: binary STDOUT
        :type stdout: typing.Optional[typing.Iterable[bytes]]
        :param stderr: binary STDERR
        :type stderr: typing.Optional[typing.Iterable[bytes]]
        :param exit_code: Exit code. If integer - try to convert to BASH enum.
        :type exit_code: typing.Union[int, proc_enums.ExitCodes]
        :param started: Timestamp of command start
        :type started: typing.Optional[datetime.datetime]
        """
        self.__stdout_lock = threading.RLock()
        self.__stderr_lock = threading.RLock()

        self.__cmd = cmd
        if isinstance(stdin, bytes):
            stdin = self._get_str_from_bin(bytearray(stdin))
        elif isinstance(stdin, bytearray):
            stdin = self._get_str_from_bin(stdin)
        self.__stdin = stdin  # type: typing.Optional[str]

        if stdout is not None:
            self._stdout = tuple(stdout)  # type: typing.Tuple[bytes, ...]
        else:
            self._stdout = ()

        if stderr is not None:
            self._stderr = tuple(stderr)  # type: typing.Tuple[bytes, ...]
        else:
            self._stderr = ()

        self.__exit_code = proc_enums.INVALID  # type: typing.Union[int, proc_enums.ExitCodes]
        self.__timestamp = None  # type: typing.Optional[datetime.datetime]
        self.exit_code = exit_code

        self.__started = started  # type: typing.Optional[datetime.datetime]

        # By default is none:
        self._stdout_str = None
        self._stderr_str = None
        self._stdout_brief = None
        self._stderr_brief = None

    @property
    def stdout_lock(self) -> threading.RLock:
        """Lock object for thread-safe operation.

        :return: internal lock for stdout
        :rtype: threading.RLock

        .. versionadded:: 2.2.0
        """
        return self.__stdout_lock

    @property
    def stderr_lock(self) -> threading.RLock:
        """Lock object for thread-safe operation.

        :return: internal lock for stderr
        :rtype: threading.RLock

        .. versionadded:: 2.2.0
        """
        return self.__stderr_lock

    @property
    def timestamp(self) -> typing.Optional[datetime.datetime]:
        """Timestamp.

        :return: exit code timestamp
        :rtype: typing.Optional[datetime.datetime]
        """
        return self.__timestamp

    def set_timestamp(self) -> None:
        """Set timestamp if empty.

        This will block future object changes.

        .. versionadded:: 2.11.0
        """
        if self.timestamp is None:
            self.__timestamp = datetime.datetime.utcnow()

    @staticmethod
    def _get_bytearray_from_array(src: typing.Iterable[bytes]) -> bytearray:
        """Get bytearray from array of bytes blocks.

        :param src: source to process
        :type src: typing.List[bytes]
        :return: bytearray
        :rtype: bytearray
        """
        return bytearray(b"".join(src))

    @staticmethod
    def _get_str_from_bin(src: bytearray) -> str:
        """Join data in list to the string.

        :param src: source to process
        :type src: bytearray
        :return: decoded string
        :rtype: str
        """
        return src.strip().decode(encoding="utf-8", errors="backslashreplace")

    @classmethod
    def _get_brief(cls, data: typing.Tuple[bytes, ...]) -> str:
        """Get brief output: 7 lines maximum (3 first + ... + 3 last).

        :param data: source to process
        :type data: typing.Tuple[bytes, ...]
        :return: brief from source
        :rtype: str
        """
        if len(data) <= 7:
            src = data  # type: typing.Tuple[bytes, ...]
        else:
            src = data[:3] + (b"...\n",) + data[-3:]
        return cls._get_str_from_bin(cls._get_bytearray_from_array(src))

    @property
    def cmd(self) -> str:
        """Executed command.

        :rtype: str
        """
        return self.__cmd

    @property
    def stdin(self) -> typing.Optional[str]:
        """Stdin input as string.

        :rtype: typing.Optional[str]
        """
        return self.__stdin

    @property
    def stdout(self) -> typing.Tuple[bytes, ...]:
        """Stdout output as list of binaries.

        :rtype: typing.Tuple[bytes, ...]
        """
        return self._stdout

    @property
    def stderr(self) -> typing.Tuple[bytes, ...]:
        """Stderr output as list of binaries.

        :rtype: typing.Tuple[bytes, ...]
        """
        return self._stderr

    @staticmethod
    def _poll_stream(
        src: typing.Iterable[bytes], log: typing.Optional[logging.Logger] = None, verbose: bool = False
    ) -> typing.List[bytes]:
        dst = []
        try:
            for line in src:
                dst.append(line)
                if log:
                    log.log(  # type: ignore
                        level=logging.INFO if verbose else logging.DEBUG,
                        msg=line.decode("utf-8", errors="backslashreplace").rstrip(),
                    )
        except IOError:
            pass
        return dst

    def read_stdout(
        self,
        src: typing.Optional[typing.Iterable[bytes]] = None,
        log: typing.Optional[logging.Logger] = None,
        verbose: bool = False,
    ) -> None:
        """Read stdout file-like object to stdout.

        :param src: source
        :type src: typing.Optional[typing.Iterable]
        :param log: logger
        :type log: typing.Optional[logging.Logger]
        :param verbose: use log.info instead of log.debug
        :type verbose: bool
        :raises RuntimeError: Exit code is already received

        .. versionchanged:: 1.2.0 - src can be None
        """
        if not src:
            return
        if self.timestamp:
            raise RuntimeError("Final exit code received.")

        with self.stdout_lock:
            self._stdout_str = self._stdout_brief = None
            self._stdout += tuple(self._poll_stream(src, log, verbose))

    def read_stderr(
        self,
        src: typing.Optional[typing.Iterable[bytes]] = None,
        log: typing.Optional[logging.Logger] = None,
        verbose: bool = False,
    ) -> None:
        """Read stderr file-like object to stdout.

        :param src: source
        :type src: typing.Optional[typing.Iterable]
        :param log: logger
        :type log: typing.Optional[logging.Logger]
        :param verbose: use log.info instead of log.debug
        :type verbose: bool
        :raises RuntimeError: Exit code is already received

        .. versionchanged:: 1.2.0 - src can be None
        """
        if not src:
            return
        if self.timestamp:
            raise RuntimeError("Final exit code received.")

        with self.stderr_lock:
            self._stderr_str = self._stderr_brief = None
            self._stderr += tuple(self._poll_stream(src, log, verbose))

    @property
    def stdout_bin(self) -> bytearray:
        """Stdout in binary format.

        Sometimes logging is used to log binary objects too (example: Session),
        and for debug purposes we can use this as data source.
        :rtype: bytearray
        """
        with self.stdout_lock:
            return self._get_bytearray_from_array(self.stdout)

    @property
    def stderr_bin(self) -> bytearray:
        """Stderr in binary format.

        :rtype: bytearray
        """
        with self.stderr_lock:
            return self._get_bytearray_from_array(self.stderr)

    @property
    def stdout_str(self) -> str:
        """Stdout output as string.

        :rtype: str
        """
        with self.stdout_lock:
            if self._stdout_str is None:
                self._stdout_str = self._get_str_from_bin(self.stdout_bin)  # type: ignore
            return self._stdout_str  # type: ignore

    @property
    def stderr_str(self) -> str:
        """Stderr output as string.

        :rtype: str
        """
        with self.stderr_lock:
            if self._stderr_str is None:
                self._stderr_str = self._get_str_from_bin(self.stderr_bin)  # type: ignore
            return self._stderr_str  # type: ignore

    @property
    def stdout_brief(self) -> str:
        """Brief stdout output (mostly for exceptions).

        :rtype: str
        """
        with self.stdout_lock:
            if self._stdout_brief is None:
                self._stdout_brief = self._get_brief(self.stdout)  # type: ignore
            return self._stdout_brief  # type: ignore

    @property
    def stderr_brief(self) -> str:
        """Brief stderr output (mostly for exceptions).

        :rtype: str
        """
        with self.stderr_lock:
            if self._stderr_brief is None:
                self._stderr_brief = self._get_brief(self.stderr)  # type: ignore
            return self._stderr_brief  # type: ignore

    @property
    def exit_code(self) -> typing.Union[int, proc_enums.ExitCodes]:
        """Return(exit) code of command.

        :return: exit code
        :rtype: typing.Union[int, proc_enums.ExitCodes]
        """
        return self.__exit_code

    @exit_code.setter
    def exit_code(self, new_val: typing.Union[int, proc_enums.ExitCodes]) -> None:
        """Return(exit) code of command.

        :param new_val: new exit code
        :type new_val: typing.Union[int, proc_enums.ExitCodes]
        :raises RuntimeError: Exit code is already received
        :raises TypeError: exit code is not int instance

        If valid exit code is set - object became read-only.
        """
        if self.timestamp:
            raise RuntimeError("Exit code is already received.")
        if not isinstance(new_val, int):
            raise TypeError("Exit code is strictly int, received: {code!r}".format(code=new_val))
        with self.stdout_lock, self.stderr_lock:
            self.__exit_code = proc_enums.exit_code_to_enum(new_val)
            if self.__exit_code != proc_enums.INVALID:
                self.__timestamp = datetime.datetime.utcnow()

    @property
    def started(self) -> typing.Optional[datetime.datetime]:
        """Timestamp of command start.

        .. versionadded:: 2.11.0
        """
        return self.__started

    def __deserialize(self, fmt: str) -> typing.Any:
        """Deserialize stdout as data format.

        :param fmt: format to decode from
        :type fmt: str
        :return: decoded object
        :rtype: typing.Any
        :raises NotImplementedError: fmt deserialization not implemented
        :raises DeserializeValueError: Not valid source format
        """
        try:
            if fmt == "json":
                return json.loads(self.stdout_str, encoding="utf-8")
            if fmt == "yaml":
                return yaml.safe_load(self.stdout_str)
        except Exception as e:
            tmpl = "{{self.cmd}} stdout is not valid {fmt}:\n" "{{stdout!r}}\n".format(fmt=fmt)
            logger.exception(tmpl.format(self=self, stdout=self.stdout_str))  # pylint: disable=logging-not-lazy

            raise exceptions.DeserializeValueError(tmpl.format(self=self, stdout=self.stdout_brief)).with_traceback(
                e.__traceback__
            ) from e

        msg = "{fmt} deserialize target is not implemented".format(fmt=fmt)
        logger.error(msg)
        raise NotImplementedError(msg)

    @property
    def stdout_json(self) -> typing.Any:
        """JSON from stdout.

        :rtype: typing.Any
        """
        with self.stdout_lock:
            return self.__deserialize(fmt="json")

    @property
    def stdout_yaml(self) -> typing.Any:
        """YAML from stdout.

        :rtype: typing.Any
        """
        with self.stdout_lock:
            return self.__deserialize(fmt="yaml")

    def __dir__(self) -> typing.List[str]:
        """Override dir for IDE and as source for getitem checks."""
        return [
            "cmd",
            "stdout",
            "stderr",
            "exit_code",
            "stdout_bin",
            "stderr_bin",
            "stdout_str",
            "stderr_str",
            "stdout_brief",
            "stderr_brief",
            "stdout_json",
            "stdout_yaml",
            "lock",
        ]

    def __getitem__(self, item: str) -> typing.Any:
        """Dict like get data.

        :param item: key
        :type item: str
        :return: item if attribute exists
        :rtype: typing.Any
        :raises IndexError: no attribute exists or not allowed to get (not in dir())
        """
        if item in dir(self):
            return getattr(self, item)
        raise IndexError('"{item}" not found in {dir}'.format(item=item, dir=dir(self)))

    def __repr__(self) -> str:
        """Representation for debugging."""
        if self.started:
            started = ", started={self.started},\n".format(self=self)
        else:
            started = ""
        return (
            "{cls}(cmd={self.cmd!r}, stdout={self.stdout}, stderr={self.stderr}, "
            "exit_code={self.exit_code!s}{started},)".format(cls=self.__class__.__name__, self=self, started=started)
        )

    def __str__(self) -> str:
        """Representation for logging."""
        if self.started:
            started = "\tstarted={started},\n".format(started=self.started.strftime('%Y-%m-%d %H:%M:%S'))
            if self.timestamp:
                _spent = (self.timestamp - self.started).seconds
                spent = "\tspent={hours:02d}:{minutes:02d}:{seconds:02d},\n".format(
                    hours=_spent // (60 * 60),
                    minutes=_spent // 60,
                    seconds=_spent % 60
                )
            else:
                spent = ""
        else:
            started = ""
            spent = ""
        return (
            "{cls}(\n\tcmd={cmd!r},"
            "\n\t stdout=\n'{stdout_brief}',"
            "\n\tstderr=\n'{stderr_brief}', "
            "\n\texit_code={exit_code!s},"
            "\n{started}{spent})".format(
                cls=self.__class__.__name__,
                cmd=self.cmd,
                stdout_brief=self.stdout_brief,
                stderr_brief=self.stderr_brief,
                exit_code=self.exit_code,
                started=started,
                spent=spent
            )
        )

    def __eq__(self, other: typing.Any) -> bool:
        """Comparision."""
        return (
            self.__class__ is other.__class__
            or issubclass(self.__class__, other.__class__)
            or issubclass(other.__class__, self.__class__)
        ) and (
            self.cmd == other.cmd
            and self.stdin == other.stdin
            and self.stdout == other.stdout
            and self.stderr == other.stderr
            and self.exit_code == other.exit_code
        )

    def __ne__(self, other: typing.Any) -> bool:
        """Comparision."""
        return not self.__eq__(other)

    def __hash__(self) -> int:
        """Hash for usage as dict key and in sets."""
        return hash((self.__class__, self.cmd, self.stdin, self.stdout, self.stderr, self.exit_code))

1	# Copyright 2018 Alexey Stepanov aka penguinolog.
2
3	# Copyright 2016 Mirantis, Inc.
4	#
5	# Licensed under the Apache License, Version 2.0 (the "License"); you may
6	# not use this file except in compliance with the License. You may obtain
7	# a copy of the License at
8	#
9	# http://www.apache.org/licenses/LICENSE-2.0
10	#
11	# Unless required by applicable law or agreed to in writing, software
12	# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13	# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14	# License for the specific language governing permissions and limitations
15	# under the License.
16
17	"""Execution result."""	1✔
18
19	import datetime	1✔
20	import json	1✔
21	import logging	1✔
22	import threading	1✔
23	import typing	1✔
24
25	import yaml	1✔
26
27	from exec_helpers import exceptions # pylint: disable=cyclic-import	1✔
28	from exec_helpers import proc_enums	1✔
29
30	__all__ = ("ExecResult",)	1✔
31
32	logger = logging.getLogger(__name__)	1✔
33
34
35	class ExecResult:	1✔
36	"""Execution result."""
37
38	__slots__ = [	1✔
39	"__cmd",
40	"__stdin",
41	"_stdout",
42	"_stderr",
43	"__exit_code",
44	"__timestamp",
45	"_stdout_str",
46	"_stderr_str",
47	"_stdout_brief",
48	"_stderr_brief",
49	"__stdout_lock",
50	"__stderr_lock",
51	"__started",
52	]
53
54	def __init__(	1✔
55	self,
56	cmd: str,
57	stdin: typing.Union[bytes, str, bytearray, None] = None,
58	stdout: typing.Optional[typing.Iterable[bytes]] = None,
59	stderr: typing.Optional[typing.Iterable[bytes]] = None,
60	exit_code: typing.Union[int, proc_enums.ExitCodes] = proc_enums.INVALID,
61	*,
62	started: typing.Optional[datetime.datetime] = None
63	) -> None:
64	"""Command execution result.
65
66	:param cmd: command
67	:type cmd: str
68	:param stdin: string STDIN
69	:type stdin: typing.Union[bytes, str, bytearray, None]
70	:param stdout: binary STDOUT
71	:type stdout: typing.Optional[typing.Iterable[bytes]]
72	:param stderr: binary STDERR
73	:type stderr: typing.Optional[typing.Iterable[bytes]]
74	:param exit_code: Exit code. If integer - try to convert to BASH enum.
75	:type exit_code: typing.Union[int, proc_enums.ExitCodes]
76	:param started: Timestamp of command start
77	:type started: typing.Optional[datetime.datetime]
78	"""
79	self.__stdout_lock = threading.RLock()	1✔
80	self.__stderr_lock = threading.RLock()	1✔
81
82	self.__cmd = cmd	1✔
83	if isinstance(stdin, bytes):	1✔
84	stdin = self._get_str_from_bin(bytearray(stdin))	1✔
85	elif isinstance(stdin, bytearray):	1✔
86	stdin = self._get_str_from_bin(stdin)	1✔
87	self.__stdin = stdin # type: typing.Optional[str]	1✔
88
89	if stdout is not None:	1✔
90	self._stdout = tuple(stdout) # type: typing.Tuple[bytes, ...]	1✔
91	else:
92	self._stdout = ()	1✔
93
94	if stderr is not None:	1✔
95	self._stderr = tuple(stderr) # type: typing.Tuple[bytes, ...]	1✔
96	else:
97	self._stderr = ()	1✔
98
99	self.__exit_code = proc_enums.INVALID # type: typing.Union[int, proc_enums.ExitCodes]	1✔
100	self.__timestamp = None # type: typing.Optional[datetime.datetime]	1✔
101	self.exit_code = exit_code	1✔
102
103	self.__started = started # type: typing.Optional[datetime.datetime]	1✔
104
105	# By default is none:
106	self._stdout_str = None	1✔
107	self._stderr_str = None	1✔
108	self._stdout_brief = None	1✔
109	self._stderr_brief = None	1✔
110
111	@property	1✔
112	def stdout_lock(self) -> threading.RLock:	1✔
113	"""Lock object for thread-safe operation.
114
115	:return: internal lock for stdout
116	:rtype: threading.RLock
117
118	.. versionadded:: 2.2.0
119	"""
120	return self.__stdout_lock	1✔
121
122	@property	1✔
123	def stderr_lock(self) -> threading.RLock:	1✔
124	"""Lock object for thread-safe operation.
125
126	:return: internal lock for stderr
127	:rtype: threading.RLock
128
129	.. versionadded:: 2.2.0
130	"""
131	return self.__stderr_lock	1✔
132
133	@property	1✔
134	def timestamp(self) -> typing.Optional[datetime.datetime]:	1✔
135	"""Timestamp.
136
137	:return: exit code timestamp
138	:rtype: typing.Optional[datetime.datetime]
139	"""
140	return self.__timestamp	1✔
141
142	def set_timestamp(self) -> None:	1✔
143	"""Set timestamp if empty.
144
145	This will block future object changes.
146
147	.. versionadded:: 2.11.0
148	"""
149	if self.timestamp is None:	1✔
150	self.__timestamp = datetime.datetime.utcnow()	1✔
151
152	@staticmethod	1✔
153	def _get_bytearray_from_array(src: typing.Iterable[bytes]) -> bytearray:	1✔
154	"""Get bytearray from array of bytes blocks.
155
156	:param src: source to process
157	:type src: typing.List[bytes]
158	:return: bytearray
159	:rtype: bytearray
160	"""
161	return bytearray(b"".join(src))	1✔
162
163	@staticmethod	1✔
164	def _get_str_from_bin(src: bytearray) -> str:	1✔
165	"""Join data in list to the string.
166
167	:param src: source to process
168	:type src: bytearray
169	:return: decoded string
170	:rtype: str
171	"""
172	return src.strip().decode(encoding="utf-8", errors="backslashreplace")	1✔
173
174	@classmethod	1✔
175	def _get_brief(cls, data: typing.Tuple[bytes, ...]) -> str:	1✔
176	"""Get brief output: 7 lines maximum (3 first + ... + 3 last).
177
178	:param data: source to process
179	:type data: typing.Tuple[bytes, ...]
180	:return: brief from source
181	:rtype: str
182	"""
183	if len(data) <= 7:	1✔
184	src = data # type: typing.Tuple[bytes, ...]	1✔
185	else:
186	src = data[:3] + (b"...\n",) + data[-3:]	1✔
187	return cls._get_str_from_bin(cls._get_bytearray_from_array(src))	1✔
188
189	@property	1✔
190	def cmd(self) -> str:	1✔
191	"""Executed command.
192
193	:rtype: str
194	"""
195	return self.__cmd	1✔
196
197	@property	1✔
198	def stdin(self) -> typing.Optional[str]:	1✔
199	"""Stdin input as string.
200
201	:rtype: typing.Optional[str]
202	"""
203	return self.__stdin	1✔
204
205	@property	1✔
206	def stdout(self) -> typing.Tuple[bytes, ...]:	1✔
207	"""Stdout output as list of binaries.
208
209	:rtype: typing.Tuple[bytes, ...]
210	"""
211	return self._stdout	1✔
212
213	@property	1✔
214	def stderr(self) -> typing.Tuple[bytes, ...]:	1✔
215	"""Stderr output as list of binaries.
216
217	:rtype: typing.Tuple[bytes, ...]
218	"""
219	return self._stderr	1✔
220
221	@staticmethod	1✔
222	def _poll_stream(	1✔
223	src: typing.Iterable[bytes], log: typing.Optional[logging.Logger] = None, verbose: bool = False
224	) -> typing.List[bytes]:
225	dst = []	1✔
226	try:	1✔
227	for line in src:	1✔
228	dst.append(line)	1✔
229	if log:	1✔
230	log.log( # type: ignore	1✔
231	level=logging.INFO if verbose else logging.DEBUG,
232	msg=line.decode("utf-8", errors="backslashreplace").rstrip(),
233	)
234	except IOError:	1✔
235	pass	1✔
236	return dst	1✔
237
238	def read_stdout(	1✔
239	self,
240	src: typing.Optional[typing.Iterable[bytes]] = None,
241	log: typing.Optional[logging.Logger] = None,
242	verbose: bool = False,
243	) -> None:
244	"""Read stdout file-like object to stdout.
245
246	:param src: source
247	:type src: typing.Optional[typing.Iterable]
248	:param log: logger
249	:type log: typing.Optional[logging.Logger]
250	:param verbose: use log.info instead of log.debug
251	:type verbose: bool
252	:raises RuntimeError: Exit code is already received
253
254	.. versionchanged:: 1.2.0 - src can be None
255	"""
256	if not src:	1✔
257	return	1✔
258	if self.timestamp:	1✔
259	raise RuntimeError("Final exit code received.")	1✔
260
261	with self.stdout_lock:	1✔
262	self._stdout_str = self._stdout_brief = None	1✔
263	self._stdout += tuple(self._poll_stream(src, log, verbose))	1✔
264
265	def read_stderr(	1✔
266	self,
267	src: typing.Optional[typing.Iterable[bytes]] = None,
268	log: typing.Optional[logging.Logger] = None,
269	verbose: bool = False,
270	) -> None:
271	"""Read stderr file-like object to stdout.
272
273	:param src: source
274	:type src: typing.Optional[typing.Iterable]
275	:param log: logger
276	:type log: typing.Optional[logging.Logger]
277	:param verbose: use log.info instead of log.debug
278	:type verbose: bool
279	:raises RuntimeError: Exit code is already received
280
281	.. versionchanged:: 1.2.0 - src can be None
282	"""
283	if not src:	1✔
284	return	1✔
285	if self.timestamp:	1✔
286	raise RuntimeError("Final exit code received.")	1✔
287
288	with self.stderr_lock:	1✔
289	self._stderr_str = self._stderr_brief = None	1✔
290	self._stderr += tuple(self._poll_stream(src, log, verbose))	1✔
291
292	@property	1✔
293	def stdout_bin(self) -> bytearray:	1✔
294	"""Stdout in binary format.
295
296	Sometimes logging is used to log binary objects too (example: Session),
297	and for debug purposes we can use this as data source.
298	:rtype: bytearray
299	"""
300	with self.stdout_lock:	1✔
301	return self._get_bytearray_from_array(self.stdout)	1✔
302
303	@property	1✔
304	def stderr_bin(self) -> bytearray:	1✔
305	"""Stderr in binary format.
306
307	:rtype: bytearray
308	"""
309	with self.stderr_lock:	1✔
310	return self._get_bytearray_from_array(self.stderr)	1✔
311
312	@property	1✔
313	def stdout_str(self) -> str:	1✔
314	"""Stdout output as string.
315
316	:rtype: str
317	"""
318	with self.stdout_lock:	1✔
319	if self._stdout_str is None:	1✔
320	self._stdout_str = self._get_str_from_bin(self.stdout_bin) # type: ignore	1✔
321	return self._stdout_str # type: ignore	1✔
322
323	@property	1✔
324	def stderr_str(self) -> str:	1✔
325	"""Stderr output as string.
326
327	:rtype: str
328	"""
329	with self.stderr_lock:	1✔
330	if self._stderr_str is None:	1✔
331	self._stderr_str = self._get_str_from_bin(self.stderr_bin) # type: ignore	1✔
332	return self._stderr_str # type: ignore	1✔
333
334	@property	1✔
335	def stdout_brief(self) -> str:	1✔
336	"""Brief stdout output (mostly for exceptions).
337
338	:rtype: str
339	"""
340	with self.stdout_lock:	1✔
341	if self._stdout_brief is None:	1✔
342	self._stdout_brief = self._get_brief(self.stdout) # type: ignore	1✔
343	return self._stdout_brief # type: ignore	1✔
344
345	@property	1✔
346	def stderr_brief(self) -> str:	1✔
347	"""Brief stderr output (mostly for exceptions).
348
349	:rtype: str
350	"""
351	with self.stderr_lock:	1✔
352	if self._stderr_brief is None:	1✔
353	self._stderr_brief = self._get_brief(self.stderr) # type: ignore	1✔
354	return self._stderr_brief # type: ignore	1✔
355
356	@property	1✔
357	def exit_code(self) -> typing.Union[int, proc_enums.ExitCodes]:	1✔
358	"""Return(exit) code of command.
359
360	:return: exit code
361	:rtype: typing.Union[int, proc_enums.ExitCodes]
362	"""
363	return self.__exit_code	1✔
364
365	@exit_code.setter	1✔
366	def exit_code(self, new_val: typing.Union[int, proc_enums.ExitCodes]) -> None:	1✔
367	"""Return(exit) code of command.
368
369	:param new_val: new exit code
370	:type new_val: typing.Union[int, proc_enums.ExitCodes]
371	:raises RuntimeError: Exit code is already received
372	:raises TypeError: exit code is not int instance
373
374	If valid exit code is set - object became read-only.
375	"""
376	if self.timestamp:	1✔
377	raise RuntimeError("Exit code is already received.")	1✔
378	if not isinstance(new_val, int):	1✔
379	raise TypeError("Exit code is strictly int, received: {code!r}".format(code=new_val))	1✔
380	with self.stdout_lock, self.stderr_lock:	1✔
381	self.__exit_code = proc_enums.exit_code_to_enum(new_val)	1✔
382	if self.__exit_code != proc_enums.INVALID:	1✔
383	self.__timestamp = datetime.datetime.utcnow()	1✔
384
385	@property	1✔
386	def started(self) -> typing.Optional[datetime.datetime]:	1✔
387	"""Timestamp of command start.
388
389	.. versionadded:: 2.11.0
390	"""
391	return self.__started	1✔
392
393	def __deserialize(self, fmt: str) -> typing.Any:	1✔
394	"""Deserialize stdout as data format.
395
396	:param fmt: format to decode from
397	:type fmt: str
398	:return: decoded object
399	:rtype: typing.Any
400	:raises NotImplementedError: fmt deserialization not implemented
401	:raises DeserializeValueError: Not valid source format
402	"""
403	try:	1✔
404	if fmt == "json":	1✔
405	return json.loads(self.stdout_str, encoding="utf-8")	1✔
406	if fmt == "yaml":	1✔
407	return yaml.safe_load(self.stdout_str)	1✔
408	except Exception as e:	1✔
409	tmpl = "{{self.cmd}} stdout is not valid {fmt}:\n" "{{stdout!r}}\n".format(fmt=fmt)	1✔
410	logger.exception(tmpl.format(self=self, stdout=self.stdout_str)) # pylint: disable=logging-not-lazy	1✔
411
412	raise exceptions.DeserializeValueError(tmpl.format(self=self, stdout=self.stdout_brief)).with_traceback(	1✔
413	e.__traceback__
414	) from e
415
416	msg = "{fmt} deserialize target is not implemented".format(fmt=fmt)	1✔
417	logger.error(msg)	1✔
418	raise NotImplementedError(msg)	1✔
419
420	@property	1✔
421	def stdout_json(self) -> typing.Any:	1✔
422	"""JSON from stdout.
423
424	:rtype: typing.Any
425	"""
426	with self.stdout_lock:	1✔
427	return self.__deserialize(fmt="json")	1✔
428
429	@property	1✔
430	def stdout_yaml(self) -> typing.Any:	1✔
431	"""YAML from stdout.
432
433	:rtype: typing.Any
434	"""
435	with self.stdout_lock:	1✔
436	return self.__deserialize(fmt="yaml")	1✔
437
438	def __dir__(self) -> typing.List[str]:	1✔
439	"""Override dir for IDE and as source for getitem checks."""
440	return [	1✔
441	"cmd",
442	"stdout",
443	"stderr",
444	"exit_code",
445	"stdout_bin",
446	"stderr_bin",
447	"stdout_str",
448	"stderr_str",
449	"stdout_brief",
450	"stderr_brief",
451	"stdout_json",
452	"stdout_yaml",
453	"lock",
454	]
455
456	def __getitem__(self, item: str) -> typing.Any:	1✔
457	"""Dict like get data.
458
459	:param item: key
460	:type item: str
461	:return: item if attribute exists
462	:rtype: typing.Any
463	:raises IndexError: no attribute exists or not allowed to get (not in dir())
464	"""
465	if item in dir(self):	1✔
466	return getattr(self, item)	1✔
467	raise IndexError('"{item}" not found in {dir}'.format(item=item, dir=dir(self)))	1✔
468
469	def __repr__(self) -> str:
470	"""Representation for debugging."""
471	if self.started:
472	started = ", started={self.started},\n".format(self=self)
473	else:
474	started = ""
475	return (
476	"{cls}(cmd={self.cmd!r}, stdout={self.stdout}, stderr={self.stderr}, "
477	"exit_code={self.exit_code!s}{started},)".format(cls=self.__class__.__name__, self=self, started=started)
478	)
479
480	def __str__(self) -> str:	1✔
481	"""Representation for logging."""
482	if self.started:	1✔
483	started = "\tstarted={started},\n".format(started=self.started.strftime('%Y-%m-%d %H:%M:%S'))	1✔
484	if self.timestamp:	1✔
485	_spent = (self.timestamp - self.started).seconds	1✔
486	spent = "\tspent={hours:02d}:{minutes:02d}:{seconds:02d},\n".format(	1✔
487	hours=_spent // (60 * 60),
488	minutes=_spent // 60,
489	seconds=_spent % 60
490	)
491	else:
492	spent = ""	×
493	else:
494	started = ""	1✔
495	spent = ""	1✔
496	return (	1✔
497	"{cls}(\n\tcmd={cmd!r},"
498	"\n\t stdout=\n'{stdout_brief}',"
499	"\n\tstderr=\n'{stderr_brief}', "
500	"\n\texit_code={exit_code!s},"
501	"\n{started}{spent})".format(
502	cls=self.__class__.__name__,
503	cmd=self.cmd,
504	stdout_brief=self.stdout_brief,
505	stderr_brief=self.stderr_brief,
506	exit_code=self.exit_code,
507	started=started,
508	spent=spent
509	)
510	)
511
512	def __eq__(self, other: typing.Any) -> bool:	1✔
513	"""Comparision."""
514	return (	1✔
515	self.__class__ is other.__class__
516	or issubclass(self.__class__, other.__class__)
517	or issubclass(other.__class__, self.__class__)
518	) and (
519	self.cmd == other.cmd
520	and self.stdin == other.stdin
521	and self.stdout == other.stdout
522	and self.stderr == other.stderr
523	and self.exit_code == other.exit_code
524	)
525
526	def __ne__(self, other: typing.Any) -> bool:	1✔
527	"""Comparision."""
528	return not self.__eq__(other)	1✔
529
530	def __hash__(self) -> int:	1✔
531	"""Hash for usage as dict key and in sets."""
532	return hash((self.__class__, self.cmd, self.stdin, self.stdout, self.stderr, self.exit_code))	1✔

python-useful-helpers / exec-helpers / 423

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous