Fix chdir and add tests for fsutils functions

src/wxflow/fsutils.py

@@ -1,34 +1,45 @@
-import contextlib
-import errno
 import grp
 import os
 import shutil
+from contextlib import contextmanager
+from logging import getLogger
 
 __all__ = ['mkdir', 'mkdir_p', 'rmdir', 'chdir', 'rm_p', 'cp',
            'get_gid', 'chgrp']
 
+logger = getLogger(__name__.split('.')[-1])
+
 
 def mkdir_p(path):
     try:
-        os.makedirs(path)
-    except OSError as exc:
-        if exc.errno == errno.EEXIST and os.path.isdir(path):
-            pass
-        else:
-            raise OSError(f"unable to create directory at {path}")
+        os.makedirs(path, exist_ok=True)
+    except OSError:
+        raise OSError(f"unable to create directory at {path}")
 
 
 mkdir = mkdir_p
 
 
-def rmdir(dir_path):
+def rmdir(dir_path, missing_ok=False):
+    """
+    Attempt to delete a directory and all of its contents.
+    If ignore_missing is True, then a missing directory will not raise an error.
+    """
+
     try:
         shutil.rmtree(dir_path)
-    except OSError as exc:
-        raise OSError(f"unable to remove {dir_path}")
 
+    except FileNotFoundError as exc:
+        if missing_ok:
+            logger.warning(f"WARNING cannot remove the target path {dir_path} because it does not exist")
+        else:
+            raise exc
+
+    except OSError:
+        raise OSError(f"Unable to remove the target directory: {dir_path}")
 
-@contextlib.contextmanager
+
+@contextmanager
 def chdir(path):
     """Change current working directory and yield.
     Upon completion, the working directory is switched back to the directory at the time of call.
@@ -45,22 +56,35 @@ def chdir(path):
         do_thing_2
     """
     cwd = os.getcwd()
+    # Try to change paths.
     try:
         os.chdir(path)
+    except OSError:
+        raise OSError(f"Failed to change directory to ({path})")
+
+    # If successful, yield to the calling "with" statement.
+    try:
         yield
     finally:
-        print(f"WARNING: Unable to chdir({path})")  # TODO: use logging
+        # Once the with is complete, head back to the original working directory
         os.chdir(cwd)
 
 
-def rm_p(path):
+def rm_p(path, missing_ok=True):
+    """
+    Attempt to delete a file.
+    If missing_ok is True, an error is not raised if the file does not exist.
+    """
+
     try:
         os.unlink(path)
-    except OSError as exc:
-        if exc.errno == errno.ENOENT:
-            pass
+    except FileNotFoundError:
+        if missing_ok:
+            logger.warning(f"WARNING cannot remove the file {path} because it does not exist")
         else:
-            raise OSError(f"unable to remove {path}")
+            raise FileNotFoundError(f"The file {path} does not exist")
+    except OSError:
+        raise OSError(f"unable to remove {path}")
 
 
 def cp(source: str, target: str) -> None:

tests/test_fsutils.py

@@ -0,0 +1,136 @@
+import os
+
+import pytest
+
+from wxflow import chdir, cp, get_gid, mkdir, rm_p, rmdir
+
+
+def test_mkdir(tmp_path):
+    """
+    Test for creating a directory:
+    Parameters
+    ----------
+    tmp_path - pytest fixture
+    """
+
+    dir_path = tmp_path / 'my_test_dir'
+    dir_path_bad = "/some/non-existent/path"
+
+    # Create the good path
+    mkdir(dir_path)
+
+    # Check if dir_path was created
+    assert os.path.exists(dir_path)
+
+    # Test that attempting to create a bad path raises an OSError
+    with pytest.raises(OSError):
+        mkdir(dir_path_bad)
+
+
+def test_rmdir(tmp_path):
+    """
+    Test for removing a directory:
+    Parameters
+    ----------
+    tmp_path - pytest fixture
+    """
+
+    dir_path = tmp_path / 'my_input_dir'
+    # Make and then delete the directory
+    mkdir(dir_path)
+    rmdir(dir_path)
+
+    # Assert that it was deleted
+    assert not os.path.exists(dir_path)
+
+    # Attempt to delete a non-existent path and ignore that it is missing
+    rmdir('/non-existent-path', missing_ok=True)
+
+    # Lastly, attempt to delete a non-existent directory and do not ignore the error
+    with pytest.raises(FileNotFoundError):
+        rmdir('/non-existent-path')
+
+
+def test_chdir(tmp_path):
+    """
+    Test for changing a directory:
+    Parameters
+    ----------
+    tmp_path - pytest fixture
+    """
+
+    dir_path = tmp_path / 'my_input_dir'
+    # Make the directory and navigate to it
+    mkdir(dir_path)
+
+    with chdir(dir_path):
+        assert os.getcwd() == os.path.abspath(dir_path)
+
+    # Now try to go somewhere that doesn't exist
+    with pytest.raises(OSError):
+        with chdir("/a/non-existent/path"):
+            raise AssertionError("Navigated to a non-existent path")
+
+
+def test_rm_p(tmp_path):
+    """
+    Test for removing a file
+    Parameters
+    ----------
+    tmp_path - pytest fixture
+    """
+
+    input_path = tmp_path / 'my_test_file.txt'
+    # Attempt to delete a non-existent file, ignoring any errors
+    rm_p(input_path)
+
+    # Now attempt to delete the same file but do not ignore errors
+    with pytest.raises(FileNotFoundError):
+        rm_p(input_path, missing_ok=False)
+
+    with open(input_path, "w") as f:
+        f.write("")
+
+    # Delete the file and assert it doesn't exist
+    rm_p(input_path)
+
+    assert not os.path.isfile(input_path)
+
+
+def test_cp(tmp_path):
+    """
+    Test copying a file:
+    Parameters
+    ----------
+    tmp_path - pytest fixture
+    """
+
+    input_path = tmp_path / 'my_test_file.txt'
+    output_path = tmp_path / 'my_output_file.txt'
+    # Attempt to copy a non-existent file
+    rm_p(input_path)  # Delete it if present
+    with pytest.raises(OSError):
+        cp(input_path, output_path)
+
+    # Now create the input file and repeat
+    with open(input_path, "w") as f:
+        f.write("")
+
+    cp(input_path, output_path)
+
+    # Assert both files exist (make sure it wasn't moved).
+    assert os.path.isfile(output_path)
+    assert os.path.isfile(input_path)
+
+
+def test_get_gid():
+    """
+    Test getting a group ID:
+    """
+
+    # Try to change groups to a non-existent one.
+    with pytest.raises(KeyError):
+        get_gid("some-non-existent-group")
+
+    # Now get the root group ID (should be 0)
+    assert get_gid("root") == 0