diff --git a/README.md b/README.md index a40f24a..4710938 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ pip install sum-buddy ### Command Line Usage ``` -usage: sum-buddy [-h] [-V] [-o OUTPUT_FILE] [-i IGNORE_FILE | -H] [-a ALGORITHM] [-l LENGTH] [--archive-dive | --no-archive-dive] input_path +usage: sum-buddy [-h] [-V] [-o OUTPUT_FILE] [-f] [-i IGNORE_FILE | -H] [-a ALGORITHM] [-l LENGTH] [--archive-dive | --no-archive-dive] input_path Generate CSV with filepath, filename, and checksums for all files in a given directory (or a single file) @@ -28,7 +28,8 @@ options: -h, --help show this help message and exit -V, --version show program's version number and exit -o OUTPUT_FILE, --output-file OUTPUT_FILE - Filepath for the output CSV file + Filepath for the output CSV file; defaults to stdout + -f, --force Overwrite the output file if it already exists -i IGNORE_FILE, --ignore-file IGNORE_FILE Filepath for the ignore patterns file -H, --include-hidden Include hidden files @@ -80,6 +81,16 @@ cat examples/checksums.csv > examples/example_content/testzip.zip/dir/file.txt,file.txt,7d52c7437e9af58dac029dd11b1024df > ``` + If the output file already exists, sum-buddy refuses to overwrite it and exits with an error: +> ```console +> The output file 'examples/checksums.csv' already exists. +> Pass force=True (Python) or use the -f/--force flag (CLI) to overwrite it. +> ``` + Pass `--force` (or `-f`) to overwrite: +```bash +sum-buddy --force --output-file examples/checksums.csv examples/example_content/ +``` + - **Ignore Contents Based on Patterns:** ```bash sum-buddy --output-file examples/checksums.csv --ignore-file examples/.sbignore_except_txt examples/example_content/ @@ -183,6 +194,9 @@ get_checksums(input_path, output_file, ignore_file=ignore_file, algorithm=alg) # outputs status bar followed by # Checksums written to examples/checksums.csv +# If output_file already exists, get_checksums raises sumbuddy.exceptions.OutputFileExistsError (a subclass of FileExistsError); pass force=True to overwrite instead +get_checksums(input_path, output_file, force=True) + # To gather a list of file paths according to ignore/include patterns file_paths = gather_file_paths(input_path, ignore_file=ignore_file) # or file_paths = gather_file_paths(input_path, include_hidden=include_hidden) diff --git a/src/sumbuddy/__init__.py b/src/sumbuddy/__init__.py index d0012ad..ca4b2ee 100644 --- a/src/sumbuddy/__init__.py +++ b/src/sumbuddy/__init__.py @@ -1,7 +1,7 @@ from sumbuddy.__about__ import __version__ from sumbuddy.__main__ import get_checksums -from sumbuddy.mapper import Mapper from sumbuddy.hasher import Hasher +from sumbuddy.mapper import Mapper # Create instances of the classes mapper_instance = Mapper() @@ -11,4 +11,4 @@ gather_file_paths = mapper_instance.gather_file_paths checksum_file = hasher_instance.checksum_file -__all__ = ["__version__", "checksum_file", "get_checksums", "gather_file_paths"] +__all__ = ["__version__", "checksum_file", "gather_file_paths", "get_checksums"] diff --git a/src/sumbuddy/__main__.py b/src/sumbuddy/__main__.py index 163bd3e..a233405 100644 --- a/src/sumbuddy/__main__.py +++ b/src/sumbuddy/__main__.py @@ -1,16 +1,25 @@ import argparse -from sumbuddy.__about__ import __version__ -from sumbuddy.hasher import Hasher -from sumbuddy.mapper import Mapper -from sumbuddy.archive import ArchiveHandler -from sumbuddy.exceptions import EmptyInputDirectoryError, NoFilesAfterFilteringError, LengthUsedForFixedLengthHashError import csv import hashlib -from tqdm import tqdm -import sys import os +import sys +from contextlib import nullcontext + +from tqdm import tqdm -def get_checksums(input_path, output_filepath=None, ignore_file=None, include_hidden=False, algorithm='md5', length=None, archive_dive=True): +from sumbuddy.__about__ import __version__ +from sumbuddy.archive import ArchiveHandler +from sumbuddy.exceptions import ( + EmptyInputDirectoryError, + LengthUsedForFixedLengthHashError, + NoFilesAfterFilteringError, + OutputFileExistsError, +) +from sumbuddy.hasher import Hasher +from sumbuddy.mapper import Mapper + + +def get_checksums(input_path, output_filepath=None, ignore_file=None, include_hidden=False, algorithm='md5', length=None, archive_dive=True, force=False): """ Generate a CSV file with the filepath, filename, and checksum of all files in the input directory according to patterns to ignore. Checksum column is labeled by the selected algorithm (e.g., 'md5' or 'sha256'). @@ -23,7 +32,11 @@ def get_checksums(input_path, output_filepath=None, ignore_file=None, include_hi algorithm - String. Algorithm to use for checksums. Default: 'md5', see options with 'hashlib.algorithms_available'. length - Integer [conditionally optional]. Length of the digest for SHAKE (required) and BLAKE (optional) algorithms in bytes. archive_dive - Boolean [optional]. Whether to descend into archive files and hash their members. When False, archives are hashed as opaque files. Default: True. + force - Boolean [optional]. Whether to overwrite output_filepath if it already exists. Default is False, which raises OutputFileExistsError when the file exists. """ + if output_filepath and not force and os.path.exists(output_filepath): + raise OutputFileExistsError(output_filepath) + mapper = Mapper() if os.path.isfile(input_path): @@ -49,9 +62,12 @@ def get_checksums(input_path, output_filepath=None, ignore_file=None, include_hi hasher = Hasher(algorithm) archive_handler = ArchiveHandler() - output_stream = open(output_filepath, 'w', newline='') if output_filepath else sys.stdout - try: + with ( + open(output_filepath, 'w', newline='') + if output_filepath + else nullcontext(sys.stdout) + ) as output_stream: writer = csv.writer(output_stream) writer.writerow(["filepath", "filename", f"{algorithm}"]) @@ -75,9 +91,6 @@ def get_checksums(input_path, output_filepath=None, ignore_file=None, include_hi checksum = hasher.checksum_file(file_obj, algorithm=algorithm, length=length) writer.writerow([virtual_path, os.path.basename(member), checksum]) pbar.update(1) - finally: - if output_filepath: - output_stream.close() if output_filepath: print(f"{algorithm} checksums for {input_path} written to {output_filepath}") @@ -89,6 +102,7 @@ def main(): parser.add_argument("-V", "--version", action="version", version=f"%(prog)s {__version__}") parser.add_argument("input_path", help="File or directory to traverse for files") parser.add_argument("-o", "--output-file", help="Filepath for the output CSV file; defaults to stdout", default=None) + parser.add_argument("-f", "--force", action="store_true", help="Overwrite the output file if it already exists") group = parser.add_mutually_exclusive_group() group.add_argument("-i", "--ignore-file", help="Filepath for the ignore patterns file") group.add_argument("-H", "--include-hidden", action="store_true", help="Include hidden files") @@ -101,11 +115,6 @@ def main(): if args.output_file and not args.output_file.endswith('.csv'): parser.error("Output file is in CSV format; extension should be '.csv'") - if args.output_file and os.path.exists(args.output_file): - overwrite = input(f"Output file '{args.output_file}' already exists. Overwrite? [y/n]: ") - if overwrite.lower() != 'y': - sys.exit("Exited without executing") - try: get_checksums( args.input_path, @@ -115,8 +124,9 @@ def main(): algorithm=args.algorithm, length=args.length, archive_dive=args.archive_dive, + force=args.force, ) - except (EmptyInputDirectoryError, NoFilesAfterFilteringError, LengthUsedForFixedLengthHashError) as e: + except (EmptyInputDirectoryError, NoFilesAfterFilteringError, LengthUsedForFixedLengthHashError, OutputFileExistsError) as e: sys.exit(str(e)) diff --git a/src/sumbuddy/exceptions.py b/src/sumbuddy/exceptions.py index 745660d..5257c33 100644 --- a/src/sumbuddy/exceptions.py +++ b/src/sumbuddy/exceptions.py @@ -17,3 +17,8 @@ class LengthUsedForFixedLengthHashError(Exception): def __init__(self, algorithm): message = f"Length paremeter is not applicable for fixed-length algorithm '{algorithm}'." super().__init__(message) + +class OutputFileExistsError(FileExistsError): + def __init__(self, output_filepath): + message = f"The output file '{output_filepath}' already exists.\nPass force=True (Python) or use the -f/--force flag (CLI) to overwrite it." + super().__init__(message) diff --git a/src/sumbuddy/filter.py b/src/sumbuddy/filter.py index 02eba2f..b6db807 100644 --- a/src/sumbuddy/filter.py +++ b/src/sumbuddy/filter.py @@ -1,6 +1,8 @@ -import pathspec import os +import pathspec + + class Filter: def __init__(self): self.spec = None diff --git a/src/sumbuddy/hasher.py b/src/sumbuddy/hasher.py index 85012bd..d7e4d4c 100644 --- a/src/sumbuddy/hasher.py +++ b/src/sumbuddy/hasher.py @@ -1,6 +1,8 @@ import hashlib + from sumbuddy.exceptions import LengthUsedForFixedLengthHashError + class Hasher: def __init__(self, algorithm='md5'): self.algorithm = algorithm diff --git a/src/sumbuddy/mapper.py b/src/sumbuddy/mapper.py index 231e1e3..cf8c4ea 100644 --- a/src/sumbuddy/mapper.py +++ b/src/sumbuddy/mapper.py @@ -1,7 +1,13 @@ import os -from sumbuddy.filter import Filter -from sumbuddy.exceptions import EmptyInputDirectoryError, NoFilesAfterFilteringError, NotADirectoryError + from sumbuddy.archive import ArchiveHandler +from sumbuddy.exceptions import ( + EmptyInputDirectoryError, + NoFilesAfterFilteringError, + NotADirectoryError, +) +from sumbuddy.filter import Filter + class Mapper: def __init__(self): diff --git a/tests/test_archive.py b/tests/test_archive.py index 5772b46..f897221 100644 --- a/tests/test_archive.py +++ b/tests/test_archive.py @@ -8,9 +8,8 @@ import pytest from sumbuddy.archive import ArchiveHandler -from sumbuddy.mapper import Mapper from sumbuddy.hasher import Hasher - +from sumbuddy.mapper import Mapper TEST_ZIP = Path(__file__).parent / "test_archive.zip" diff --git a/tests/test_filter.py b/tests/test_filter.py index 60a060b..5c2c967 100644 --- a/tests/test_filter.py +++ b/tests/test_filter.py @@ -1,10 +1,12 @@ -import unittest import os +import unittest from unittest.mock import patch + +from sumbuddy.archive import ArchiveHandler from sumbuddy.filter import Filter from sumbuddy.hasher import Hasher -from sumbuddy.archive import ArchiveHandler + class TestFilter(unittest.TestCase): def setUp(self): diff --git a/tests/test_force.py b/tests/test_force.py new file mode 100644 index 0000000..c89fdf3 --- /dev/null +++ b/tests/test_force.py @@ -0,0 +1,70 @@ +import sys +from pathlib import Path +from unittest.mock import patch + +import pytest + +from sumbuddy import __main__ as sb_main +from sumbuddy import get_checksums +from sumbuddy.exceptions import OutputFileExistsError + +EXAMPLES_DIR = Path(__file__).parent.parent / "examples" + + +def test_get_checksums_refuses_existing_output(monkeypatch, tmp_path): + """Without force, an existing output file raises before any walking or hashing, leaving the file untouched.""" + monkeypatch.chdir(EXAMPLES_DIR) + output_file = tmp_path / "checksums.csv" + output_file.write_text("sentinel content") + + with patch("sumbuddy.Mapper.gather_file_paths") as mock_gather, pytest.raises(OutputFileExistsError) as excinfo: + get_checksums("example_content", str(output_file)) + + mock_gather.assert_not_called() + assert isinstance(excinfo.value, FileExistsError) + assert output_file.read_text() == "sentinel content" + assert "force=True" in str(excinfo.value) + assert "--force" in str(excinfo.value) + + +def test_get_checksums_overwrites_with_force(monkeypatch, tmp_path): + """With force=True, an existing output file is regenerated from scratch.""" + monkeypatch.chdir(EXAMPLES_DIR) + output_file = tmp_path / "checksums.csv" + output_file.write_text("stale content") + + get_checksums("example_content", str(output_file), force=True) + + actual = output_file.read_text() + assert "stale content" not in actual + expected = (EXAMPLES_DIR / "expected_outputs" / "default.csv").read_text().splitlines() + assert sorted(actual.splitlines()) == sorted(expected) + + +def test_main_exits_when_output_exists(monkeypatch, tmp_path): + """CLI exits with a clear message, without prompting, when the output file exists.""" + monkeypatch.chdir(EXAMPLES_DIR) + output_file = tmp_path / "checksums.csv" + output_file.write_text("sentinel content") + + monkeypatch.setattr("builtins.input", lambda *_: pytest.fail("CLI prompted for input")) + monkeypatch.setattr(sys, "argv", ["sum-buddy", "-o", str(output_file), "example_content"]) + with pytest.raises(SystemExit) as excinfo: + sb_main.main() + + assert "already exists" in str(excinfo.value) + assert "--force" in str(excinfo.value) + assert output_file.read_text() == "sentinel content" + + +@pytest.mark.parametrize("flag", ["-f", "--force"]) +def test_main_force_overwrites_existing_output(monkeypatch, tmp_path, flag): + """Both force spellings overwrite an existing output file end-to-end.""" + monkeypatch.chdir(EXAMPLES_DIR) + output_file = tmp_path / "checksums.csv" + output_file.write_text("stale content") + + monkeypatch.setattr(sys, "argv", ["sum-buddy", flag, "-o", str(output_file), "example_content"]) + sb_main.main() + + assert output_file.read_text().startswith("filepath,filename,md5") diff --git a/tests/test_getChecksums.py b/tests/test_getChecksums.py index 93156ee..765e2be 100644 --- a/tests/test_getChecksums.py +++ b/tests/test_getChecksums.py @@ -1,11 +1,12 @@ -import unittest -from unittest.mock import patch, mock_open import os import tempfile +import unittest from io import StringIO +from unittest.mock import mock_open, patch from sumbuddy import get_checksums + class TestGetChecksums(unittest.TestCase): def setUp(self): @@ -20,7 +21,7 @@ def setUp(self): @patch('builtins.open', new_callable=mock_open) @patch('sumbuddy.Hasher.checksum_file', return_value='dummychecksum') def test_get_checksums_single_file_to_file(self, mock_checksum, mock_open, mock_isfile): - get_checksums(self.input_path, self.output_filepath, ignore_file=None, include_hidden=False, algorithm=self.algorithm) + get_checksums(self.input_path, self.output_filepath, ignore_file=None, include_hidden=False, algorithm=self.algorithm, force=True) mock_open.assert_called_with(self.output_filepath, 'w', newline='') handle = mock_open() @@ -90,7 +91,7 @@ def test_get_checksums_normalizes_dot_single_file_path_in_stdout(self, mock_chec @patch('sumbuddy.Mapper.gather_file_paths', return_value=(['file1.txt', 'file2.txt'], [])) @patch('sumbuddy.Hasher.checksum_file', side_effect=lambda x, **kwargs: 'dummychecksum') def test_get_checksums_to_file(self, mock_checksum, mock_gather, mock_open, mock_exists, mock_abspath): - get_checksums(self.input_path, self.output_filepath, ignore_file=None, include_hidden=False, algorithm=self.algorithm) + get_checksums(self.input_path, self.output_filepath, ignore_file=None, include_hidden=False, algorithm=self.algorithm, force=True) mock_open.assert_called_with(self.output_filepath, 'w', newline='') handle = mock_open() diff --git a/tests/test_hasher.py b/tests/test_hasher.py index 91c229b..7935fcb 100644 --- a/tests/test_hasher.py +++ b/tests/test_hasher.py @@ -1,10 +1,13 @@ import hashlib import os import tempfile -import pytest from unittest.mock import patch -from sumbuddy.hasher import Hasher + +import pytest + from sumbuddy.exceptions import LengthUsedForFixedLengthHashError +from sumbuddy.hasher import Hasher + def is_algorithm_available(algorithm): return algorithm in hashlib.algorithms_available @@ -39,12 +42,14 @@ def is_algorithm_guaranteed(algorithm): @pytest.fixture(scope="module") def temp_file(): - temp_file = tempfile.NamedTemporaryFile(delete=False) - temp_file.write(b'This is a test file.') - temp_file_path = temp_file.name - temp_file.close() - yield temp_file_path - os.remove(temp_file_path) + with tempfile.NamedTemporaryFile(delete=False) as temp_file: + temp_file.write(b'This is a test file.') + temp_file_path = temp_file.name + + try: + yield temp_file_path + finally: + os.remove(temp_file_path) @pytest.mark.parametrize("algorithm", checksums.keys()) def test_all_algorithms_with_defaults(temp_file, algorithm): diff --git a/tests/test_mapper.py b/tests/test_mapper.py index aa919cb..a788aa8 100644 --- a/tests/test_mapper.py +++ b/tests/test_mapper.py @@ -1,10 +1,16 @@ import os import tempfile import unittest -from unittest.mock import patch, mock_open -from sumbuddy.mapper import Mapper +from unittest.mock import mock_open, patch + +from sumbuddy.exceptions import ( + EmptyInputDirectoryError, + NoFilesAfterFilteringError, + NotADirectoryError, +) from sumbuddy.filter import Filter -from sumbuddy.exceptions import EmptyInputDirectoryError, NoFilesAfterFilteringError, NotADirectoryError +from sumbuddy.mapper import Mapper + class TestMapper(unittest.TestCase): @patch('sumbuddy.filter.open', new_callable=mock_open, read_data="# This is a sample ignore file\n") @@ -35,7 +41,7 @@ def test_gather_file_paths(self): with open(os.path.join(subdir_path, '.hidden.txt'), 'w') as file: file.write('Some content') - regular_files, archive_files = mapper.gather_file_paths(temp_dir) + regular_files, _ = mapper.gather_file_paths(temp_dir) self.assertEqual(len(regular_files), 3) self.assertIn(os.path.join(temp_dir, 'file1.txt'), regular_files) self.assertIn(os.path.join(temp_dir, 'file2.txt'), regular_files) @@ -47,12 +53,12 @@ def test_gather_file_paths(self): with open(ignore_file_path, 'w') as ignore_file: ignore_file.write("*.txt") - regular_files, archive_files = mapper.gather_file_paths(temp_dir, ignore_file=ignore_file_path) + regular_files, _ = mapper.gather_file_paths(temp_dir, ignore_file=ignore_file_path) self.assertEqual(len(regular_files), 1) self.assertIn(os.path.join(temp_dir, 'ignore_file'), regular_files) # Test including hidden files - regular_files, archive_files = mapper.gather_file_paths(temp_dir, include_hidden=True) + regular_files, _ = mapper.gather_file_paths(temp_dir, include_hidden=True) self.assertEqual(len(regular_files), 6) self.assertIn(os.path.join(temp_dir, 'file1.txt'), regular_files) self.assertIn(os.path.join(temp_dir, 'file2.txt'), regular_files) @@ -61,7 +67,7 @@ def test_gather_file_paths(self): self.assertIn(os.path.join(subdir_path, 'file3.txt'), regular_files) self.assertIn(os.path.join(subdir_path, '.hidden.txt'), regular_files) - regular_files, archive_files = mapper.gather_file_paths(temp_dir) + regular_files, _ = mapper.gather_file_paths(temp_dir) self.assertEqual(len(regular_files), 4) self.assertIn(os.path.join(temp_dir, 'file1.txt'), regular_files) self.assertIn(os.path.join(temp_dir, 'file2.txt'), regular_files) @@ -70,9 +76,11 @@ def test_gather_file_paths(self): def test_gather_file_paths_empty(self): mapper = Mapper() - with tempfile.TemporaryDirectory() as temp_dir: - with self.assertRaises(EmptyInputDirectoryError): - mapper.gather_file_paths(temp_dir) + with ( + tempfile.TemporaryDirectory() as temp_dir, + self.assertRaises(EmptyInputDirectoryError), + ): + mapper.gather_file_paths(temp_dir) def test_gather_file_paths_filtered_files(self): mapper = Mapper() @@ -93,9 +101,11 @@ def test_gather_file_paths_filtered_files(self): def test_gather_file_paths_input_not_a_directory(self): mapper = Mapper() - with tempfile.NamedTemporaryFile() as temp_file: - with self.assertRaises(NotADirectoryError): - mapper.gather_file_paths(temp_file.name) + with ( + tempfile.NamedTemporaryFile() as temp_file, + self.assertRaises(NotADirectoryError), + ): + mapper.gather_file_paths(temp_file.name) if __name__ == '__main__': unittest.main()