compression/_bzip2: add tests

Add tests for all modes used by compression._bzip2 to compress and
decompress data.

Signed-off-by: Arthur Zamarin <arthurzam@gentoo.org>

3 files changed, 22 insertions, 12 deletions

diff --git a/src/snakeoil/compression/_bzip2.py b/src/snakeoil/compression/_bzip2.py
index 05db221c..b10e3ec3 100644
--- a/src/snakeoil/compression/_bzip2.py
+++ b/src/snakeoil/compression/_bzip2.py
@@ -24,9 +24,7 @@ bz2_path = process.find_binary("bzip2")
 
 
 try:
-    from bz2 import BZ2File
-    from bz2 import compress as _compress_data
-    from bz2 import decompress as _decompress_data
+    from bz2 import BZ2File, compress as _compress_data, decompress as _decompress_data
     native = True
 except ImportError:
 
@@ -40,16 +38,15 @@ except ImportError:
 _compress_handle = partial(_util.compress_handle, bz2_path)
 _decompress_handle = partial(_util.decompress_handle, bz2_path)
 
-lbzip2_path = None
-parallelizable = False
-lbzip2_compress_args = lbzip2_decompress_args = ()
 try:
     lbzip2_path = process.find_binary("lbzip2")
-    lbzip2_compress_args = ('-n%i' % multiprocessing.cpu_count(),)
+    lbzip2_compress_args = (f'-n{multiprocessing.cpu_count()}', )
     lbzip2_decompress_args = lbzip2_compress_args
     parallelizable = True
 except process.CommandNotFound:
-    pass
+    lbzip2_path = None
+    parallelizable = False
+    lbzip2_compress_args = lbzip2_decompress_args = ()
 
 
 def compress_data(data, level=9, parallelize=False):
@@ -76,6 +73,6 @@ def decompress_handle(handle, parallelize=False):
     if parallelize and parallelizable:
         return _util.decompress_handle(lbzip2_path, handle,
                                        extra_args=lbzip2_decompress_args)
-    elif (native and isinstance(handle, str)):
+    elif native and isinstance(handle, str):
         return BZ2File(handle, mode='r')
     return _decompress_handle(handle)
diff --git a/src/snakeoil/compression/_util.py b/src/snakeoil/compression/_util.py
index 70da8563..6a3c7258 100644
--- a/src/snakeoil/compression/_util.py
+++ b/src/snakeoil/compression/_util.py
@@ -12,9 +12,9 @@ def _drive_process(args, mode, data):
     try:
         stdout, stderr = p.communicate(data)
         if p.returncode != 0:
+            args = ' '.join(args)
             raise ValueError(
-                "%s returned %i exitcode from '%s', stderr=%r" %
-                (mode, p.returncode, ' '.join(args), stderr))
+                f"{mode} returned {p.returncode} exitcode from '{args}', stderr={stderr.decode()}")
         return stdout
     finally:
         if p is not None and p.returncode is None:
diff --git a/src/snakeoil/test/__init__.py b/src/snakeoil/test/__init__.py
index 72ea2b4e..bb9381f7 100644
--- a/src/snakeoil/test/__init__.py
+++ b/src/snakeoil/test/__init__.py
@@ -5,6 +5,7 @@ import random
 import string
 import subprocess
 import sys
+from unittest.mock import patch
 
 # not relative imports so protect_process() works properly
 from snakeoil import klass
@@ -67,8 +68,20 @@ def protect_process(functor, name=None):
             if wipe:
                 os.environ.pop(_PROTECT_ENV_VAR, None)
 
-    for x in "__doc__ __name__".split():
+    for x in ("__doc__", "__name__"):
         if hasattr(functor, x):
             setattr(_inner_run, x, getattr(functor, x))
     method_name = getattr(functor, '__name__', None)
     return _inner_run
+
+
+def hide_imports(*import_names: str):
+    """Hide imports from the specified modules."""
+    orig_import = __import__
+
+    def mock_import(name, *args, **kwargs):
+        if name in import_names:
+            raise ImportError()
+        return orig_import(name, *args, **kwargs)
+
+    return patch('builtins.__import__', side_effect=mock_import)