tests/test_duplicate_file_report.py — Duplicate File and Photo Report

# By Zero Width Lab (zerowidthlab.com)
"""Synthetic local tests for the duplicate report."""

from __future__ import annotations

import json
import hashlib
import os
import stat
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

__author__ = "Zero Width Lab"

ROOT = Path(__file__).resolve().parent.parent
SCRIPT = ROOT / "scripts" / "duplicate_file_report.py"
sys.path.insert(0, str(SCRIPT.parent))
import duplicate_file_report as report  # noqa: E402


def run_cli(*args: str) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [sys.executable, str(SCRIPT), "scan", *args],
        text=True, capture_output=True, check=False,
    )


class DuplicateReportTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        self.addCleanup(self.temporary.cleanup)
        self.root = Path(self.temporary.name).resolve()

    def write(self, name: str, data: bytes) -> Path:
        path = self.root / name
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(data)
        return path

    def test_duplicate_groups_and_exit_codes(self) -> None:
        self.write("first.txt", b"same")
        self.write("second.txt", b"same")
        self.write("different.txt", b"else")
        process = run_cli(str(self.root), "--json")
        self.assertEqual(process.returncode, 1, process.stderr)
        result = json.loads(process.stdout)
        self.assertEqual(result["reclaimable_bytes"], 4)
        self.assertEqual(len(result["groups"]), 1)
        self.assertEqual(len(result["groups"][0]["files"]), 2)
        self.assertFalse(result["incomplete"])
        self.assertEqual(run_cli(str(self.root), "--min-size", "5").returncode, 0)

    def test_partial_and_full_hashes(self) -> None:
        prefix = b"a" * report.HASH_CHUNK
        suffix = b"z" * report.HASH_CHUNK
        self.write("a", prefix + b"middle-a" + suffix)
        self.write("b", prefix + b"middle-b" + suffix)
        self.assertEqual(run_cli(str(self.root), "--json").returncode, 0)
        overlap = self.write("overlap", b"o" * (report.HASH_CHUNK + 7))
        expected = hashlib.sha256(overlap.read_bytes()[:report.HASH_CHUNK] + overlap.read_bytes()[-report.HASH_CHUNK:]).hexdigest()
        self.assertEqual(report.file_hash(str(overlap), overlap.stat().st_size, True), expected)

    def test_hidden_packages_and_symlinks_are_skipped(self) -> None:
        self.write("visible", b"same")
        self.write(".hidden", b"same")
        self.write("Album.photoslibrary/photo", b"same")
        (self.root / "shortcut").symlink_to(self.root / "visible")
        basic = json.loads(run_cli(str(self.root), "--json").stdout)
        self.assertEqual(len(basic["groups"]), 0)
        self.assertIn("package_directory", {item["reason"] for item in basic["skipped"]})
        included = json.loads(run_cli(str(self.root), "--include-hidden", "--include-packages", "--json").stdout)
        self.assertEqual(len(included["groups"][0]["files"]), 3)
        explicit_package = json.loads(run_cli(str(self.root / "Album.photoslibrary"), "--json").stdout)
        self.assertEqual(explicit_package["scanned_files"], 0)

    def test_package_suffix_uses_photos_bundle_name(self) -> None:
        self.write("Album.photoslibrary/photo", b"same")
        self.write("Album.photolibrary/photo", b"same")
        result = json.loads(run_cli(str(self.root), "--json").stdout)
        self.assertEqual(result["scanned_files"], 1)
        self.assertEqual(
            [item["path"] for item in result["skipped"] if item["reason"] == "package_directory"],
            [str(self.root / "Album.photoslibrary")],
        )

    def test_dataless_flag_is_never_read(self) -> None:
        path = self.write("cloud", b"content")
        real_stat = os.stat

        class FlaggedStat:
            def __init__(self, original: os.stat_result) -> None:
                self.st_mode = original.st_mode
                self.st_size = original.st_size
                self.st_mtime_ns = original.st_mtime_ns
                self.st_flags = report.SF_DATALESS

        def fake_stat(name: str, *, follow_symlinks: bool = True):
            value = real_stat(name, follow_symlinks=follow_symlinks)
            return FlaggedStat(value) if name == str(path) else value

        with patch.object(report, "file_hash", side_effect=AssertionError("dataless read")):
            result = report.scan_paths([str(self.root)], stat_func=fake_stat)
        self.assertEqual(result["not_downloaded"], 1)
        self.assertTrue(result["incomplete"])
        self.assertEqual(result["skipped"][0]["reason"], "not_downloaded")

    def test_unreadable_file_is_partial(self) -> None:
        path = self.write("unreadable", b"same")
        self.write("other", b"same")
        original = report.file_hash

        def fail_one(name: str, size: int, partial: bool, expected=None) -> str:
            if name == str(path):
                raise PermissionError("synthetic refusal")
            return original(name, size, partial, expected)

        with patch.object(report, "file_hash", side_effect=fail_one):
            result = report.scan_paths([str(self.root)])
        self.assertEqual(result["unreadable"], 1)
        self.assertTrue(result["incomplete"])
        self.assertEqual(len(result["groups"]), 0)

    def test_unreadable_subdirectory_is_partial(self) -> None:
        child = self.root / "unreadable"
        child.mkdir()
        original_scandir = os.scandir

        def fail_child(path: str):
            if isinstance(path, int) and os.fstat(path).st_ino == child.stat().st_ino:
                raise PermissionError("synthetic refusal")
            return original_scandir(path)

        with patch.object(report.os, "scandir", side_effect=fail_child):
            result = report.scan_paths([str(self.root)])
        self.assertEqual(result["unreadable"], 1)
        self.assertTrue(result["incomplete"])
        self.assertIn({"path": str(child), "reason": "unreadable"}, result["skipped"])

    def test_unreadable_root_cli_exit_and_partial_opt_in(self) -> None:
        missing = str(self.root / "missing")
        self.assertEqual(run_cli(missing, "--json").returncode, 2)
        self.assertEqual(run_cli(missing, "--allow-partial", "--json").returncode, 0)

    def test_report_requires_matching_dry_run_and_create_only(self) -> None:
        self.write("a", b"same")
        self.write("b", b"same")
        destination = self.root / "result.json"
        base = (str(self.root), "--report", str(destination))
        self.assertEqual(run_cli(*base).returncode, 2)
        self.assertFalse(destination.exists())
        dry = run_cli(*base, "--dry-run")
        self.assertEqual(dry.returncode, 1, dry.stderr)
        self.assertFalse(destination.exists())
        digest = dry.stderr.strip().split()[-1]
        self.assertEqual(run_cli(*base, "--approve", "0" * 64).returncode, 2)
        live = run_cli(*base, "--approve", digest)
        self.assertEqual(live.returncode, 1, live.stderr)
        self.assertEqual(json.loads(destination.read_text())["reclaimable_bytes"], 4)
        self.assertEqual(run_cli(*base, "--approve", digest).returncode, 2)

    def test_report_refuses_symlink_and_changed_plan(self) -> None:
        source = self.write("source", b"same")
        self.write("copy", b"same")
        destination = self.root / "result.json"
        dry = run_cli(str(self.root), "--report", str(destination), "--dry-run")
        digest = dry.stderr.strip().split()[-1]
        source.write_bytes(b"changed")
        self.assertEqual(run_cli(str(self.root), "--report", str(destination), "--approve", digest).returncode, 2)
        self.assertFalse(destination.exists())
        destination.symlink_to(source)
        self.assertEqual(run_cli(str(self.root), "--report", str(destination), "--approve", digest).returncode, 2)
        self.assertEqual(source.read_bytes(), b"changed")

    def test_untrusted_filename_stays_on_one_line(self) -> None:
        self.write("a\nINJECTED", b"same")
        self.write("b", b"same")
        process = run_cli(str(self.root))
        self.assertEqual(process.returncode, 1)
        self.assertIn("\\u000aINJECTED", process.stdout)
        self.assertNotIn("\nINJECTED", process.stdout)

    def test_output_parent_swap_refused_before_write(self) -> None:
        parent = self.root / "approved"
        parent.mkdir()
        outside = self.root / "outside"
        outside.mkdir()
        destination = parent / "report.json"
        plan = report.output_plan(str(destination))
        digest = report.approval_digest(str(destination), b"{}", plan)
        parent.rename(self.root / "original")
        parent.symlink_to(outside, target_is_directory=True)
        with self.assertRaises((OSError, report.ScanError)):
            report.create_report(str(destination), b"{}", plan)
        self.assertFalse((outside / "report.json").exists())
        with self.assertRaises((OSError, report.ScanError)):
            report.approval_digest(str(destination), b"{}")
        self.assertEqual(len(digest), 64)

    def test_directory_replacement_changes_digest_and_refuses_old_plan(self) -> None:
        parent = self.root / "approved"
        parent.mkdir()
        destination = parent / "report.json"
        plan = report.output_plan(str(destination))
        before = report.approval_digest(str(destination), b"{}", plan)
        parent.rename(self.root / "original")
        parent.mkdir()
        self.assertNotEqual(before, report.approval_digest(str(destination), b"{}"))
        with self.assertRaises(report.ScanError):
            report.create_report(str(destination), b"{}", plan)
        self.assertFalse(destination.exists())

    def test_missing_output_parent_is_never_created(self) -> None:
        destination = self.root / "missing" / "report.json"
        self.assertEqual(run_cli(str(self.root), "--report", str(destination), "--dry-run").returncode, 2)
        self.assertFalse(destination.parent.exists())

    def test_local_percent_and_backslash_paths_are_literal(self) -> None:
        folder = self.root / "Photos 100%\\literal"
        folder.mkdir()
        (folder / "first").write_bytes(b"same")
        (folder / "second").write_bytes(b"same")
        destination = self.root / "report%41\\literal.json"
        dry = run_cli(str(folder), "--report", str(destination), "--dry-run")
        self.assertEqual(dry.returncode, 1, dry.stderr)
        digest = dry.stderr.strip().split()[-1]
        live = run_cli(str(folder), "--report", str(destination), "--approve", digest)
        self.assertEqual(live.returncode, 1, live.stderr)
        self.assertTrue(destination.is_file())
        self.assertEqual(report.validated_path("%2e%2e/literal"), os.path.abspath("%2e%2e/literal"))

    def test_symlink_component_explains_physical_path(self) -> None:
        real = self.root / "physical"
        real.mkdir()
        alias = self.root / "alias"
        alias.symlink_to(real, target_is_directory=True)
        with self.assertRaisesRegex(report.ScanError, "a path component is a symlink; use the physical path"):
            report.open_directory(str(alias))

    def test_root_symlink_ancestor_is_fatal_with_physical_path_hint(self) -> None:
        real = self.root / "physical"
        real.mkdir()
        (real / "selected").mkdir()
        alias = self.root / "alias"
        alias.symlink_to(real, target_is_directory=True)
        for options in ((), ("--allow-partial",)):
            process = run_cli(str(alias / "selected"), *options)
            self.assertEqual(process.returncode, 2)
            self.assertIn("a path component is a symlink; use the physical path (for example /private/tmp)", process.stderr)
            self.assertEqual(process.stdout, "")

    def test_regular_file_ancestor_is_not_reported_as_symlink(self) -> None:
        file = self.write("regular", b"synthetic")
        process = run_cli(str(file / "selected"))
        self.assertEqual(process.returncode, 2)
        self.assertIn("a path component is not a directory", process.stderr)
        self.assertNotIn("symlink", process.stderr)
        self.assertEqual(process.stdout, "")
        with self.assertRaisesRegex(report.ScanError, "is a symlink or not a directory"):
            with patch.object(report.os, "stat", side_effect=OSError("unavailable")):
                report.open_directory(str(file))

    def test_dry_run_refuses_existing_report_and_dangling_symlink(self) -> None:
        for name, symlink in (("existing.json", False), ("dangling.json", True)):
            destination = self.root / name
            if symlink:
                destination.symlink_to(self.root / "absent")
            else:
                destination.write_bytes(b"keep")
            dry = run_cli(str(self.root), "--report", str(destination), "--dry-run")
            self.assertEqual(dry.returncode, 2, dry.stderr)
            self.assertIn("already exists", dry.stderr)
            self.assertNotIn("Approval digest", dry.stderr)
            if not symlink:
                self.assertEqual(destination.read_bytes(), b"keep")

    def test_traversal_forms_rejected_before_read(self) -> None:
        for path in ("../outside", "safe/../outside", "bad\0name"):
            with self.subTest(path=path), patch.object(report.os, "stat", side_effect=AssertionError("read")):
                with self.assertRaises(report.ScanError):
                    report.scan_paths([path])

    def test_source_parent_swap_cannot_read_outside_root(self) -> None:
        selected = self.root / "selected"
        selected.mkdir()
        (selected / "a").write_bytes(b"same")
        (selected / "b").write_bytes(b"same")
        outside = self.root / "outside"
        outside.mkdir()
        (outside / "a").write_bytes(b"evil")
        (outside / "b").write_bytes(b"evil")
        original_hash = report.file_hash
        swapped = False
        def swap(path, size, partial, expected=None):
            nonlocal swapped
            if not swapped:
                selected.rename(self.root / "saved")
                selected.symlink_to(outside, target_is_directory=True)
                swapped = True
            return original_hash(path, size, partial, expected)
        with patch.object(report, "file_hash", side_effect=swap):
            result = report.scan_paths([str(selected)])
        self.assertTrue(result["incomplete"])
        self.assertEqual(result["groups"], [])

    def test_unsupported_platform_before_io(self) -> None:
        with patch.object(report.os, "name", "nt"), patch.object(report.os, "stat", side_effect=AssertionError("I/O")):
            self.assertEqual(report.main(["scan", "selected"]), 2)

    def test_hash_bounds_reads_and_rejects_growth(self) -> None:
        path = self.write("growing", b"same")
        real_fstat = os.fstat
        calls = 0
        def grow(descriptor):
            nonlocal calls
            value = real_fstat(descriptor)
            if stat.S_ISREG(value.st_mode):
                calls += 1
                if calls == 1:
                    with path.open("ab") as stream:
                        stream.write(b"extra")
            return value
        with patch.object(report.os, "fstat", side_effect=grow):
            with self.assertRaisesRegex(OSError, "grew"):
                report.file_hash(str(path), 4, False)

    def test_help_signature(self) -> None:
        process = subprocess.run([sys.executable, str(SCRIPT), "scan", "--help"], capture_output=True, text=True, check=False)
        self.assertEqual(process.returncode, 0)
        self.assertIn("default: 1; zero-byte files are skipped", " ".join(process.stdout.split()))
        self.assertEqual(process.stdout.rstrip().splitlines()[-1], report.SIGNATURE)

    def test_zero_byte_files_need_explicit_min_size_zero(self) -> None:
        self.write("empty-a", b"")
        self.write("empty-b", b"")
        default = json.loads(run_cli(str(self.root), "--json").stdout)
        included = json.loads(run_cli(str(self.root), "--min-size", "0", "--json").stdout)
        self.assertEqual(default["scanned_files"], 0)
        self.assertEqual(default["groups"], [])
        self.assertEqual(included["scanned_files"], 2)
        self.assertEqual(len(included["groups"]), 1)


if __name__ == "__main__":
    unittest.main()

License: MIT (read it).

Back to the listing