diff --git a/src/xopen/__init__.py b/src/xopen/__init__.py index f401fbc..875b287 100644 --- a/src/xopen/__init__.py +++ b/src/xopen/__init__.py @@ -95,6 +95,7 @@ class ProgramSettings(typing.NamedTuple): program_args: Tuple[str, ...] + acceptable_compression_levels: Tuple[int, ...] = tuple(range(1, 10)) threads_flag: Optional[str] = None # This exit code is not interpreted as an error when terminating the process allowed_exit_code: Optional[int] = -signal.SIGTERM @@ -106,14 +107,15 @@ class ProgramSettings(typing.NamedTuple): PROGRAM_SETTINGS: Dict[str, ProgramSettings] = { "pbzip2": ProgramSettings( ("pbzip2",), + tuple(range(1, 10)), "-p", allowed_exit_code=None, allowed_exit_message=b"\n *Control-C or similar caught [sig=15], quitting...", ), - "xz": ProgramSettings(("xz",), "-T"), - "zstd": ProgramSettings(("zstd",), "-T"), - "pigz": ProgramSettings(("pigz", "--no-name"), "-p"), - "gzip": ProgramSettings(("gzip", "--no-name")), + "xz": ProgramSettings(("xz",), tuple(range(0, 10)), "-T"), + "zstd": ProgramSettings(("zstd",), tuple(range(1, 20)), "-T"), + "pigz": ProgramSettings(("pigz", "--no-name"), tuple(range(0, 10)) + (11,), "-p"), + "gzip": ProgramSettings(("gzip", "--no-name"), tuple(range(1, 10))), } @@ -208,6 +210,13 @@ def __init__( # noqa: C901 raise ValueError( f"Mode is '{mode}', but it must be 'r', 'rb', 'w', 'wb', 'a', or 'ab'" ) + if ( + compresslevel + and compresslevel not in program_settings.acceptable_compression_levels + ): + raise ValueError( + f"compresslevel must be in {program_settings.acceptable_compression_levels}." + ) path = os.fspath(path) if isinstance(path, bytes) and sys.platform == "win32": path = path.decode() diff --git a/tests/test_xopen.py b/tests/test_xopen.py index a0e8fd2..0e3ee11 100644 --- a/tests/test_xopen.py +++ b/tests/test_xopen.py @@ -258,10 +258,10 @@ def test_filename_not_a_string(): def test_invalid_compression_level(tmp_path): # Can be any kind of exception depending on backend, OSError, zlib.error, # zlib_ng.error, isal_zlib.error, ValueError (etc.) - with pytest.raises(Exception) as e: + with pytest.raises(ValueError) as e: with xopen(tmp_path / "out.gz", mode="w", compresslevel=17) as f: f.write("hello") # pragma: no cover - assert "level" in e.value.args[0] + assert "compresslevel must be" in e.value.args[0] @pytest.mark.parametrize("ext", extensions)