diff --git a/CHANGELOG.md b/CHANGELOG.md index e8e4b015..a03e7fcb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,7 @@ - Add the keyword `dumpactints` to `BlockOutput` (#245). - Added fallback keyword argument for `get_final_energy`, `get_gradient`, and `get_structure` that allows to parse these properties from the `.out` file if no JSON output is available (#237). - Added `get_frequencies`, `get_imaginary_frequencies`, `is_pes_minimum`, and `is_pes_transition_state` to the `Output` class (#247). +- Add `functions to clean up files created by ORCA jobs.(#262) - `Output.parse()`, `Output.parse_property()`, `Output.parse_gbw()`, and `Output.__init__()` now accept a `strict` parameter (default True). When set to False, output fields that fail Pydantic validation are silently set to None and a UserWarning is emitted instead of raising a ValidationError (#248). - Added `rmsd` and `rmsd_kabsch` to calculate RMSD without and with rotational alignment to `Structure` class (#230). - Added `calc_rotational_constants` to calculate molecular rotational constants to `Structure` class (#230). diff --git a/src/opi/output/core.py b/src/opi/output/core.py index faab0b02..09ebfc9a 100644 --- a/src/opi/output/core.py +++ b/src/opi/output/core.py @@ -4,7 +4,7 @@ """ import json -from collections.abc import Callable, Sequence +from collections.abc import Callable, Iterable, Sequence from pathlib import Path from typing import Any, cast from warnings import warn @@ -536,6 +536,67 @@ def get_file(self, suffix: str, /, *, basename: str = "") -> Path: basename = self.basename return self.working_dir / (basename + suffix) + def _delete_files( + self, basename: str | None = None, *, suffixes: Sequence[str] = tuple() + ) -> None: + """ + Delete files in `working_dir` belonging to the job with the given basename. + + Parameters + ---------- + basename : str | None, default: None + Basename of the job whose files should be deleted. + If not given, `self.basename` is used. + suffixes : Sequence[str] | None, default: None + If given, treated as glob patterns matched against `{basename}*{suffix}`; + only matching files are deleted. + If None, all files matching the basename are deleted. + """ + basename = basename if basename else self.basename + if not basename: + raise ValueError("No basename specified") + if "/" in basename or "\\" in basename: + raise ValueError(f"Basename must not contain path separators: {basename!r}") + files: Iterable[Path] + if not suffixes: + files = self.working_dir.glob(f"{basename}*") + else: + files = ( + file + for suffix in suffixes + for file in self.working_dir.glob(f"{basename}*{suffix}") + ) + for file in files: + if file.is_file(): + file.unlink(missing_ok=True) + + def cleanup_files(self, basename: str | None = None) -> None: + """ + Delete all files in `working_dir` belonging to the job with the given basename. This will also include files whose + names contain the basename, for example, if the basename is `job` and there exists files with basename 'job_1', + the files will be deleted. + + Parameters + ---------- + basename : str | None, default: None + Basename of the job whose files should be deleted. + If not given, `self.basename` is used. + """ + self._delete_files(basename) + + def cleanup_temp_files(self, basename: str | None = None) -> None: + """ + Delete temporary files in `working_dir` belonging to the job with the given basename. + + Parameters + ---------- + basename : str | None, default: None + Basename of the job whose temporary files should be deleted. + If not given, `self.basename` is used. + """ + temp_file_suffixes: tuple[str, ...] = (".tmp", ".proc", ".tmp.*", ".proc.*") + self._delete_files(basename, suffixes=temp_file_suffixes) + def _get_version(self) -> "OrcaVersion": """Gets the ORCA version from the property-JSON file.""" diff --git a/tests/unit/test_output_cleanup.py b/tests/unit/test_output_cleanup.py new file mode 100644 index 00000000..9a0b313d --- /dev/null +++ b/tests/unit/test_output_cleanup.py @@ -0,0 +1,108 @@ +from pathlib import Path + +import pytest + +from opi.output.core import Output + +""" +Unit tests for Output cleanup functions. + +This module contains tests for the file-deletion helpers: +- `Output.cleanup_files()` +- `Output.cleanup_temp_files()` +- `Output._delete_files()` +""" + + +def _make_output(basename: str, working_dir: Path) -> Output: + output = Output(basename, working_dir=working_dir, version_check=False) + return output + + +def _create_file(directory: Path, *names: str) -> None: + for name in names: + (directory / name).touch() + + +@pytest.mark.unit +@pytest.mark.output +def test_cleanup_files_deletes_matching_basename_files(tmp_path: Path): + """`cleanup_files()` deletes all files matching the basename, including files whose + name only contains the basename as a prefix, and leaves unrelated files untouched.""" + _create_file(tmp_path, "job.out", "job.gbw", "job_1.gbw", "unrelated.txt") + output = _make_output("job", tmp_path) + + output.cleanup_files() + + remaining = {p.name for p in tmp_path.iterdir()} + assert remaining == {"unrelated.txt"} + + +@pytest.mark.unit +@pytest.mark.output +def test_cleanup_files_explicit_basename_overrides_self_basename(tmp_path: Path): + """`cleanup_files(basename=...)` deletes files for the given basename, not `self.basename`.""" + _create_file(tmp_path, "job_1.out", "job_2.out") + output = _make_output("job_1", tmp_path) + + output.cleanup_files(basename="job_2") + + remaining = {p.name for p in tmp_path.iterdir()} + assert remaining == {"job_1.out"} + + +@pytest.mark.unit +@pytest.mark.output +def test_cleanup_files_raises_without_basename(tmp_path: Path): + """`cleanup_files()` raises `ValueError` rather than deleting everything in `working_dir`.""" + _create_file(tmp_path, "unrelated.txt") + output = _make_output("", tmp_path) + + with pytest.raises(ValueError, match="No basename specified"): + output.cleanup_files() + + assert (tmp_path / "unrelated.txt").exists() + + +@pytest.mark.unit +@pytest.mark.output +@pytest.mark.parametrize("basename", ["../secret", "sub/job", "sub\\job"]) +def test_cleanup_files_rejects_path_separators_in_basename(tmp_path: Path, basename: str): + """A basename containing a path separator must be rejected rather than letting the + glob pattern escape `working_dir`.""" + working_dir = tmp_path / "workdir" + working_dir.mkdir() + _create_file(tmp_path, "secret_file.txt") + output = _make_output(basename, working_dir) + + with pytest.raises(ValueError, match="path separators"): + output.cleanup_files() + + assert (tmp_path / "secret_file.txt").exists() + + +@pytest.mark.unit +@pytest.mark.output +def test_cleanup_temp_files_deletes_only_temp_suffixes(tmp_path: Path): + """`cleanup_temp_files()` only deletes `.tmp.*`/`.proc.*` files, not other job files.""" + _create_file(tmp_path, "job.out", "job.gbw", "job.tmp", "job.proc", "job_scan.tmp") + output = _make_output("job", tmp_path) + + output.cleanup_temp_files() + + remaining = {p.name for p in tmp_path.iterdir()} + assert remaining == {"job.out", "job.gbw"} + + +@pytest.mark.unit +@pytest.mark.output +def test_delete_files_empty_suffixes_deletes_all(tmp_path: Path): + """An empty `suffixes` sequence behaves like no filter at all: all files matching + the basename are deleted, same as omitting `suffixes`.""" + _create_file(tmp_path, "job.out", "job.gbw", "unrelated.txt") + output = _make_output("job", tmp_path) + + output._delete_files(None, suffixes=()) + + remaining = {p.name for p in tmp_path.iterdir()} + assert remaining == {"unrelated.txt"}