add function that raises an error if output file(s) already exist, and add a test for this

Author: cadenmyers13

src/diffpy/labpdfproc/labpdfprocapp.py

@@ -316,33 +316,11 @@ def get_args_cli(override=None):
     return parser.parse_args(argv)
 
 
-def _check_saved_file_exists(args):
-    """Check if the output files already exist based on the input paths
-    and output directory."""
-    existing_files = []
-    for path in args.input_paths:
-        outfile = args.output_directory / (path.stem + "_corrected.chi")
-        if outfile.exists() and not args.force:
-            existing_files.append(outfile)
-        if args.output_correction:
-            corrfile = args.output_directory / (path.stem + "_cve.chi")
-            if corrfile.exists() and not args.force:
-                existing_files.append(corrfile)
-    if existing_files:
-        existing_files_str = "\n".join(str(f) for f in existing_files)
-        raise FileExistsError(
-            "The following output files already exist:"
-            f"\n{existing_files_str}\n"
-            "Use --force to overwrite them."
-        )
-
-
 def main():
     use_gui = len(sys.argv) == 1 or "--gui" in sys.argv
     args = get_args_gui() if use_gui else get_args_cli()
     args = _handle_old_api_conversion(args)
     args = preprocessing_args(args)
-    _check_saved_file_exists(args)
     apply_absorption_correction(args)
 
 

src/diffpy/labpdfproc/tools.py

@@ -528,6 +528,27 @@ def load_package_info(args):
     return args
 
 
+def _check_saved_file_exists(args):
+    """Check if the output files already exist based on the input paths
+    and output directory."""
+    existing_files = []
+    for path in args.input_paths:
+        outfile = args.output_directory / (path.stem + "_corrected.chi")
+        if outfile.exists() and not args.force:
+            existing_files.append(outfile)
+        if args.output_correction:
+            corrfile = args.output_directory / (path.stem + "_cve.chi")
+            if corrfile.exists() and not args.force:
+                existing_files.append(corrfile)
+    if existing_files:
+        existing_files_str = "\n".join(str(f) for f in existing_files)
+        raise FileExistsError(
+            "The following output files already exist:"
+            f"\n{existing_files_str}\n"
+            "Use --force to overwrite them."
+        )
+
+
 def preprocessing_args(args):
     """Perform preprocessing on the provided args. The process includes
     loading package and user information, setting input, output,
@@ -542,6 +563,11 @@ def preprocessing_args(args):
     -------
     args : argparse.Namespace
         The updated argparse Namespace with arguments preprocessed.
+
+    Raises
+    ------
+    FileExistsError
+        If the output files already exist and --force is not used.
     """
     args = load_wavelength_from_config_file(args)
     args = set_mud(args)
@@ -552,6 +578,7 @@ def preprocessing_args(args):
     args = load_user_metadata(args)
     args = load_user_info(args)
     args = load_package_info(args)
+    _check_saved_file_exists(args)
     return args
 
 

tests/conftest.py

@@ -13,7 +13,8 @@ def user_filesystem(tmp_path):
     home_dir.mkdir(parents=True, exist_ok=True)
     test_dir = base_dir / "test_dir"
     test_dir.mkdir(parents=True, exist_ok=True)
-
+    output_dir = base_dir / "output_dir"
+    output_dir.mkdir(parents=True, exist_ok=True)
     chi_data = (
         "dataformat = twotheta\n mode = "
         "xray\n # chi_Q chi_I\n 1 2\n 3 4\n 5 6\n 7 8\n"
@@ -24,10 +25,6 @@ def user_filesystem(tmp_path):
     )
     binary_data = b"\x00\x01\x02\x03\x04"
 
-    with open(base_dir / "data.chi", "w") as f:
-        f.write(chi_data)
-    with open(base_dir / "data_corrected.chi", "w") as f:
-        f.write(chi_data)
     with open(base_dir / "good_data.chi", "w") as f:
         f.write(chi_data)
     with open(base_dir / "good_data.xy", "w") as f:
@@ -63,6 +60,10 @@ def user_filesystem(tmp_path):
         f.write("good_data.xy \n")
         f.write(f"{str(input_dir.resolve() / 'good_data.txt')}\n")
 
+    with open(output_dir / "good_data_corrected.chi", "w") as f:
+        f.write(chi_data)
+    with open(output_dir / "good_data_cve.chi", "w") as f:
+        f.write(chi_data)
     home_config_data = {
         "wavelength": 0.3,
         "owner_name": "home_username",

tests/test_labpdfprocapp.py

@@ -847,3 +847,32 @@ def test_load_metadata(mocker, user_filesystem, inputs, expected):
             **expected,
         }
         assert actual_metadata == expected_metadata
+
+
+def test_preprocess_args_bad(user_filesystem):
+    # Case: user tries to run absorption correction, but the output
+    # filenames already for *_corrected.chi and *_cve.chi exists.
+    # expected: preprocess_args catches this early and raises an Error
+    input_data_file = str(user_filesystem / "good_data.chi")
+    existing_corrected_file = str(
+        user_filesystem / "output_dir" / "good_data_corrected.chi"
+    )
+    existing_corrected_cve_file = str(
+        user_filesystem / "output_dir" / "good_data_cve.chi"
+    )
+    cli_inputs = (
+        ["mud"]
+        + [input_data_file]
+        + ["2.5", "-w", "Mo", "-o", str(user_filesystem / "output_dir")]
+        + ["-c"]  # -c flag saves the cve file
+    )
+    args = get_args_cli(cli_inputs)
+    args = set_input_lists(args)
+    msg = (
+        "The following output files already exist:"
+        f"\n{existing_corrected_file}\n"
+        f"{existing_corrected_cve_file}\n"
+        "Use --force to overwrite them."
+    )
+    with pytest.raises(FileExistsError, match=re.escape(msg)):
+        preprocessing_args(args)