diff --git a/src/cautiousrobot/roll_call.py b/src/cautiousrobot/roll_call.py index fc9aee9..d8c684a 100644 --- a/src/cautiousrobot/roll_call.py +++ b/src/cautiousrobot/roll_call.py @@ -1,10 +1,11 @@ -# CSV validation and input handling functions. - import os import sys +import logging from sumbuddy import gather_file_paths from sumbuddy.exceptions import EmptyInputDirectoryError +logger = logging.getLogger(__name__) + class RollCall: """Pre-download validation and directory verification.""" @@ -12,38 +13,67 @@ class RollCall: def __init__(self, csv_path=None): self.csv_path = csv_path + # --------------------------------------------------------- + # CSV extension validation + # --------------------------------------------------------- def validate_csv_extension(self, csv_path): """Validate that the input file has a .csv extension.""" if not csv_path.lower().endswith(".csv"): - sys.exit("Expected CSV for input file; extension should be '.csv'") + msg = "Expected CSV for input file; extension should be '.csv'" + logger.error(msg) + sys.exit(msg) + # --------------------------------------------------------- + # Filename uniqueness validation + # --------------------------------------------------------- def validate_filename_uniqueness(self, data_df, filename_col): """Validate that the filename column contains unique values.""" - if data_df.loc[data_df[filename_col].notna()].shape[0] != data_df[filename_col].nunique(): - sys.exit(f"{filename_col} is not a unique identifier for this dataset, please choose a column with unique values for filenames.") + non_null = data_df[filename_col].dropna() + if non_null.nunique() != len(non_null): + msg = ( + f"{filename_col} is not a unique identifier for this dataset, " + "please choose a column with unique values for filenames." + ) + logger.error(msg) + sys.exit(msg) + # --------------------------------------------------------- + # Missing filename handling + # --------------------------------------------------------- def handle_missing_filenames(self, data_df, filename_col, url_col): """Handle cases where URLs exist but filenames are missing.""" - # Rows where filename is missing but URL exists missing = data_df.loc[data_df[filename_col].isna() & data_df[url_col].notna()] + if missing.empty: + logger.info("No missing filenames detected.") return None - if len(missing) <= 5: - print("\n❗ Missing filenames detected (showing all):") + + count = len(missing) + + if count <= 5: + logger.info(f"Missing filenames detected in {count} rows — previewing.") + print("\nMissing filenames detected (showing all):") print(missing) else: - # Save path follows BuddyCheck pattern: _missing_filenames.csv csv_base = os.path.splitext(self.csv_path)[0] save_path = f"{csv_base}_missing_filenames.csv" missing.to_csv(save_path, index=False) + + logger.info( + f"Missing filenames detected in {count} rows — saved to {save_path}." + ) + print( - f"\n❗ Missing filenames detected for {len(missing)} rows.\n" + f"\nMissing filenames detected for {count} rows.\n" f"Because there are more than 5, they were saved to:\n {save_path}\n" f"Please correct the CSV and re-run." ) - # Return missing rows for reference + return missing - + + # --------------------------------------------------------- + # Expected column setup + # --------------------------------------------------------- def setup_expected_columns(self, args): """Set up the expected columns dictionary for CSV processing.""" subfolders = args.subdir_col @@ -51,64 +81,117 @@ def setup_expected_columns(self, args): "filename_col": args.img_name_col.lower(), "url_col": args.url_col.lower() } + if subfolders: subfolders = subfolders.lower() expected_cols["subfolders"] = subfolders + + logger.info("Expected columns configured.") return expected_cols, subfolders + # --------------------------------------------------------- + # Existing image detection + # --------------------------------------------------------- def check_existing_images(self, csv_path, img_dir, source_df, filename_col, subfolders=None): """Checks which files from the CSV already exist in the image directory.""" df = source_df.copy() + # Case 1: Directory does not exist if not os.path.exists(img_dir): + logger.info( + f"Image directory '{img_dir}' does not exist; all images will be downloaded." + ) df["in_img_dir"] = False - filtered_df = df[~df["in_img_dir"]].copy() - return df, filtered_df + return df, df.copy() + # Case 2: Directory exists but may be empty try: existing_files = gather_file_paths(img_dir) except EmptyInputDirectoryError: + logger.info( + f"Image directory '{img_dir}' is empty; all images will be downloaded." + ) existing_files = [] - existing_full_paths = {os.path.normpath(os.path.relpath(f, img_dir)) for f in existing_files} + # Normalize existing paths relative to img_dir + existing_full_paths = { + os.path.normpath(os.path.relpath(f, img_dir)) + for f in existing_files + } + # Build expected relative paths if subfolders: - raw_paths = df[subfolders].astype(str) + os.sep + df[filename_col].astype(str) + raw_paths = ( + df[subfolders].astype(str) + os.sep + df[filename_col].astype(str) + ) df["expected_path"] = raw_paths.apply(os.path.normpath) else: df["expected_path"] = df[filename_col].astype(str).apply(os.path.normpath) - expected_present = df["expected_path"].isin(existing_full_paths) - df["in_img_dir"] = expected_present.copy() + # Compare expected vs actual + df["in_img_dir"] = df["expected_path"].isin(existing_full_paths) df = df.drop(columns=["expected_path"]) filtered_df = df[~df["in_img_dir"]].copy() + # Case 3: All images already present if filtered_df.empty: - sys.exit(f"'{img_dir}' already contains all images. Exited without executing.") - else: - num_existing = len(existing_files) - print(f"There are {num_existing} of the desired files already in {img_dir}. Based on {csv_path}, {filtered_df.shape[0]} images should be downloaded.") + msg = ( + f"'{img_dir}' already contains all images listed in '{csv_path}'. Exiting." + ) + logger.info(msg) + sys.exit(msg) + + # Summary logging + num_present = df["in_img_dir"].sum() + num_missing = filtered_df.shape[0] + + logger.info( + f"{num_present} of the images listed in '{csv_path}' already exist in '{img_dir}'. " + f"{num_missing} images still need downloading." + ) + + print( + f"There are {num_present} of the desired files already in {img_dir}. " + f"Based on {csv_path}, {num_missing} images should be downloaded." + ) return df, filtered_df + # --------------------------------------------------------- + # Preview or save helper + # --------------------------------------------------------- def _preview_or_save(self, label, df): """Print up to 5 rows or save to CSV if larger.""" - if len(df) <= 5: - print(f"\n❗ {label.replace('_', ' ').title()} detected (showing all):") - print(df) - else: - csv_base = os.path.splitext(self.csv_path)[0] - save_path = f"{csv_base}_{label}.csv" - df.to_csv(save_path, index=False) - print( - f"\n❗ {len(df)} {label.replace('_', ' ')} detected.\n" - f"Full list saved to:\n {save_path}\n" + count = len(df) + + if count <= 5: + logger.info( + f"{label.replace('_', ' ')} detected in {count} rows — previewing." ) - + print(f"\n{label.replace('_', ' ').title()} detected (showing all):") + print(df) + return + + # Save for larger sets + csv_base = os.path.splitext(self.csv_path)[0] + save_path = f"{csv_base}_{label}.csv" + df.to_csv(save_path, index=False) + + logger.info( + f"{label.replace('_', ' ')} detected in {count} rows — saved to {save_path}." + ) + + print( + f"\n{count} {label.replace('_', ' ')} detected.\n" + f"Full list saved to:\n {save_path}\n" + ) + + # --------------------------------------------------------- + # Duplicate checksum detection + # --------------------------------------------------------- def check_duplicate_checksums(self, data_df, hash_col, ignore_duplicates=False): """Detect duplicate checksum values and optionally block execution.""" - # Only consider non-null hashes dupes = ( data_df[data_df[hash_col].notna()] .groupby(hash_col) @@ -116,28 +199,37 @@ def check_duplicate_checksums(self, data_df, hash_col, ignore_duplicates=False): ) if dupes.empty: + logger.info("No duplicate checksums detected.") print("✔ No duplicate checksums detected.") return - # If duplicates exist, preview or save + count = len(dupes) + logger.warning(f"Duplicate checksums detected in {count} rows.") self._preview_or_save("duplicate_checksums", dupes) if ignore_duplicates: + logger.warning("Ignoring duplicate checksums due to --ignore-duplicates flag.") print( - f"\n⚠ Duplicate checksums detected ({len(dupes)} rows), " + f"\n⚠ Duplicate checksums detected in {count} rows, " f"but --ignore-duplicates was passed. Continuing.\n" ) return - # Default behavior: block execution - sys.exit( - "❗ Duplicate checksums detected. " + msg = ( + f"❗ Duplicate checksums detected in {count} rows. " "Use --ignore-duplicates to allow downloading duplicates." ) + logger.error(msg) + sys.exit(msg) + # --------------------------------------------------------- + # Download summary + # --------------------------------------------------------- def print_download_summary(self, img_dir, downsample_dir, subfolders, num_images): """Print a summary of where images and downsized images will be saved.""" - print("\n📦 Download Summary") + logger.info("Printing download summary.") + + print("\nDownload Summary") print("--------------------") print(f"Images will be downloaded to: {img_dir}") @@ -151,4 +243,4 @@ def print_download_summary(self, img_dir, downsample_dir, subfolders, num_images else: print("Subfolders: none") - print(f"Images to download: {num_images}\n") \ No newline at end of file + print(f"Images to download: {num_images}\n") diff --git a/tests/test_download_images.py b/tests/test_download_images.py index f9cbbdb..32dc581 100644 --- a/tests/test_download_images.py +++ b/tests/test_download_images.py @@ -391,7 +391,7 @@ def test_main_non_unique_filenames(self, mock_process_csv, mock_parse_args): mock_process_csv.return_value = mock_data - with self.assertRaises(SystemExit) as cm: + with self.assertRaises(SystemExit) as cm: main() self.assertEqual( diff --git a/tests/test_roll_call.py b/tests/test_roll_call.py index 98bf5cf..105252f 100644 --- a/tests/test_roll_call.py +++ b/tests/test_roll_call.py @@ -1,520 +1,210 @@ import unittest -from unittest.mock import patch, MagicMock +from unittest.mock import patch import pandas as pd from cautiousrobot.roll_call import RollCall +from sumbuddy.exceptions import EmptyInputDirectoryError -class TestValidateCsvExtension(unittest.TestCase): - """Test CSV file extension validation.""" - - def setUp(self): - self.rollcall = RollCall() - - def test_valid_csv_extension(self): - """Should not raise exception for valid .csv file.""" - # Should not raise any exception - try: - self.rollcall.validate_csv_extension("data.csv") - except SystemExit: - self.fail("validate_csv_extension raised SystemExit unexpectedly") - - def test_csv_extension_case_sensitive(self): - """Should accept .csv regardless of path content.""" - try: - self.rollcall.validate_csv_extension("MyData.csv") - self.rollcall.validate_csv_extension("path/to/data.csv") - except SystemExit: - self.fail("validate_csv_extension raised SystemExit unexpectedly") - - def test_invalid_txt_extension(self): - """Should exit if file has .txt extension.""" - with self.assertRaises(SystemExit) as cm: - self.rollcall.validate_csv_extension("data.txt") - self.assertIn("csv", cm.exception.code.lower()) - - def test_invalid_xlsx_extension(self): - """Should exit if file has .xlsx extension.""" - with self.assertRaises(SystemExit) as cm: - self.rollcall.validate_csv_extension("data.xlsx") - self.assertIn("csv", cm.exception.code.lower()) - - def test_invalid_no_extension(self): - """Should exit if file has no extension.""" - with self.assertRaises(SystemExit) as cm: - self.rollcall.validate_csv_extension("data") - self.assertIn("csv", cm.exception.code.lower()) - - def test_invalid_csv_as_substring(self): - """Should require .csv as actual extension, not substring.""" - with self.assertRaises(SystemExit) as cm: - self.rollcall.validate_csv_extension("data.csv.backup") - self.assertIn("csv", cm.exception.code.lower()) - - -class TestValidateFilenameUniqueness(unittest.TestCase): - """Test filename uniqueness validation.""" - - def setUp(self): - self.rollcall = RollCall() - - def test_unique_filenames(self): - """Should not raise exception for unique filenames.""" - df = pd.DataFrame({ - "filename": ["image1.jpg", "image2.jpg", "image3.jpg"] - }) - try: - self.rollcall.validate_filename_uniqueness(df, "filename") - except SystemExit: - self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") - - def test_duplicate_filenames(self): - """Should exit if filenames are not unique.""" - df = pd.DataFrame({ - "filename": ["image1.jpg", "image1.jpg", "image2.jpg"] - }) - with self.assertRaises(SystemExit) as cm: - self.rollcall.validate_filename_uniqueness(df, "filename") - self.assertIn("unique identifier", cm.exception.code.lower()) - - def test_filenames_with_missing_values(self): - """Should only count non-null filenames for uniqueness.""" - df = pd.DataFrame({ - "filename": ["image1.jpg", None, "image2.jpg", None] - }) - try: - self.rollcall.validate_filename_uniqueness(df, "filename") - except SystemExit: - self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") - - def test_all_missing_filenames(self): - """Should not raise exception if all filenames are missing.""" - df = pd.DataFrame({ - "filename": [None, None, None] - }) - try: - self.rollcall.validate_filename_uniqueness(df, "filename") - except SystemExit: - self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") - - def test_single_filename(self): - """Should not raise exception for single unique filename.""" - df = pd.DataFrame({ - "filename": ["image1.jpg"] - }) - try: - self.rollcall.validate_filename_uniqueness(df, "filename") - except SystemExit: - self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") - - def test_empty_dataframe(self): - """Should not raise exception for empty dataframe.""" - df = pd.DataFrame({ - "filename": [] - }) - try: - self.rollcall.validate_filename_uniqueness(df, "filename") - except SystemExit: - self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") - - def test_custom_column_name(self): - """Should work with custom column names.""" - df = pd.DataFrame({ - "image_name": ["photo1.jpg", "photo2.jpg", "photo3.jpg"] - }) - try: - self.rollcall.validate_filename_uniqueness(df, "image_name") - except SystemExit: - self.fail("validate_filename_uniqueness raised SystemExit unexpectedly") - - def test_duplicate_with_custom_column(self): - """Should detect duplicates in custom column names.""" - df = pd.DataFrame({ - "image_id": ["img_001", "img_001", "img_002"] - }) - with self.assertRaises(SystemExit) as cm: - self.rollcall.validate_filename_uniqueness(df, "image_id") - self.assertIn("unique identifier", cm.exception.code.lower()) - - -class TestHandleMissingFilenames(unittest.TestCase): - """Test handling of missing filename values under new non-interactive behavior.""" - - def setUp(self): - self.filename_col = "filename" - self.url_col = "file_url" - # RollCall now requires csv_path for saving missing CSVs - self.rollcall = RollCall(csv_path="testdata.csv") - - @patch("builtins.print") - def test_no_missing_filenames(self, mock_print): - """Should not print anything when no filenames are missing.""" - df = pd.DataFrame({ - self.filename_col: ["image1.jpg", "image2.jpg"], - self.url_col: ["http://url1.com", "http://url2.com"] - }) - result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) - self.assertIsNone(result) - mock_print.assert_not_called() +class TestRollCall(unittest.TestCase): - @patch("builtins.print") - def test_missing_filenames_prints_when_five_or_fewer(self, mock_print): - """Should print missing rows when count <= 5.""" - df = pd.DataFrame({ - self.filename_col: [None, None, "img3.jpg"], - self.url_col: ["url1", "url2", "url3"] - }) - result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) - - self.assertEqual(len(result), 2) - mock_print.assert_called() # printed the missing rows - - @patch("pandas.DataFrame.to_csv") - @patch("builtins.print") - def test_missing_filenames_saved_when_more_than_five(self, mock_print, mock_to_csv): - """Should save missing rows to CSV when count > 5.""" - df = pd.DataFrame({ - self.filename_col: [None] * 6, - self.url_col: ["url"] * 6 - }) + # --------------------------------------------------------- + # validate_csv_extension + # --------------------------------------------------------- + def test_validate_csv_extension_valid(self): + rc = RollCall() + rc.validate_csv_extension("file.csv") # should not raise - result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) + def test_validate_csv_extension_uppercase(self): + rc = RollCall() + rc.validate_csv_extension("FILE.CSV") # should not raise - self.assertEqual(len(result), 6) - mock_to_csv.assert_called_once() - args, kwargs = mock_to_csv.call_args - self.assertIn("testdata_missing_filenames.csv", args[0]) + def test_validate_csv_extension_invalid(self): + rc = RollCall() + with self.assertRaises(SystemExit): + rc.validate_csv_extension("file.txt") + + # --------------------------------------------------------- + # validate_filename_uniqueness + # --------------------------------------------------------- + def test_validate_filename_uniqueness_unique(self): + df = pd.DataFrame({"file": ["a.jpg", "b.jpg", "c.jpg"]}) + rc = RollCall() + rc.validate_filename_uniqueness(df, "file") + + def test_validate_filename_uniqueness_all_null(self): + df = pd.DataFrame({"file": [None, None]}) + rc = RollCall() + rc.validate_filename_uniqueness(df, "file") # should not raise + + def test_validate_filename_uniqueness_duplicates(self): + df = pd.DataFrame({"file": ["a.jpg", "a.jpg", "b.jpg"]}) + rc = RollCall() + with self.assertRaises(SystemExit): + rc.validate_filename_uniqueness(df, "file") - @patch("builtins.print") - def test_missing_filenames_no_urls(self, mock_print): - """Should not print anything when URLs are missing (no actionable missing filenames).""" + # --------------------------------------------------------- + # handle_missing_filenames + # --------------------------------------------------------- + def test_handle_missing_filenames_none_multiple(self): df = pd.DataFrame({ - self.filename_col: ["img1.jpg", None, "img3.jpg"], - self.url_col: [None, None, None] + "file": ["a.jpg", "b.jpg"], + "url": ["x", "y"] }) - result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) - + rc = RollCall(csv_path="dummy.csv") + result = rc.handle_missing_filenames(df, "file", "url") self.assertIsNone(result) - mock_print.assert_not_called() - @patch("builtins.print") - def test_missing_filenames_empty_dataframe(self, mock_print): - """Should not print anything for empty dataframe.""" + def test_handle_missing_filenames_some_missing(self): df = pd.DataFrame({ - self.filename_col: [], - self.url_col: [] + "file": ["a.jpg", None, "c.jpg"], + "url": ["x", "y", "z"] }) - result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) + rc = RollCall(csv_path="dummy.csv") + result = rc.handle_missing_filenames(df, "file", "url") + self.assertEqual(len(result), 1) - self.assertIsNone(result) - mock_print.assert_not_called() - - @patch("builtins.print") - def test_missing_filenames_returns_correct_subset(self, mock_print): - """Should return only rows with missing filenames and valid URLs.""" + def test_handle_missing_filenames_all_missing(self): df = pd.DataFrame({ - self.filename_col: [None, "img2.jpg", None, "img4.jpg"], - self.url_col: ["url1", None, "url3", "url4"] + "file": [None, None], + "url": ["x", "y"] }) - - result = self.rollcall.handle_missing_filenames(df, self.filename_col, self.url_col) - - # Only rows 0 and 2 qualify + rc = RollCall(csv_path="dummy.csv") + result = rc.handle_missing_filenames(df, "file", "url") self.assertEqual(len(result), 2) - self.assertListEqual(result.index.tolist(), [0, 2]) - - -class TestSetupExpectedColumns(unittest.TestCase): - """Test expected columns setup.""" - - def setUp(self): - self.rollcall = RollCall() - - def test_basic_columns_setup(self): - """Should return expected columns with basic arguments.""" - args = MagicMock() - args.img_name_col = "filename" - args.url_col = "file_url" - args.subdir_col = None - - expected_cols, subfolders = self.rollcall.setup_expected_columns(args) - - self.assertEqual(expected_cols["filename_col"], "filename") - self.assertEqual(expected_cols["url_col"], "file_url") - self.assertNotIn("subfolders", expected_cols) - self.assertIsNone(subfolders) - - def test_columns_lowercase_conversion(self): - """Should convert column names to lowercase.""" - args = MagicMock() - args.img_name_col = "FileName" - args.url_col = "FileURL" - args.subdir_col = None - - expected_cols, subfolders = self.rollcall.setup_expected_columns(args) - - self.assertEqual(expected_cols["filename_col"], "filename") - self.assertEqual(expected_cols["url_col"], "fileurl") - - def test_subdir_column_setup(self): - """Should include subdir column when provided.""" - args = MagicMock() - args.img_name_col = "filename" - args.url_col = "file_url" - args.subdir_col = "species" - - expected_cols, subfolders = self.rollcall.setup_expected_columns(args) - self.assertEqual(expected_cols["subfolders"], "species") - self.assertEqual(subfolders, "species") - - def test_subdir_column_lowercase(self): - """Should convert subdir column name to lowercase.""" - args = MagicMock() - args.img_name_col = "filename" - args.url_col = "file_url" - args.subdir_col = "Species" - - expected_cols, subfolders = self.rollcall.setup_expected_columns(args) - - self.assertEqual(expected_cols["subfolders"], "species") - self.assertEqual(subfolders, "species") + # --------------------------------------------------------- + # setup_expected_columns + # --------------------------------------------------------- + class Args: + img_name_col = "FILENAME" + url_col = "URL" + subdir_col = "CATEGORY" + + def test_setup_expected_columns(self): + rc = RollCall() + expected, subfolders = rc.setup_expected_columns(self.Args) + self.assertEqual(expected["filename_col"], "filename") + self.assertEqual(expected["url_col"], "url") + self.assertEqual(expected["subfolders"], "category") + self.assertEqual(subfolders, "category") - def test_custom_column_names_with_subdir(self): - """Should handle custom column names with subdir.""" - args = MagicMock() - args.img_name_col = "ImageID" - args.url_col = "DownloadURL" - args.subdir_col = "Category" + # --------------------------------------------------------- + # check_existing_images + # --------------------------------------------------------- + @patch("os.path.exists", return_value=False) + def test_check_existing_images_directory_missing(self, mock_exists): + df = pd.DataFrame({"file": ["a.jpg", "b.jpg"]}) + rc = RollCall(csv_path="dummy.csv") + full_df, filtered = rc.check_existing_images( + "dummy.csv", "missing_dir", df, "file" + ) + self.assertEqual(len(filtered), 2) - expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + @patch("cautiousrobot.roll_call.gather_file_paths") + def test_check_existing_images_empty_directory(self, mock_gather): + mock_gather.side_effect = EmptyInputDirectoryError("fake_dir") - self.assertEqual(expected_cols["filename_col"], "imageid") - self.assertEqual(expected_cols["url_col"], "downloadurl") - self.assertEqual(expected_cols["subfolders"], "category") - self.assertEqual(subfolders, "category") + df = pd.DataFrame({"file": ["a.jpg", "b.jpg"]}) + rc = RollCall(csv_path="dummy.csv") - def test_returns_subfolders_none_when_not_provided(self): - """Should return None for subfolders when not provided in args.""" - args = MagicMock() - args.img_name_col = "filename" - args.url_col = "file_url" - args.subdir_col = None + full_df, filtered = rc.check_existing_images( + "dummy.csv", ".", df, "file" + ) - expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + self.assertEqual(len(filtered), 2) - self.assertIsNone(subfolders) - self.assertNotIn("subfolders", expected_cols) + @patch("cautiousrobot.roll_call.gather_file_paths") + def test_check_existing_images_none_exist(self, mock_gather): + mock_gather.return_value = [] + df = pd.DataFrame({"file": ["a.jpg", "b.jpg"]}) + rc = RollCall(csv_path="dummy.csv") + full_df, filtered = rc.check_existing_images( + "dummy.csv", ".", df, "file" + ) + self.assertEqual(len(filtered), 2) + + @patch("cautiousrobot.roll_call.gather_file_paths") + def test_check_existing_images_some_exist(self, mock_gather): + mock_gather.return_value = ["a.jpg"] + df = pd.DataFrame({"file": ["a.jpg", "b.jpg"]}) + rc = RollCall(csv_path="dummy.csv") + full_df, filtered = rc.check_existing_images( + "dummy.csv", ".", df, "file" + ) + self.assertEqual(len(filtered), 1) + self.assertIn("b.jpg", filtered["file"].values) - def test_returns_subfolders_value_when_provided(self): - """Should return subfolders value when provided.""" - args = MagicMock() - args.img_name_col = "filename" - args.url_col = "file_url" - args.subdir_col = "region" + @patch("cautiousrobot.roll_call.gather_file_paths") + def test_check_existing_images_subfolders_all_exist(self, mock_gather): + mock_gather.return_value = ["cat/a.jpg"] - expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + df = pd.DataFrame({"file": ["a.jpg"], "category": ["cat"]}) + rc = RollCall(csv_path="dummy.csv") - self.assertEqual(subfolders, "region") - self.assertIn("subfolders", expected_cols) + with self.assertRaises(SystemExit): + rc.check_existing_images( + "dummy.csv", ".", df, "file", subfolders="category" + ) - def test_dictionary_structure(self): - """Should return correctly structured dictionary.""" - args = MagicMock() - args.img_name_col = "fname" - args.url_col = "url" - args.subdir_col = "folder" - expected_cols, subfolders = self.rollcall.setup_expected_columns(args) + @patch("cautiousrobot.roll_call.gather_file_paths") + def test_check_existing_images_subfolders(self, mock_gather): + mock_gather.return_value = [] # nothing exists - # Check dictionary has correct keys - self.assertIn("filename_col", expected_cols) - self.assertIn("url_col", expected_cols) - self.assertIn("subfolders", expected_cols) + df = pd.DataFrame({"file": ["a.jpg"], "category": ["cat"]}) + rc = RollCall(csv_path="dummy.csv") - # Check values are correct - self.assertEqual(expected_cols["filename_col"], "fname") - self.assertEqual(expected_cols["url_col"], "url") - self.assertEqual(expected_cols["subfolders"], "folder") + full_df, filtered = rc.check_existing_images( + "dummy.csv", ".", df, "file", subfolders="category" + ) -class TestCheckDuplicateChecksums(unittest.TestCase): - """Tests for RollCall.check_duplicate_checksums.""" + self.assertEqual(len(filtered), 1) + self.assertIn("a.jpg", filtered["file"].values) - def setUp(self): - self.rollcall = RollCall(csv_path="testdata.csv") - @patch("builtins.print") - def test_no_duplicates(self, mock_print): - """Should not exit when no duplicate checksums exist.""" + # --------------------------------------------------------- + # check_duplicate_checksums + # --------------------------------------------------------- + def test_check_duplicate_checksums_none(self): df = pd.DataFrame({"hash": ["a", "b", "c"]}) + rc = RollCall() + rc.check_duplicate_checksums(df, "hash") - try: - self.rollcall.check_duplicate_checksums(df, "hash") - except SystemExit: - self.fail("Unexpected SystemExit for no duplicates") - - mock_print.assert_called_once() # "✔ No duplicate checksums detected." - - @patch("builtins.print") - def test_duplicates_without_ignore(self, mock_print): - """Should exit when duplicates exist and ignore flag is False.""" + def test_check_duplicate_checksums_some(self): df = pd.DataFrame({"hash": ["a", "a", "b"]}) - + rc = RollCall(csv_path="dummy.csv") with self.assertRaises(SystemExit): - self.rollcall.check_duplicate_checksums(df, "hash", ignore_duplicates=False) + rc.check_duplicate_checksums(df, "hash") - # Should have printed preview/save message - mock_print.assert_called() - - @patch("builtins.print") - def test_duplicates_with_ignore(self, mock_print): - """Should not exit when duplicates exist and ignore flag is True.""" + def test_check_duplicate_checksums_ignore(self): df = pd.DataFrame({"hash": ["a", "a", "b"]}) + rc = RollCall(csv_path="dummy.csv") + rc.check_duplicate_checksums(df, "hash", ignore_duplicates=True) - try: - self.rollcall.check_duplicate_checksums(df, "hash", ignore_duplicates=True) - except SystemExit: - self.fail("Unexpected SystemExit when ignore_duplicates=True") - - # Should print warning - self.assertTrue(any("ignore-duplicates" in call.args[0] for call in mock_print.call_args_list)) - - @patch("builtins.print") - def test_preview_mode_for_small_duplicate_set(self, mock_print): - """Should print preview when <=5 duplicate rows.""" - df = pd.DataFrame({"hash": ["x", "x", "x"]}) - + def test_check_duplicate_checksums_multiple_groups(self): + df = pd.DataFrame({"hash": ["a", "a", "b", "b"]}) + rc = RollCall(csv_path="dummy.csv") with self.assertRaises(SystemExit): - self.rollcall.check_duplicate_checksums(df, "hash") + rc.check_duplicate_checksums(df, "hash") - # Should print the DataFrame preview - mock_print.assert_called() - - @patch("pandas.DataFrame.to_csv") + # --------------------------------------------------------- + # _preview_or_save + # --------------------------------------------------------- @patch("builtins.print") - def test_save_mode_for_large_duplicate_set(self, mock_print, mock_to_csv): - """Should save CSV when >5 duplicate rows.""" - df = pd.DataFrame({"hash": ["x"] * 10}) + def test_preview_or_save_small(self, mock_print): + df = pd.DataFrame({"x": [1, 2]}) + rc = RollCall(csv_path="dummy.csv") + rc._preview_or_save("test_label", df) + self.assertTrue(mock_print.called) - with self.assertRaises(SystemExit): - self.rollcall.check_duplicate_checksums(df, "hash") - - mock_to_csv.assert_called_once() - args, kwargs = mock_to_csv.call_args - self.assertIn("testdata_duplicate_checksums.csv", args[0]) - -class TestPreviewOrSave(unittest.TestCase): - """Tests for RollCall._preview_or_save.""" - - def setUp(self): - self.rollcall = RollCall(csv_path="testdata.csv") - - @patch("builtins.print") @patch("pandas.DataFrame.to_csv") - def test_preview_small_dataframe(self, mock_to_csv, mock_print): - """Should print when <=5 rows.""" - df = pd.DataFrame({"x": [1, 2, 3]}) - - self.rollcall._preview_or_save("test_label", df) - - mock_print.assert_called() - mock_to_csv.assert_not_called() - @patch("builtins.print") - @patch("pandas.DataFrame.to_csv") - def test_save_large_dataframe(self, mock_to_csv, mock_print): - """Should save CSV when >5 rows.""" + def test_preview_or_save_large(self, mock_print, mock_csv): df = pd.DataFrame({"x": list(range(10))}) - - self.rollcall._preview_or_save("test_label", df) - - mock_to_csv.assert_called_once() - args, kwargs = mock_to_csv.call_args - self.assertIn("testdata_test_label.csv", args[0]) - mock_print.assert_called() - -class TestPrintDownloadSummary(unittest.TestCase): - """Tests for RollCall.print_download_summary.""" - - def setUp(self): - self.rollcall = RollCall(csv_path="testdata.csv") - - @patch("builtins.print") - def test_summary_with_downsampling_and_subfolders(self, mock_print): - """Should print full summary including downsized path and subfolders.""" - self.rollcall.print_download_summary( - img_dir="/images", - downsample_dir="/images_downsized", - subfolders="species", - num_images=42 - ) - - printed = " ".join(call.args[0] for call in mock_print.call_args_list) - - self.assertIn("Images will be downloaded to: /images", printed) - self.assertIn("Downsampled images will be saved to: /images_downsized", printed) - self.assertIn("Subfolders enabled: species", printed) - self.assertIn("Images to download: 42", printed) - - @patch("builtins.print") - def test_summary_without_downsampling(self, mock_print): - """Should indicate that downsampling is not requested.""" - self.rollcall.print_download_summary( - img_dir="/images", - downsample_dir=None, - subfolders=None, - num_images=10 - ) - - printed = " ".join(call.args[0] for call in mock_print.call_args_list) - - self.assertIn("Images will be downloaded to: /images", printed) - self.assertIn("Downsampled images: not requested", printed) - self.assertIn("Subfolders: none", printed) - self.assertIn("Images to download: 10", printed) - - @patch("builtins.print") - def test_summary_with_subfolders_only(self, mock_print): - """Should print subfolder info even without downsampling.""" - self.rollcall.print_download_summary( - img_dir="/images", - downsample_dir=None, - subfolders="category", - num_images=5 - ) - - printed = " ".join(call.args[0] for call in mock_print.call_args_list) - - self.assertIn("Subfolders enabled: category", printed) - self.assertIn("Images to download: 5", printed) - - @patch("builtins.print") - def test_summary_zero_images(self, mock_print): - """Should correctly print zero image count.""" - self.rollcall.print_download_summary( - img_dir="/images", - downsample_dir=None, - subfolders=None, - num_images=0 - ) - - printed = " ".join(call.args[0] for call in mock_print.call_args_list) - self.assertIn("Images to download: 0", printed) - - @patch("builtins.print") - def test_summary_formatting_header(self, mock_print): - """Should print the summary header and separator.""" - self.rollcall.print_download_summary( - img_dir="/images", - downsample_dir=None, - subfolders=None, - num_images=1 - ) - - printed_lines = [call.args[0] for call in mock_print.call_args_list] - - self.assertIn("📦 Download Summary", printed_lines[0]) - self.assertIn("--------------------", printed_lines[1]) + rc = RollCall(csv_path="dummy.csv") + rc._preview_or_save("test_label", df) + mock_csv.assert_called_once() if __name__ == "__main__":