r.pack: Replace traceback by error message for output issues (#6346)

Add check for issues with the output file parent directory, and report error messages. This avoids a traceback originating at the very end of the tool execution when calling shutil.move.

This adds tests for the newly handled error states as well as for other issues with the inputs.

Skip the read-only dir test for Windows. At least in GitHub Actions Windows CI, this does not cause r.pack to fail. Assuming here that the directory still allows writes, we can't fully set up the test, so we don't test that for Windows.

It adds the new pytest fixtures to scripts conftest.py file given there is at least some potential for reusability for similar simple tests.

Author: wenzeslaus

scripts/conftest.py

@@ -4,13 +4,54 @@
 SPDX-License-Identifier: GPL-2.0-or-later
 """
 
+import os
 from collections.abc import Generator
 
 import pytest
 
+import grass.script as gs
+from grass.tools import Tools
+from grass.experimental import TemporaryMapsetSession
+
 
 @pytest.fixture(scope="module")
 def monkeypatch_module() -> Generator[pytest.MonkeyPatch]:
     """Yield a monkeypatch context, through a module-scoped fixture"""
     with pytest.MonkeyPatch.context() as mp:
         yield mp
+
+
+@pytest.fixture(scope="module")
+def xy_raster_dataset_session_for_module(tmp_path_factory):
+    """Active session in an XY location with small data (scope: function)"""
+    project = tmp_path_factory.mktemp("scripts_raster_dataset") / "xy_test"
+    gs.create_project(project)
+    with (
+        gs.setup.init(project, env=os.environ.copy()) as session,
+        Tools(session=session) as tools,
+    ):
+        tools.g_region(s=0, n=5, w=0, e=6, res=1, flags="s")
+        tools.r_mapcalc(expression="rows_raster = row()")
+        yield session
+
+
+@pytest.fixture
+def xy_raster_dataset_session_mapset(xy_raster_dataset_session_for_module):
+    """A session in a temporary mapset"""
+    with TemporaryMapsetSession(
+        env=xy_raster_dataset_session_for_module.env
+    ) as session:
+        yield session
+
+
+@pytest.fixture
+def xy_empty_dataset_session(tmp_path):
+    """Active session in an XY location (scope: function)"""
+    project = tmp_path / "xy_test"
+    gs.create_project(project)
+    with (
+        gs.setup.init(project, env=os.environ.copy()) as session,
+        Tools(session=session) as tools,
+    ):
+        tools.g_region(s=0, n=5, w=0, e=6, res=1)
+        yield session

scripts/r.pack/r.pack.py

@@ -39,6 +39,7 @@
 
 from pathlib import Path
 
+import grass.script as gs
 from grass.script.utils import try_rmdir, try_remove
 from grass.script import core as grass
 
@@ -71,6 +72,7 @@ def main():
     if not gfile["name"]:
         grass.fatal(_("Raster map <%s> not found") % infile)
 
+    parent_dir = Path(outfile).parent
     if os.path.exists(outfile):
         if os.getenv("GRASS_OVERWRITE"):
             grass.warning(
@@ -79,6 +81,12 @@ def main():
             try_remove(outfile)
         else:
             grass.fatal(_("option <output>: <%s> exists.") % outfile)
+    elif not parent_dir.exists():
+        gs.fatal(_("Directory '{path}' does not exist").format(path=parent_dir))
+    elif not parent_dir.is_dir():
+        gs.fatal(_("Path '{path}' is not a directory").format(path=parent_dir))
+    elif not os.access(parent_dir, os.W_OK):
+        gs.fatal(_("Directory '{path}' is not writable").format(path=parent_dir))
 
     grass.message(_("Packing <%s> to <%s>...") % (gfile["fullname"], outfile))
     basedir = os.path.sep.join(os.path.normpath(gfile["file"]).split(os.path.sep)[:-2])

scripts/r.pack/tests/r_pack_test.py

@@ -0,0 +1,133 @@
+import os
+import re
+import stat
+import sys
+
+import pytest
+
+from grass.exceptions import CalledModuleError
+from grass.tools import Tools
+
+
+@pytest.mark.parametrize("compression_flag", ["", "c"])
+@pytest.mark.parametrize(
+    "suffix",
+    [".pack", ".rpack", ".r_pack", ".grass_raster", ".grr", ".arbitrary_suffix_1"],
+)
+def test_pack_different_suffixes(
+    xy_raster_dataset_session_for_module, tmp_path, suffix, compression_flag
+):
+    """Check that non-zero output is created with different suffixes
+
+    The compression should not change the allowed suffixes.
+    """
+    tools = Tools(session=xy_raster_dataset_session_for_module)
+    output = (tmp_path / "output_1").with_suffix(suffix)
+    tools.r_pack(input="rows_raster", output=output, flags=compression_flag)
+    assert output.exists()
+    assert output.is_file()
+    assert output.stat().st_size > 0
+
+
+def test_input_does_not_exist(xy_empty_dataset_session, tmp_path):
+    """Check we raise when input does not exists"""
+    tools = Tools(session=xy_empty_dataset_session)
+    output = tmp_path / "output_1.rpack"
+    with pytest.raises(CalledModuleError, match="does_not_exist"):
+        tools.r_pack(input="does_not_exist", output=output)
+
+
+def test_output_exists_without_overwrite(
+    xy_raster_dataset_session_for_module, tmp_path
+):
+    """Check behavior when output already exists"""
+    tools = Tools(session=xy_raster_dataset_session_for_module)
+    output = tmp_path / "output_1.rpack"
+    assert not output.exists()
+    tools.r_pack(input="rows_raster", output=output)
+    assert output.exists()
+    with pytest.raises(
+        CalledModuleError, match=rf"(?s){re.escape(str(output))}.*[Oo]verwrite"
+    ):
+        # Same call repeated again.
+        tools.r_pack(input="rows_raster", output=output)
+    assert output.exists()
+
+
+def test_output_exists_with_overwrite(xy_raster_dataset_session_for_module, tmp_path):
+    """Check behavior when output already exists and overwrite is set"""
+    tools = Tools(session=xy_raster_dataset_session_for_module)
+    output = tmp_path / "output_1.rpack"
+    assert not output.exists()
+    tools.r_pack(input="rows_raster", output=output)
+    assert output.exists()
+    # Same call repeated again but with overwrite.
+    tools.r_pack(input="rows_raster", output=output, overwrite=True)
+    assert output.exists()
+
+
+def test_output_dir_does_not_exist(xy_raster_dataset_session_for_module, tmp_path):
+    """Check behavior when directory for output does not exist"""
+    tools = Tools(session=xy_raster_dataset_session_for_module)
+    output = tmp_path / "does_not_exist" / "output_1.rpack"
+    assert not output.exists()
+    assert not output.parent.exists()
+    with pytest.raises(
+        CalledModuleError,
+        match=rf"(?s)'{re.escape(str(output.parent))}'.*does not exist",
+    ):
+        tools.r_pack(input="rows_raster", output=output)
+
+
+def test_output_dir_is_file(xy_raster_dataset_session_for_module, tmp_path):
+    """Check behavior when what should be a directory is a file"""
+    tools = Tools(session=xy_raster_dataset_session_for_module)
+    parent = tmp_path / "file_as_dir"
+    parent.touch()
+    output = parent / "output_1.rpack"
+    assert not output.exists()
+    assert output.parent.exists()
+    assert output.parent.is_file()
+    with pytest.raises(
+        CalledModuleError,
+        match=rf"(?s)'{re.escape(str(output.parent))}'.*is not.*directory",
+    ):
+        tools.r_pack(input="rows_raster", output=output)
+
+
+@pytest.mark.skipif(sys.platform.startswith("win"), reason="Dir still writable")
+def test_output_dir_is_not_writable(xy_raster_dataset_session_for_module, tmp_path):
+    """Check behavior when directory is not writable"""
+    tools = Tools(session=xy_raster_dataset_session_for_module)
+    parent = tmp_path / "parent_dir"
+    parent.mkdir()
+    # This should work for Windows according to the doc, but it does not.
+    parent.chmod(stat.S_IREAD)
+    output = parent / "output_1.rpack"
+    # Calling output.exists() gives permission denied on Linux, but os.path does not.
+    assert not os.path.exists(output)
+    assert output.parent.exists()
+    assert output.parent.is_dir()
+    with pytest.raises(
+        CalledModuleError,
+        match=rf"(?s)'{re.escape(str(output.parent))}'.*is not.*writable",
+    ):
+        tools.r_pack(input="rows_raster", output=output)
+
+
+@pytest.mark.parametrize("compression_flag", ["", "c"])
+def test_round_trip_with_r_unpack(
+    xy_raster_dataset_session_mapset, tmp_path, compression_flag
+):
+    """Check that we get the same values with r.unpack"""
+    tools = Tools(session=xy_raster_dataset_session_mapset)
+    output = tmp_path / "output_1.rpack"
+    tools.r_pack(input="rows_raster", output=output, flags=compression_flag)
+    imported = "imported"
+    tools.r_unpack(input=output, output=imported)
+    tools.r_mapcalc(expression="diff = rows_raster - imported")
+    stats = tools.r_univar(map="diff", format="json")
+    assert stats["n"] > 1
+    assert stats["sum"] == 0
+    assert stats["min"] == 0
+    assert stats["max"] == 0