Merge pull request #38 from pycompression/tostdout

Allow compressing to stdout like gzip and pigz on the isal.igzip CLI.

Author: rhpvorderman

CHANGELOG.rst

@@ -7,6 +7,12 @@ Changelog
 .. This document is user facing. Please word the changes in such a way
 .. that users understand how the changes affect the new version.
 
+version 0.5.0-dev
+-----------------
++ A ``-c`` or ``--stdout`` flag was added to the CLI interface of isal.igzip.
+  This allows it to behave more like the ``gzip`` or ``pigz`` command line
+  interfaces.
+
 version 0.4.0
 -----------------
 + Move wheel building to cibuildwheel on github actions CI. Wheels are now

src/isal/igzip.py

@@ -276,30 +276,38 @@ def main():
         "-d", "--decompress", action="store_false",
         dest="compress",
         help="Decompress the file instead of compressing.")
+    parser.add_argument("-c", "--stdout", action="store_true",
+                        help="write on standard output")
     args = parser.parse_args()
 
     compresslevel = args.compresslevel or _COMPRESS_LEVEL_TRADEOFF
 
-    if args.file is None:
-        if args.compress:
-            in_file = sys.stdin.buffer
-            out_file = IGzipFile(mode="wb", compresslevel=compresslevel,
-                                 fileobj=sys.stdout.buffer)
-        else:
-            in_file = IGzipFile(mode="rb", fileobj=sys.stdin.buffer)
-            out_file = sys.stdout.buffer
-    else:
-        if args.compress:
-            in_file = io.open(args.file, mode="rb")
-            out_file = open(args.file + ".gz", mode="wb",
-                            compresslevel=compresslevel)
-        else:
-            base, extension = os.path.splitext(args.file)
-            if extension != ".gz":
-                print(f"filename doesn't end in .gz: {args.file}")
-                return
-            in_file = open(args.file, "rb")
-            out_file = io.open(base, "wb")
+    # Determine output file
+    if args.compress and (args.file is None or args.stdout):
+        out_file = IGzipFile(mode="wb", compresslevel=compresslevel,
+                             fileobj=sys.stdout.buffer)
+    elif args.compress and args.file is not None:
+        out_file = open(args.file + ".gz", mode="wb",
+                        compresslevel=compresslevel)
+    elif not args.compress and (args.file is None or args.stdout):
+        out_file = sys.stdout.buffer
+    elif not args.compress and args.file is not None:
+        base, extension = os.path.splitext(args.file)
+        if extension != ".gz":
+            raise ValueError(f"filename doesn't end in .gz: {args.file}. "
+                             f"Cannot determine filename for output")
+        out_file = io.open(base, "wb")
+
+    # Determine input file
+    if args.compress and args.file is None:
+        in_file = sys.stdin.buffer
+    elif args.compress and args.file is not None:
+        in_file = io.open(args.file, mode="rb")
+    elif not args.compress and args.file is None:
+        in_file = IGzipFile(mode="rb", fileobj=sys.stdin.buffer)
+    elif not args.compress and args.file is not None:
+        in_file = open(args.file, "rb")
+
     try:
         while True:
             block = in_file.read(BUFFER_SIZE)

tests/test_gzip_compliance.py

@@ -816,11 +816,17 @@ def test_decompress_infile_outfile(self):
         self.assertTrue(os.path.exists(igzipname))
 
     def test_decompress_infile_outfile_error(self):
-        rc, out, err = assert_python_ok('-m', 'isal.igzip', '-d',
-                                        'thisisatest.out')
-        self.assertIn(b"filename doesn't end in .gz:", out)
-        self.assertEqual(rc, 0)
-        self.assertEqual(err, b'')
+        rc, out, err = assert_python_failure('-m', 'isal.igzip', '-d',
+                                             'thisisatest.out')
+        # We take a divide from the original gzip module here. Error messages
+        # should be printed in stderr. Also exit code should not be 0!
+        # in python -m gzip -d mycompressedfile > decompressed
+        # will simply make decompressed contents 'filename doesn't end in .gz'
+        # without throwing an error. Crazy!
+        # TODO: Report a bug in CPython for gzip module
+        self.assertIn(b"filename doesn't end in .gz:", err)
+        self.assertNotEqual(rc, 0)
+        self.assertEqual(out, b'')
 
     @create_and_remove_directory(TEMPDIR)
     def test_compress_stdin_outfile(self):

tests/test_igzip_cli.py

@@ -0,0 +1,113 @@
+# Copyright (c) 2020 Leiden University Medical Center
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+"""Tests for the igzip CLI. Uses pytest which works better than unittest for
+these sort of tests. Meant to complement the gzip module compliance tests.
+It should improve coverage as well."""
+
+import gzip
+import io
+import sys
+
+from isal import igzip
+
+import pytest
+
+DATA = b'This is a simple test with igzip'
+COMPRESSED_DATA = gzip.compress(DATA)
+
+
+@pytest.mark.parametrize("level", range(1, 10))
+def test_decompress_stdin_stdout(capsysbinary, level):
+    """Test if the command line can decompress data that has been compressed
+    by gzip at all levels."""
+    mock_stdin = io.BytesIO(gzip.compress(DATA, level))
+    sys.stdin = io.TextIOWrapper(mock_stdin)
+    sys.argv = ["", "-d"]
+    igzip.main()
+    out, err = capsysbinary.readouterr()
+    assert err == b''
+    assert out == DATA
+
+
+@pytest.mark.parametrize("level", [str(x) for x in range(4)])
+def test_compress_stdin_stdout(capsysbinary, level):
+    mock_stdin = io.BytesIO(DATA)
+    sys.stdin = io.TextIOWrapper(mock_stdin)
+    sys.argv = ["", f"-{level}"]
+    igzip.main()
+    out, err = capsysbinary.readouterr()
+    assert err == b''
+    assert gzip.decompress(out) == DATA
+
+
+def test_decompress_infile_outfile(tmp_path, capsysbinary):
+    test_file = tmp_path / "test"
+    compressed_temp = test_file.with_suffix(".gz")
+    compressed_temp.write_bytes(gzip.compress(DATA))
+    sys.argv = ['', '-d', str(compressed_temp)]
+    igzip.main()
+    out, err = capsysbinary.readouterr()
+    assert err == b''
+    assert out == b''
+    assert test_file.exists()
+    assert test_file.read_bytes() == DATA
+
+
+def test_compress_infile_outfile(tmp_path, capsysbinary):
+    test_file = tmp_path / "test"
+    test_file.write_bytes(DATA)
+    sys.argv = ['', str(test_file)]
+    igzip.main()
+    out_file = test_file.with_suffix(".gz")
+    out, err = capsysbinary.readouterr()
+    assert err == b''
+    assert out == b''
+    assert out_file.exists()
+    assert gzip.decompress(out_file.read_bytes()) == DATA
+
+
+def test_decompress_infile_outfile_error(capsysbinary):
+    sys.argv = ['', '-d', 'thisisatest.out']
+    with pytest.raises(ValueError) as error:
+        igzip.main()
+    assert error.match("filename doesn't end")
+    out, err = capsysbinary.readouterr()
+    assert out == b''
+
+
+def test_decompress_infile_stdout(capsysbinary, tmp_path):
+    test_gz = tmp_path / "test.gz"
+    test_gz.write_bytes(gzip.compress(DATA))
+    sys.argv = ['', '-cd', str(test_gz)]
+    igzip.main()
+    out, err = capsysbinary.readouterr()
+    assert out == DATA
+    assert err == b''
+
+
+def test_compress_infile_stdout(capsysbinary, tmp_path):
+    test = tmp_path / "test"
+    test.write_bytes(DATA)
+    sys.argv = ['', '-c', str(test)]
+    igzip.main()
+    out, err = capsysbinary.readouterr()
+    assert gzip.decompress(out) == DATA
+    assert err == b''