14900591534

Committed 08 May 2025 06:58AM CUT coverage: 54.224%. Remained the same

Build # 14900591534

Build Type

push

github

Committed by

web-flow

Commit Message

Bump actions/download-artifact from 4.1.8 to 4.3.0 (#159)

Bumps [actions/download-artifact](https://github.com/actions/download-artifact) from 4.1.8 to 4.3.0.
- [Release notes](https://github.com/actions/download-artifact/releases)
- [Commits](https://github.com/actions/download-artifact/compare/v4.1.8...v4.3.0)

---
updated-dependencies:
- dependency-name: actions/download-artifact
  dependency-version: 4.3.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Run Details

98 of 201 branches covered (48.76%)

Branch coverage included in aggregate %.

1083 of 1977 relevant lines covered (54.78%)

5.48 hits per line

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

0.0

/exec_helpers/async_api/exec_result.py

#    Copyright 2018 - 2023 Aleksei Stepanov aka penguinolog.

#    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.

"""Async execution result.

.. versionadded:: 3.0.0
"""

from __future__ import annotations

import contextlib
import logging
import typing

if typing.TYPE_CHECKING:
    from collections.abc import AsyncIterable

from exec_helpers import exec_result  # pylint: disable=wrong-import-position

__all__ = ("ExecResult",)


class ExecResult(exec_result.ExecResult):
    """Execution result."""

    __slots__ = ()

    @staticmethod
    async def _poll_stream(  # type: ignore[override]  # pylint: disable=invalid-overridden-method
        src: AsyncIterable[bytes],
        log: logging.Logger | None = None,
        verbose: bool = False,
    ) -> list[bytes]:
        """Stream poll helper.

        :param src: Source to read from.
        :param log: Logger instance, if line per-line logging is expected.
        :param verbose: Use INFO level for logging.
        :return: Read result as a list of byte strings.
        """
        dst: list[bytes] = []
        with contextlib.suppress(IOError):
            async for line in src:
                dst.append(line)
                if log:
                    log.log(
                        level=logging.INFO if verbose else logging.DEBUG,
                        msg=line.decode("utf-8", errors="backslashreplace").rstrip(),
                    )
        return dst

    async def read_stdout(  # type: ignore[override]  # pylint: disable=invalid-overridden-method
        self,
        src: AsyncIterable[bytes] | None = None,
        log: logging.Logger | None = None,
        verbose: bool = False,
    ) -> None:
        """Read asyncio stdout transport to stdout.

        :param src: Source.
        :type src: AsyncIterable[bytes] | None
        :param log: Logger.
        :type log: logging.Logger | None
        :param verbose: Use log.info instead of log.debug.
        :type verbose: bool
        :raises RuntimeError: Exit code is already received.

        .. versionadded:: 3.0.0
        """
        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(await self._poll_stream(src, log, verbose))

    async def read_stderr(  # type: ignore[override]  # pylint: disable=invalid-overridden-method
        self,
        src: AsyncIterable[bytes] | None = None,
        log: logging.Logger | None = None,
        verbose: bool = False,
    ) -> None:
        """Read asyncio stderr transport to stdout.

        :param src: Source.
        :type src: AsyncIterable[bytes] | None
        :param log: Logger.
        :type log: logging.Logger | None
        :param verbose: Use log.info instead of log.debug.
        :type verbose: bool
        :raises RuntimeError: Exit code is already received.

        .. versionadded:: 3.0.0
        """
        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(await self._poll_stream(src, log, verbose))

1	# Copyright 2018 - 2023 Aleksei Stepanov aka penguinolog.
2
3	# Licensed under the Apache License, Version 2.0 (the "License"); you may
4	# not use this file except in compliance with the License. You may obtain
5	# a copy of the License at
6
7	# http://www.apache.org/licenses/LICENSE-2.0
8
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11	# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12	# License for the specific language governing permissions and limitations
13	# under the License.
14
15	"""Async execution result.
16
17	.. versionadded:: 3.0.0
18	"""
19
20	from __future__ import annotations	×
21
22	import contextlib	×
23	import logging	×
24	import typing	×
25
26	if typing.TYPE_CHECKING:
27	from collections.abc import AsyncIterable
28
29	from exec_helpers import exec_result # pylint: disable=wrong-import-position	×
30
31	__all__ = ("ExecResult",)	×
32
33
34	class ExecResult(exec_result.ExecResult):	×
35	"""Execution result."""
36
37	__slots__ = ()	×
38
39	@staticmethod	×
40	async def _poll_stream( # type: ignore[override] # pylint: disable=invalid-overridden-method	×
41	src: AsyncIterable[bytes],
42	log: logging.Logger \| None = None,
43	verbose: bool = False,
44	) -> list[bytes]:
45	"""Stream poll helper.
46
47	:param src: Source to read from.
48	:param log: Logger instance, if line per-line logging is expected.
49	:param verbose: Use INFO level for logging.
50	:return: Read result as a list of byte strings.
51	"""
52	dst: list[bytes] = []	×
53	with contextlib.suppress(IOError):	×
54	async for line in src:	×
55	dst.append(line)	×
56	if log:	×
57	log.log(	×
58	level=logging.INFO if verbose else logging.DEBUG,
59	msg=line.decode("utf-8", errors="backslashreplace").rstrip(),
60	)
61	return dst	×
62
63	async def read_stdout( # type: ignore[override] # pylint: disable=invalid-overridden-method	×
64	self,
65	src: AsyncIterable[bytes] \| None = None,
66	log: logging.Logger \| None = None,
67	verbose: bool = False,
68	) -> None:
69	"""Read asyncio stdout transport to stdout.
70
71	:param src: Source.
72	:type src: AsyncIterable[bytes] \| None
73	:param log: Logger.
74	:type log: logging.Logger \| None
75	:param verbose: Use log.info instead of log.debug.
76	:type verbose: bool
77	:raises RuntimeError: Exit code is already received.
78
79	.. versionadded:: 3.0.0
80	"""
81	if not src:	×
82	return	×
83	if self.timestamp:	×
84	raise RuntimeError("Final exit code received.")	×
85
86	with self.stdout_lock:	×
87	self._stdout_str = self._stdout_brief = None	×
88	self._stdout += tuple(await self._poll_stream(src, log, verbose))	×
89
90	async def read_stderr( # type: ignore[override] # pylint: disable=invalid-overridden-method	×
91	self,
92	src: AsyncIterable[bytes] \| None = None,
93	log: logging.Logger \| None = None,
94	verbose: bool = False,
95	) -> None:
96	"""Read asyncio stderr transport to stdout.
97
98	:param src: Source.
99	:type src: AsyncIterable[bytes] \| None
100	:param log: Logger.
101	:type log: logging.Logger \| None
102	:param verbose: Use log.info instead of log.debug.
103	:type verbose: bool
104	:raises RuntimeError: Exit code is already received.
105
106	.. versionadded:: 3.0.0
107	"""
108	if not src:	×
109	return	×
110	if self.timestamp:	×
111	raise RuntimeError("Final exit code received.")	×
112
113	with self.stderr_lock:	×
114	self._stderr_str = self._stderr_brief = None	×
115	self._stderr += tuple(await self._poll_stream(src, log, verbose))	×

python-useful-helpers / exec-helpers / 14900591534

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