diff --git a/README.md b/README.md index 528ec58c6..64e6d1c48 100644 --- a/README.md +++ b/README.md @@ -139,7 +139,7 @@ pypi: https://pypi.org/project/dependatool #### [envoy.base.utils](envoy.base.utils) -version: 0.5.3.dev0 +version: 0.5.3 pypi: https://pypi.org/project/envoy.base.utils @@ -156,6 +156,7 @@ pypi: https://pypi.org/project/envoy.base.utils - [orjson](https://pypi.org/project/orjson) - [packaging](https://pypi.org/project/packaging) >=23.0 - [protobuf](https://pypi.org/project/protobuf) +- [python-gnupg](https://pypi.org/project/python-gnupg) - [pytz](https://pypi.org/project/pytz) - [pyyaml](https://pypi.org/project/pyyaml) - [trycast](https://pypi.org/project/trycast) >=0.7.3 diff --git a/envoy.base.utils/VERSION b/envoy.base.utils/VERSION index aaa0831b0..be14282b7 100644 --- a/envoy.base.utils/VERSION +++ b/envoy.base.utils/VERSION @@ -1 +1 @@ -0.5.3-dev +0.5.3 diff --git a/envoy.base.utils/envoy/base/utils/BUILD b/envoy.base.utils/envoy/base/utils/BUILD index 93a556b82..794ca8f08 100644 --- a/envoy.base.utils/envoy/base/utils/BUILD +++ b/envoy.base.utils/envoy/base/utils/BUILD @@ -13,6 +13,7 @@ toolshed_library( "//deps:reqs#packaging", "//deps:reqs#pytz", "//deps:reqs#protobuf", + "//deps:reqs#python-gnupg", "//deps:reqs#pyyaml", "//deps:reqs#trycast", "//deps:reqs#zstandard", diff --git a/envoy.base.utils/envoy/base/utils/exceptions.py b/envoy.base.utils/envoy/base/utils/exceptions.py index eaed6d569..6425487eb 100644 --- a/envoy.base.utils/envoy/base/utils/exceptions.py +++ b/envoy.base.utils/envoy/base/utils/exceptions.py @@ -35,3 +35,7 @@ class PublishError(Exception): class ChecksumError(Exception): pass + + +class SignatureError(Exception): + pass diff --git a/envoy.base.utils/envoy/base/utils/fetch_runner.py b/envoy.base.utils/envoy/base/utils/fetch_runner.py index e199f2186..1be650040 100644 --- a/envoy.base.utils/envoy/base/utils/fetch_runner.py +++ b/envoy.base.utils/envoy/base/utils/fetch_runner.py @@ -2,18 +2,20 @@ import asyncio import json import hashlib +import io import os import pathlib import time from functools import cached_property -from typing import Optional +from typing import IO, Optional from urllib.parse import urlsplit import aiohttp -from aio.core.tasks import concurrent -from aio.run import runner +import gnupg # type:ignore +from aio.core.tasks import concurrent, ConcurrentExecutionError +from aio.run import runner from envoy.base import utils @@ -38,6 +40,10 @@ def excludes(self) -> list[str]: if self.args.excludes else []) + @cached_property + def gpg(self) -> gnupg.GPG: + return gnupg.GPG() + @property def headers(self) -> dict: return ( @@ -47,6 +53,11 @@ def headers(self) -> dict: Authorization=f"token {self.token}", Accept="application/octet-stream")) + @cached_property + def session(self) -> aiohttp.ClientSession: + """HTTP client session.""" + return aiohttp.ClientSession() + @property def time_elapsed(self) -> float: start = self.time_start @@ -64,11 +75,6 @@ def token(self) -> Optional[str]: elif self.args.token: return os.getenv(self.args.token) - @cached_property - def session(self) -> aiohttp.ClientSession: - """HTTP client session.""" - return aiohttp.ClientSession() - def add_arguments(self, parser) -> None: super().add_arguments(parser) parser.add_argument("downloads", help="JSON k/v of downloads/info") @@ -103,66 +109,83 @@ def add_arguments(self, parser) -> None: "--token-path", help="Path to auth token") - def download_path(self, url: str) -> Optional[pathlib.Path]: + async def cleanup(self): + await super().cleanup() + await self.session.close() + + def download_path( + self, url: str, + create: bool = True) -> Optional[pathlib.Path]: if "path" not in self.downloads[url]: return None - return self.downloads_path.joinpath( + _download_path = self.downloads_path.joinpath( self.downloads[url]["path"], self.filename(url)) + if create: + _download_path.parent.mkdir(parents=True, exist_ok=True) + return _download_path def excluded(self, url: str) -> bool: path = self.downloads[url].get("path") return bool(path and path in self.excludes) - async def fetch(self, url: str) -> tuple[str, Optional[bytes]]: + async def fetch(self, url: str) -> tuple[str, bytes]: self.log.debug( f"{self.time_elapsed} Fetching:\n" f" {url}\n") download_path = self.download_path(url) - if download_path: - download_path.parent.mkdir(parents=True, exist_ok=True) - return await self.fetch_bytes(url, path=download_path) + buffer: IO[bytes] = ( + download_path.open("wb+") + if download_path + else io.BytesIO()) + with buffer as fd: + await self.fetch_bytes(url, fd) + await self.validate(url, fd) + content: bytes = ( + fd.read() + if not download_path + else b'') + if download_path and self.args.extract_downloads: + await asyncio.to_thread( + utils.extract, + download_path.parent, + download_path) + download_path.unlink() + return url, content async def fetch_bytes( self, url: str, - path: Optional[pathlib.Path] = None) -> ( - tuple[str, Optional[bytes]]): + fd: IO[bytes]) -> None: + dest = ( + f" -> {fd.name}" + if hasattr(fd, "name") + else "") async with self.session.get(url, headers=self.headers) as response: response.raise_for_status() - if not path: - return url, await response.read() - self.log.debug( f"{self.time_elapsed} " f"Writing chunks({self.args.chunk_size}):\n" f" {url}\n" - f" -> {path}") - with path.open("wb") as f: - chunks = response.content.iter_chunked(self.args.chunk_size) - async for chunk in chunks: - f.write(chunk) - - if "checksum" in self.downloads[url]: - await self.validate_checksum(url) - - if self.args.extract_downloads: - await asyncio.to_thread(utils.extract, path.parent, path) - path.unlink() - - return url, None + f"{dest}") + chunks = response.content.iter_chunked(self.args.chunk_size) + async for chunk in chunks: + fd.write(chunk) def filename(self, url: str) -> str: parsed_url = urlsplit(url) path_parts = parsed_url.path.split("/") return path_parts[-1] - def hashed(self, content: bytes) -> str: + def hashed(self, fd: IO[bytes]) -> str: hash_object = hashlib.sha256() - hash_object.update(content) + hash_object.update(fd.read()) return hash_object.hexdigest() @runner.cleansup + @runner.catches( + (utils.exceptions.SignatureError, + ConcurrentExecutionError)) async def run(self) -> Optional[int]: result = {} downloads = concurrent( @@ -172,18 +195,17 @@ async def run(self) -> Optional[int]: if not self.excluded(url)), limit=self.args.concurrency) - async for (url, response) in downloads: + async for (url, content) in downloads: self.log.debug( f"{self.time_elapsed} " f"Received:\n" f" {url}\n") if self.args.output == "json": - result[url] = response.decode() + result[url] = content.decode() if self.args.output == "json": print(json.dumps(result)) return 0 - if not self.args.output_path: return 0 if not any(self.downloads_path.iterdir()): @@ -198,21 +220,31 @@ async def run(self) -> Optional[int]: self.downloads_path, self.args.output_path) - async def cleanup(self): - await super().cleanup() - await self.session.close() - - async def validate_checksum(self, url: str) -> None: - path = self.download_path(url) - if not path: - return + async def validate( + self, + url: str, + fd: IO[bytes]) -> None: + # These cant be run in parallel without passing + # the data rather than a buffer/file descriptor. + if "signature" in self.downloads[url]: + fd.seek(0) + await self.validate_signature(url, fd) + if "checksum" in self.downloads[url]: + fd.seek(0) + await self.validate_checksum(url, fd) + fd.seek(0) + + async def validate_checksum( + self, + url: str, + fd: IO[bytes]) -> None: + checksum = self.downloads[url]["checksum"] hashed = await asyncio.to_thread( self.hashed, - path.read_bytes()) - checksum = self.downloads[url]["checksum"] + fd) self.log.debug( f"{self.time_elapsed} " - f"Validating:\n" + f"Validating checksum:\n" f" {url}\n" f" {checksum}\n") if hashed != checksum: @@ -220,3 +252,25 @@ async def validate_checksum(self, url: str) -> None: f"Checksums do not match({url}):\n" f" expected: {checksum}\n" f" received: {hashed}") + + async def validate_signature( + self, + url: str, + fd: IO[bytes]) -> None: + digest = self.gpg.verify_file(fd) + signature = self.downloads[url]["signature"] + self.log.debug( + f"{self.time_elapsed} " + f"Validating signature:\n" + f" {url}\n" + f" {signature}\n") + if not digest.valid: + problems = "\n ".join(digest.problems) + raise utils.exceptions.SignatureError( + f"Signature not valid:\n" + f" {problems}") + if not digest.username == signature: + raise utils.exceptions.SignatureError( + f"Signature not correct:\n" + f" expected: {signature}\n" + f" received: {digest.username}") diff --git a/envoy.base.utils/setup.cfg b/envoy.base.utils/setup.cfg index baf0dde90..d2d5dc9a1 100644 --- a/envoy.base.utils/setup.cfg +++ b/envoy.base.utils/setup.cfg @@ -39,6 +39,7 @@ install_requires = orjson packaging>=23.0 protobuf + python-gnupg pytz pyyaml trycast>=0.7.3 diff --git a/envoy.base.utils/tests/test_fetch_runner.py b/envoy.base.utils/tests/test_fetch_runner.py index c6f6e38c5..1e89d4dcd 100644 --- a/envoy.base.utils/tests/test_fetch_runner.py +++ b/envoy.base.utils/tests/test_fetch_runner.py @@ -110,6 +110,23 @@ def test_fetchrunner_excludes(patches, excludes): assert "excludes" in runner.__dict__ +def test_fetchrunner_gpg(patches): + runner = utils.FetchRunner() + patched = patches( + "gnupg", + prefix="envoy.base.utils.fetch_runner") + + with patched as (m_gpg, ): + assert ( + runner.gpg + == m_gpg.GPG.return_value) + + assert ( + m_gpg.GPG.call_args + == [(), {}]) + assert "gpg" in runner.__dict__ + + @pytest.mark.parametrize("token", ["", "TOKEN"]) def test_fetchrunner_headers(patches, token): runner = utils.FetchRunner() @@ -139,6 +156,23 @@ def test_fetchrunner_headers(patches, token): assert "headers" not in runner.__dict__ +def test_fetchrunner_session(patches): + runner = utils.FetchRunner() + patched = patches( + "aiohttp", + prefix="envoy.base.utils.fetch_runner") + + with patched as (m_aiohttp, ): + assert ( + runner.session + == m_aiohttp.ClientSession.return_value) + + assert ( + m_aiohttp.ClientSession.call_args + == [(), {}]) + assert "session" in runner.__dict__ + + def test_fetchrunner_time_elapsed(patches): runner = utils.FetchRunner() patched = patches( @@ -230,23 +264,6 @@ def test_fetchrunner_token(patches, token, token_path): assert "token" not in runner.__dict__ -def test_fetchrunner_session(patches): - runner = utils.FetchRunner() - patched = patches( - "aiohttp", - prefix="envoy.base.utils.fetch_runner") - - with patched as (m_aiohttp, ): - assert ( - runner.session - == m_aiohttp.ClientSession.return_value) - - assert ( - m_aiohttp.ClientSession.call_args - == [(), {}]) - assert "session" in runner.__dict__ - - def test_fetchrunner_add_arguments(patches): runner = utils.FetchRunner() parser = MagicMock() @@ -292,7 +309,8 @@ def test_fetchrunner_add_arguments(patches): @pytest.mark.parametrize("path", [True, False]) -def test_fetchrunner_download_path(patches, path): +@pytest.mark.parametrize("create", [True, False]) +def test_fetchrunner_download_path(patches, path, create): runner = utils.FetchRunner() url = MagicMock() patched = patches( @@ -307,7 +325,7 @@ def test_fetchrunner_download_path(patches, path): (m_downloads.return_value.__getitem__.return_value .__contains__.return_value) = path assert ( - runner.download_path(url) + runner.download_path(url, create=create) == (None if not path else m_path.return_value.joinpath.return_value)) @@ -318,8 +336,9 @@ def test_fetchrunner_download_path(patches, path): assert not ( m_downloads.return_value.__getitem__.return_value .__getitem__.called) + assert ( + not m_path.return_value.joinpath.return_value.parent.mkdir.called) return - assert ( m_path.return_value.joinpath.call_args == [(m_downloads.return_value.__getitem__.return_value @@ -334,6 +353,13 @@ def test_fetchrunner_download_path(patches, path): assert ( m_downloads.return_value.__getitem__.return_value.__getitem__.call_args == [("path", ), {}]) + if create: + assert ( + m_path.return_value.joinpath.return_value.parent.mkdir.call_args + == [(), dict(parents=True, exist_ok=True)]) + else: + assert not ( + m_path.return_value.joinpath.return_value.parent.mkdir.called) @pytest.mark.parametrize("path", [True, False]) @@ -379,27 +405,41 @@ def test_fetchrunner_excluded(patches, path, contains): @pytest.mark.parametrize("path", [True, False]) -async def test_fetchrunner_fetch(patches, path): +@pytest.mark.parametrize("extract", [True, False]) +async def test_fetchrunner_fetch(patches, path, extract): runner = utils.FetchRunner() url = MagicMock() - _path = ( - MagicMock() - if path - else None) patched = patches( + "asyncio", + "io", + "utils", "FetchRunner.download_path", "FetchRunner.fetch_bytes", + "FetchRunner.validate", + ("FetchRunner.args", + dict(new_callable=PropertyMock)), ("FetchRunner.log", dict(new_callable=PropertyMock)), ("FetchRunner.time_elapsed", dict(new_callable=PropertyMock)), prefix="envoy.base.utils.fetch_runner") - with patched as (m_path, m_fetch, m_log, m_elapsed): - m_path.return_value = _path + with patched as patchy: + (m_asyncio, m_io, m_utils, m_path, m_fetch, + m_valid, m_args, m_log, m_elapsed) = patchy + if not path: + m_path.return_value = None + m_args.return_value.extract_downloads = extract + _to_thread = AsyncMock() + m_asyncio.to_thread = _to_thread + fd = (m_io.BytesIO.return_value.__enter__ + .return_value) assert ( await runner.fetch(url) - == m_fetch.return_value) + == (url, + (fd.read.return_value + if not path + else b""))) assert ( m_log.return_value.debug.call_args @@ -410,32 +450,50 @@ async def test_fetchrunner_fetch(patches, path): == [(url, ), {}]) if path: assert ( - _path.parent.mkdir.call_args - == [(), dict(parents=True, exist_ok=True)]) + m_path.return_value.open.call_args + == [("wb+", ), {}]) + assert not m_io.BytesIO.called + fd = m_path.return_value.open.return_value + assert not fd.__enter__.return_value.read.called + else: + assert ( + m_io.BytesIO.call_args + == [(), {}]) + fd = m_io.BytesIO.return_value + assert ( + fd.__enter__.return_value.read.call_args + == [(), {}]) + assert ( m_fetch.call_args - == [(url, ), dict(path=_path)]) + == [(url, fd.__enter__.return_value), {}]) + assert ( + m_valid.call_args + == [(url, fd.__enter__.return_value), {}]) + if path and extract: + assert ( + m_asyncio.to_thread.call_args + == [(m_utils.extract, + m_path.return_value.parent, + m_path.return_value), {}]) + assert ( + m_path.return_value.unlink.call_args + == [(), {}]) + else: + assert not m_asyncio.to_thread.called -@pytest.mark.parametrize("path", [True, False]) +@pytest.mark.parametrize("named", [True, False]) @pytest.mark.parametrize("checksum", [True, False]) @pytest.mark.parametrize("extract", [True, False]) -async def test_fetchrunner_fetch_bytes(patches, path, checksum, extract): +async def test_fetchrunner_fetch_bytes(patches, named, checksum, extract): runner = utils.FetchRunner() url = MagicMock() - _path = ( - MagicMock() - if path - else None) + fd = MagicMock() patched = patches( - "asyncio", - "utils", - "FetchRunner.download_path", - "FetchRunner.validate_checksum", + "hasattr", ("FetchRunner.args", dict(new_callable=PropertyMock)), - ("FetchRunner.downloads", - dict(new_callable=PropertyMock)), ("FetchRunner.log", dict(new_callable=PropertyMock)), ("FetchRunner.headers", @@ -447,8 +505,8 @@ async def test_fetchrunner_fetch_bytes(patches, path, checksum, extract): prefix="envoy.base.utils.fetch_runner") with patched as patchy: - (m_asyncio, m_utils, m_download_path, m_validate, - m_args, m_downloads, m_log, m_headers, + (m_attr, + m_args, m_log, m_headers, m_session, m_elapsed) = patchy async def _chunked(size): @@ -456,22 +514,13 @@ async def _chunked(size): for x in range(0, 3): yield f"CHUNK{x}" - _to_thread = AsyncMock() - m_asyncio.to_thread = _to_thread + m_attr.return_value = named _get = AsyncMock() m_session.return_value.get.return_value = _get response = _get.__aenter__.return_value response.content.iter_chunked = _chunked response.raise_for_status = MagicMock() - (m_downloads.return_value.__getitem__.return_value - .__contains__.return_value) = checksum - m_args.return_value.extract_downloads = extract - assert ( - await runner.fetch_bytes(url, _path) - == (url, - (response.read.return_value - if not path - else None))) + assert not await runner.fetch_bytes(url, fd) assert ( m_session.return_value.get.call_args @@ -479,50 +528,20 @@ async def _chunked(size): assert ( response.raise_for_status.call_args == [(), {}]) - if not path: - assert ( - response.read.call_args - == [(), {}]) - assert not m_log.return_value.debug.called - assert not response.content.iter_chunk.called - assert not m_downloads.return_value.__getitem__.called - assert not m_asyncio.to_thread.called - assert not m_validate.called - return - - assert not response.read.called + dest = ( + f" -> {fd.name}" + if named + else "") assert ( m_log.return_value.debug.call_args == [((f"{m_elapsed.return_value} " f"Writing chunks({m_args.return_value.chunk_size}):\n" f" {url}\n" - f" -> {_path}"), ), {}]) + f"{dest}"), ), {}]) assert ( - _path.open.call_args - == [("wb", ), {}]) - assert ( - _path.open.return_value.__enter__.return_value.write.call_args_list + fd.write.call_args_list == [[(f"CHUNK{x}", ), {}] for x in range(0, 3)]) - assert ( - m_downloads.return_value.__getitem__.call_args - == [(url, ), {}]) - if checksum: - assert ( - m_validate.call_args - == [(url, ), {}]) - else: - assert not m_validate.called - if not extract: - assert not m_asyncio.to_thread.called - assert not _path.unlink.called - return - assert ( - m_asyncio.to_thread.call_args - == [(m_utils.extract, _path.parent, _path), {}]) - assert ( - _path.unlink.call_args - == [(), {}]) def test_fetchrunner_filename(patches): @@ -551,14 +570,13 @@ def test_fetchrunner_filename(patches): def test_fetchrunner_hashed(patches): runner = utils.FetchRunner() - content = MagicMock() + fd = MagicMock() patched = patches( "hashlib", prefix="envoy.base.utils.fetch_runner") - with patched as (m_hash, ): assert ( - runner.hashed(content) + runner.hashed(fd) == m_hash.sha256.return_value.hexdigest.return_value) assert ( @@ -566,7 +584,10 @@ def test_fetchrunner_hashed(patches): == [(), {}]) assert ( m_hash.sha256.return_value.update.call_args - == [(content, ), {}]) + == [(fd.read.return_value, ), {}]) + assert ( + fd.read.call_args + == [(), {}]) assert ( m_hash.sha256.return_value.hexdigest.call_args == [(), {}]) @@ -681,18 +702,62 @@ async def _concurrent(): m_args.return_value.output_path), {}]) -@pytest.mark.parametrize("path", [True, False]) +@pytest.mark.parametrize("signature", [True, False]) +@pytest.mark.parametrize("checksum", [True, False]) +async def test_fetchrunner_validate(patches, signature, checksum): + runner = utils.FetchRunner() + url = MagicMock() + fd = MagicMock() + patched = patches( + "FetchRunner.validate_checksum", + "FetchRunner.validate_signature", + ("FetchRunner.downloads", + dict(new_callable=PropertyMock)), + prefix="envoy.base.utils.fetch_runner") + + def _contains(x): + if x == "checksum": + return checksum + if x == "signature": + return signature + + with patched as (m_checksum, m_signature, m_downloads): + (m_downloads.return_value.__getitem__ + .return_value.__contains__.side_effect) = _contains + assert not await runner.validate(url, fd) + + assert ( + (m_downloads.return_value.__getitem__ + .return_value.__contains__.call_args_list) + == [[("signature", ), {}], + [("checksum", ), {}]]) + seeks = [[(0, ), {}]] + if checksum: + seeks.append([(0, ), {}]) + assert ( + m_checksum.call_args + == [(url, fd), {}]) + else: + assert not m_checksum.called + if signature: + seeks.append([(0, ), {}]) + assert ( + m_signature.call_args + == [(url, fd), {}]) + else: + assert not m_signature.called + assert ( + fd.seek.call_args_list + == seeks) + + @pytest.mark.parametrize("matches", [True, False]) -async def test_fetchrunner_validate_checksum(patches, path, matches): +async def test_fetchrunner_validate_checksum(patches, matches): runner = utils.FetchRunner() url = MagicMock() - _path = ( - MagicMock() - if path - else None) + fd = MagicMock() patched = patches( "asyncio", - "FetchRunner.download_path", ("FetchRunner.downloads", dict(new_callable=PropertyMock)), ("FetchRunner.log", @@ -701,38 +766,26 @@ async def test_fetchrunner_validate_checksum(patches, path, matches): dict(new_callable=PropertyMock)), prefix="envoy.base.utils.fetch_runner") - with patched as (m_asyncio, m_path, m_downloads, m_log, m_elapsed): - m_path.return_value = _path + with patched as (m_asyncio, m_downloads, m_log, m_elapsed): _to_thread = AsyncMock() m_asyncio.to_thread = _to_thread _to_thread.return_value.__ne__.return_value = not matches - if path and not matches: + if not matches: with pytest.raises(utils.exceptions.ChecksumError) as e: - await runner.validate_checksum(url) + await runner.validate_checksum(url, fd) else: - assert not await runner.validate_checksum(url) + assert not await runner.validate_checksum(url, fd) - assert ( - m_path.call_args - == [(url, ), {}]) - if not path: - assert not _to_thread.called - assert not m_log.return_value.debug.called - assert not m_downloads.called - return assert ( _to_thread.call_args - == [(runner.hashed, _path.read_bytes.return_value), {}]) - assert ( - _path.read_bytes.call_args - == [(), {}]) + == [(runner.hashed, fd), {}]) configured_checksum = ( m_downloads.return_value.__getitem__.return_value .__getitem__.return_value) assert ( m_log.return_value.debug.call_args == [((f"{m_elapsed.return_value} " - f"Validating:\n" + f"Validating checksum:\n" f" {url}\n" f" {configured_checksum}\n"), ), {}]) assert ( @@ -751,3 +804,75 @@ async def test_fetchrunner_validate_checksum(patches, path, matches): == (f"Checksums do not match({url}):\n" f" expected: {configured_checksum}\n" f" received: {_to_thread.return_value}")) + + +@pytest.mark.parametrize("valid", [True, False]) +@pytest.mark.parametrize("matches", [True, False]) +async def test_fetchrunner_validate_signature(patches, iters, valid, matches): + runner = utils.FetchRunner() + url = MagicMock() + fd = MagicMock() + patched = patches( + ("FetchRunner.downloads", + dict(new_callable=PropertyMock)), + ("FetchRunner.gpg", + dict(new_callable=PropertyMock)), + ("FetchRunner.log", + dict(new_callable=PropertyMock)), + ("FetchRunner.time_elapsed", + dict(new_callable=PropertyMock)), + prefix="envoy.base.utils.fetch_runner") + + with patched as (m_downloads, m_gpg, m_log, m_elapsed): + m_gpg.return_value.verify_file.return_value.valid = valid + (m_gpg.return_value.verify_file + .return_value.username.__eq__.return_value) = matches + m_gpg.return_value.verify_file.return_value.problems = iters() + if not valid or not matches: + with pytest.raises(utils.exceptions.SignatureError) as e: + await runner.validate_signature(url, fd) + else: + assert not await runner.validate_signature(url, fd) + + signature = ( + m_downloads.return_value.__getitem__ + .return_value.__getitem__.return_value) + assert ( + m_log.return_value.debug.call_args + == [((f"{m_elapsed.return_value} " + f"Validating signature:\n" + f" {url}\n" + f" {signature}\n"), ), {}]) + assert ( + m_gpg.return_value.verify_file.call_args + == [(fd, ), {}]) + assert ( + m_downloads.return_value.__getitem__.call_args + == [(url, ), {}]) + assert ( + m_downloads.return_value.__getitem__.return_value.__getitem__.call_args + == [("signature", ), {}]) + if not valid: + assert ( + e.value.args[0] + == "Signature not valid:\n I0\n I1\n I2\n I3\n I4") + assert not ( + m_gpg.return_value.verify_file + .return_value.username.__eq__.called) + return + assert ( + m_gpg.return_value.verify_file.return_value.username.__eq__.call_args + == [((m_downloads.return_value.__getitem__ + .return_value.__getitem__.return_value), ), + {}]) + if not matches: + signature = ( + m_downloads.return_value.__getitem__ + .return_value.__getitem__.return_value) + received = ( + m_gpg.return_value.verify_file.return_value.username) + assert ( + e.value.args[0] + == ("Signature not correct:\n" + f" expected: {signature}\n" + f" received: {received}"))