fix: async utils: aio_run_command(): check returncode (#1181)

Raise subprocess.CalledProcessError if subprocess fails with a non-zero
return code.

Fixes: #1177
Signed-off-by: John Andersen <[email protected]>

Co-authored-by: Terri Oda <[email protected]>

Author: pdxjohnny

.github/workflows/pythonapp.yml

@@ -282,6 +282,7 @@ jobs:
           test/test_output_engine.py
           test/test_util.py
           test/test_condensed_downloads.py
+          test/test_async_utils.py
           test/test_package_list_parser.py
           test/test_merge.py
       - name: Run Synchronous test

cve_bin_tool/async_utils.py

@@ -9,6 +9,7 @@
 import itertools
 import os
 import shutil
+import subprocess
 import sys
 import tempfile
 from functools import partial, wraps
@@ -54,6 +55,10 @@ async def aio_run_command(args):
         stderr=asyncio.subprocess.PIPE,
     )
     stdout, stderr = await process.communicate()
+    if process.returncode != 0:
+        raise subprocess.CalledProcessError(
+            args, process.returncode, output=stdout, stderr=stderr
+        )
     return stdout, stderr  # binary encoded
 
 

test/test_async_utils.py

@@ -0,0 +1,42 @@
+# Copyright (C) 2021 Intel Corporation
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+"""
+CVE-bin-tool async util tests
+"""
+import dataclasses
+import subprocess
+import unittest.mock
+from typing import Callable, Coroutine, Tuple
+
+import pytest
+
+from cve_bin_tool.async_utils import aio_run_command
+
+
+@dataclasses.dataclass
+class FakeProcess:
+    returncode: int
+
+    async def communicate(self) -> Tuple[bytes, bytes]:
+        return b"", b""
+
+
+def mkexec(returncode: int) -> Callable[..., Coroutine[None, None, FakeProcess]]:
+    async def return_fake_process(*args, **kwargs) -> FakeProcess:
+        return FakeProcess(returncode=returncode)
+
+    return return_fake_process
+
+
+@pytest.mark.asyncio
+async def test_aio_run_command_success():
+    with unittest.mock.patch("asyncio.create_subprocess_exec", new=mkexec(0)):
+        await aio_run_command(("echo", "hello"))
+
+
+@pytest.mark.asyncio
+async def test_aio_run_command_returncode_non_zero():
+    with unittest.mock.patch("asyncio.create_subprocess_exec", new=mkexec(1)):
+        with pytest.raises(subprocess.CalledProcessError):
+            await aio_run_command(("echo", "hello"))