From a7b49e9cc76ef4b50cc1c28d4b7959ebde99c5f5 Mon Sep 17 00:00:00 2001 From: David Huber <69919478+DavidHuber-NOAA@users.noreply.github.com> Date: Fri, 15 Nov 2024 07:47:24 -0500 Subject: [PATCH] Fix chdir and add tests for fsutils functions (#45) --- src/wxflow/fsutils.py | 60 ++++++++++----- tests/test_configuration.py | 2 +- tests/test_fsutils.py | 150 ++++++++++++++++++++++++++++++++++++ 3 files changed, 193 insertions(+), 19 deletions(-) create mode 100644 tests/test_fsutils.py diff --git a/src/wxflow/fsutils.py b/src/wxflow/fsutils.py index af9e5b8..70466d7 100644 --- a/src/wxflow/fsutils.py +++ b/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: + if missing_ok: + logger.warning(f"WARNING cannot remove the target path {dir_path} because it does not exist") + else: + raise FileNotFoundError(f"Target directory ({dir_path}) cannot be removed because it does not exist") + + 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: diff --git a/tests/test_configuration.py b/tests/test_configuration.py index 077b00a..059af31 100644 --- a/tests/test_configuration.py +++ b/tests/test_configuration.py @@ -176,7 +176,7 @@ def test_configuration_config_dir(tmp_path, create_configs): def test_configuration_config_files(tmp_path, create_configs): cfg = Configuration(tmp_path) config_files = [str(tmp_path / 'config.file0'), str(tmp_path / 'config.file1')] - assert config_files == cfg.config_files + assert sorted(config_files) == sorted(cfg.config_files) def test_find_config(tmp_path, create_configs): diff --git a/tests/test_fsutils.py b/tests/test_fsutils.py new file mode 100644 index 0000000..8e7b6e8 --- /dev/null +++ b/tests/test_fsutils.py @@ -0,0 +1,150 @@ +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) + + # Get the CWD to verify that we come back after the with. + cwd = os.getcwd() + + with chdir(dir_path): + assert os.getcwd() == os.path.abspath(dir_path) + + assert os.getcwd() == cwd + + # 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") + + # Lastly, test that we return to the orignial working directory when there is an error + try: + with chdir(dir_path): + 1 / 0 + except ZeroDivisionError: + pass + + assert os.getcwd() == cwd + + +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