paperless-ngx/src/documents/tests/test_management_consumer.py

"""
Tests for the document consumer management command.

Tests are organized into classes by component:
- TestFileStabilityTracker: Unit tests for FileStabilityTracker
- TestConsumerFilter: Unit tests for ConsumerFilter
- TestConsumeFile: Unit tests for the _consume_file function
- TestTagsFromPath: Unit tests for _tags_from_path
- TestCommandValidation: Tests for command argument validation
- TestCommandOneshot: Tests for oneshot mode
- TestCommandWatch: Integration tests for the watch loop
"""

from __future__ import annotations

import re
import shutil
from pathlib import Path
from threading import Thread
from time import monotonic
from time import sleep
from typing import TYPE_CHECKING

import pytest
from django import db
from django.core.management import CommandError
from django.db import DatabaseError
from django.test import override_settings
from watchfiles import Change

from documents.data_models import ConsumableDocument
from documents.data_models import DocumentSource
from documents.management.commands.document_consumer import Command
from documents.management.commands.document_consumer import ConsumerFilter
from documents.management.commands.document_consumer import FileStabilityTracker
from documents.management.commands.document_consumer import TrackedFile
from documents.management.commands.document_consumer import _consume_file
from documents.management.commands.document_consumer import _tags_from_path
from documents.models import Tag

if TYPE_CHECKING:
    from collections.abc import Callable
    from collections.abc import Generator
    from unittest.mock import MagicMock

    from pytest_django.fixtures import Settings
    from pytest_mock import MockerFixture


@pytest.fixture
def stability_tracker() -> FileStabilityTracker:
    """Create a FileStabilityTracker with a short delay for testing."""
    return FileStabilityTracker(stability_delay=0.1)


@pytest.fixture
def temp_file(tmp_path: Path) -> Path:
    """Create a temporary file for testing."""
    file_path = tmp_path / "test_file.pdf"
    file_path.write_bytes(b"test content")
    return file_path


@pytest.fixture
def consumption_dir(tmp_path: Path) -> Path:
    """Create a temporary consumption directory for testing."""
    consume_dir = tmp_path / "consume"
    consume_dir.mkdir()
    return consume_dir


@pytest.fixture
def scratch_dir(tmp_path: Path) -> Path:
    """Create a temporary scratch directory for testing."""
    scratch = tmp_path / "scratch"
    scratch.mkdir()
    return scratch


@pytest.fixture
def sample_pdf(tmp_path: Path) -> Path:
    """Create a sample PDF file."""
    pdf_content = b"%PDF-1.4\n%test\n1 0 obj\n<<>>\nendobj\ntrailer\n<<>>\n%%EOF"
    pdf_path = tmp_path / "sample.pdf"
    pdf_path.write_bytes(pdf_content)
    return pdf_path


@pytest.fixture
def consumer_filter() -> ConsumerFilter:
    """Create a ConsumerFilter for testing."""
    return ConsumerFilter(
        supported_extensions=frozenset({".pdf", ".png", ".jpg"}),
        ignore_patterns=[r"^custom_ignore"],
    )


@pytest.fixture
def mock_consume_file_delay(mocker: MockerFixture) -> MagicMock:
    """Mock the consume_file task."""
    return mocker.patch(
        "documents.management.commands.document_consumer.consume_file",
    )


@pytest.fixture
def mock_supported_extensions(mocker: MockerFixture) -> MagicMock:
    """Mock get_supported_file_extensions to return only .pdf."""
    return mocker.patch(
        "documents.management.commands.document_consumer.get_supported_file_extensions",
        return_value={".pdf"},
    )


def wait_for_mock_call(
    mock_obj: MagicMock,
    timeout_s: float = 5.0,
    poll_interval_s: float = 0.1,
) -> bool:
    """
    Actively wait for a mock to be called.

    Args:
        mock_obj: The mock object to check (e.g., mock.delay)
        timeout_s: Maximum time to wait in seconds
        poll_interval_s: How often to check in seconds

    Returns:
        True if mock was called within timeout, False otherwise
    """
    start_time = monotonic()
    while monotonic() - start_time < timeout_s:
        if mock_obj.called:
            return True
        sleep(poll_interval_s)
    return False


def sleep_past_stability(
    owner: FileStabilityTracker | ConsumerThread,
    *,
    windows: float = 1.5,
) -> None:
    """
    Block until a tracked file's stability window has certainly elapsed.

    Args:
        owner: The tracker, or the consumer thread running one, whose
               configured stability delay sets the wait.
        windows: How many stability windows to wait, giving slop for a slow
                 or loaded test runner.
    """
    sleep(owner.stability_delay * windows)


class TestTrackedFile:
    """Tests for the TrackedFile dataclass."""

    def test_update_stats_existing_file(self, temp_file: Path) -> None:
        """Test update_stats succeeds for existing file."""
        tracked = TrackedFile(path=temp_file, last_event_time=monotonic())
        assert tracked.update_stats() is True
        assert tracked.last_mtime is not None
        assert tracked.last_size is not None
        assert tracked.last_size == len(b"test content")

    def test_update_stats_nonexistent_file(self, tmp_path: Path) -> None:
        """Test update_stats fails for nonexistent file."""
        tracked = TrackedFile(
            path=tmp_path / "nonexistent.pdf",
            last_event_time=monotonic(),
        )
        assert tracked.update_stats() is False
        assert tracked.last_mtime is None
        assert tracked.last_size is None

    def test_is_unchanged_same_stats(self, temp_file: Path) -> None:
        """Test is_unchanged returns True when stats haven't changed."""
        tracked = TrackedFile(path=temp_file, last_event_time=monotonic())
        tracked.update_stats()
        assert tracked.is_unchanged() is True

    def test_is_unchanged_modified_file(self, temp_file: Path) -> None:
        """Test is_unchanged returns False when file is modified."""
        tracked = TrackedFile(path=temp_file, last_event_time=monotonic())
        tracked.update_stats()
        temp_file.write_bytes(b"modified content that is longer")
        assert tracked.is_unchanged() is False

    def test_is_unchanged_deleted_file(self, temp_file: Path) -> None:
        """Test is_unchanged returns False when file is deleted."""
        tracked = TrackedFile(path=temp_file, last_event_time=monotonic())
        tracked.update_stats()
        temp_file.unlink()
        assert tracked.is_unchanged() is False


class TestFileStabilityTracker:
    """Tests for the FileStabilityTracker class."""

    def test_track_new_file(
        self,
        stability_tracker: FileStabilityTracker,
        temp_file: Path,
    ) -> None:
        """Test tracking a new file adds it to pending."""
        stability_tracker.track(temp_file, Change.added)
        assert stability_tracker.pending_count == 1
        assert stability_tracker.has_pending_files() is True

    def test_track_modified_file(
        self,
        stability_tracker: FileStabilityTracker,
        temp_file: Path,
    ) -> None:
        """Test tracking a modified file updates its event time."""
        stability_tracker.track(temp_file, Change.added)
        sleep(0.05)
        stability_tracker.track(temp_file, Change.modified)
        assert stability_tracker.pending_count == 1

    def test_track_deleted_file(
        self,
        stability_tracker: FileStabilityTracker,
        temp_file: Path,
    ) -> None:
        """Test tracking a deleted file removes it from pending."""
        stability_tracker.track(temp_file, Change.added)
        assert stability_tracker.pending_count == 1
        stability_tracker.track(temp_file, Change.deleted)
        assert stability_tracker.pending_count == 0
        assert stability_tracker.has_pending_files() is False

    def test_track_nonexistent_file(
        self,
        stability_tracker: FileStabilityTracker,
        tmp_path: Path,
    ) -> None:
        """Test tracking a nonexistent file doesn't add it."""
        nonexistent = tmp_path / "nonexistent.pdf"
        stability_tracker.track(nonexistent, Change.added)
        assert stability_tracker.pending_count == 0

    def test_get_stable_files_before_delay(
        self,
        stability_tracker: FileStabilityTracker,
        temp_file: Path,
    ) -> None:
        """Test get_stable_files returns nothing before delay expires."""
        stability_tracker.track(temp_file, Change.added)
        stable = list(stability_tracker.get_stable_files())
        assert len(stable) == 0
        assert stability_tracker.pending_count == 1

    def test_get_stable_files_after_delay(
        self,
        stability_tracker: FileStabilityTracker,
        temp_file: Path,
    ) -> None:
        """Test get_stable_files returns file after delay expires."""
        stability_tracker.track(temp_file, Change.added)
        sleep(0.15)
        stable = list(stability_tracker.get_stable_files())
        assert len(stable) == 1
        assert stable[0] == temp_file
        assert stability_tracker.pending_count == 0

    def test_get_stable_files_modified_during_check(
        self,
        stability_tracker: FileStabilityTracker,
        temp_file: Path,
    ) -> None:
        """Test file is not returned if modified during stability check."""
        stability_tracker.track(temp_file, Change.added)
        sleep(0.12)
        temp_file.write_bytes(b"modified content")
        stable = list(stability_tracker.get_stable_files())
        assert len(stable) == 0
        assert stability_tracker.pending_count == 1

    def test_get_stable_files_skips_empty_file(
        self,
        stability_tracker: FileStabilityTracker,
        tmp_path: Path,
    ) -> None:
        """
        GIVEN:
            - A zero byte file, tracked and past its stability delay
        WHEN:
            - Stable files are collected
        THEN:
            - The file is not yielded for consumption
            - The file is dropped from tracking rather than held, so an
              abandoned placeholder does not keep the watch loop awake
        """
        empty = tmp_path / "scan.pdf"
        empty.write_bytes(b"")
        stability_tracker.track(empty, Change.added)
        sleep_past_stability(stability_tracker)

        stable = list(stability_tracker.get_stable_files())

        assert stable == []
        assert stability_tracker.pending_count == 0

    def test_empty_file_is_yielded_once_content_arrives(
        self,
        stability_tracker: FileStabilityTracker,
        tmp_path: Path,
    ) -> None:
        """
        GIVEN:
            - A zero byte file which was dropped from tracking while empty
        WHEN:
            - The writer fills the file and a new event re-tracks it
        THEN:
            - The file is yielded for consumption once it is stable
        """
        target = tmp_path / "scan.pdf"
        target.write_bytes(b"")
        stability_tracker.track(target, Change.added)
        sleep_past_stability(stability_tracker)
        assert list(stability_tracker.get_stable_files()) == []

        target.write_bytes(b"%PDF-1.4 content")
        stability_tracker.track(target, Change.modified)
        sleep_past_stability(stability_tracker)

        assert list(stability_tracker.get_stable_files()) == [target]

    def test_get_stable_files_deleted_during_check(self, temp_file: Path) -> None:
        """Test deleted file is not returned during stability check."""
        tracker = FileStabilityTracker(stability_delay=0.1)
        tracker.track(temp_file, Change.added)
        sleep(0.12)
        temp_file.unlink()
        stable = list(tracker.get_stable_files())
        assert len(stable) == 0
        assert tracker.pending_count == 0

    def test_get_stable_files_error_during_check(
        self,
        temp_file: Path,
        mocker: MockerFixture,
    ) -> None:
        """Test a file which has become inaccessible is removed from tracking"""

        mocker.patch.object(Path, "stat", side_effect=PermissionError("denied"))

        tracker = FileStabilityTracker(stability_delay=0.1)
        tracker.track(temp_file, Change.added)
        stable = list(tracker.get_stable_files())
        assert len(stable) == 0
        assert tracker.pending_count == 0

    def test_multiple_files_tracking(
        self,
        stability_tracker: FileStabilityTracker,
        tmp_path: Path,
    ) -> None:
        """Test tracking multiple files independently."""
        file1 = tmp_path / "file1.pdf"
        file2 = tmp_path / "file2.pdf"
        file1.write_bytes(b"content1")
        file2.write_bytes(b"content2")

        stability_tracker.track(file1, Change.added)
        sleep(0.05)
        stability_tracker.track(file2, Change.added)

        assert stability_tracker.pending_count == 2

        sleep(0.06)
        stable = list(stability_tracker.get_stable_files())
        assert len(stable) == 1
        assert stable[0] == file1

        sleep(0.06)
        stable = list(stability_tracker.get_stable_files())
        assert len(stable) == 1
        assert stable[0] == file2

    def test_track_resolves_path(
        self,
        stability_tracker: FileStabilityTracker,
        temp_file: Path,
    ) -> None:
        """Test that tracking resolves paths consistently."""
        stability_tracker.track(temp_file, Change.added)
        stability_tracker.track(temp_file.resolve(), Change.modified)
        assert stability_tracker.pending_count == 1


class TestConsumerFilter:
    """Tests for the ConsumerFilter class."""

    @pytest.mark.parametrize(
        ("filename", "should_accept"),
        [
            pytest.param("document.pdf", True, id="supported_pdf"),
            pytest.param("image.png", True, id="supported_png"),
            pytest.param("photo.jpg", True, id="supported_jpg"),
            pytest.param("document.PDF", True, id="case_insensitive"),
            pytest.param("document.xyz", False, id="unsupported_ext"),
            pytest.param("document", False, id="no_extension"),
            pytest.param(".DS_Store", False, id="ds_store"),
            pytest.param(".DS_STORE", False, id="ds_store_upper"),
            pytest.param("._document.pdf", False, id="macos_resource_fork"),
            pytest.param("._hidden", False, id="macos_resource_no_ext"),
            pytest.param("Thumbs.db", False, id="thumbs_db"),
            pytest.param("desktop.ini", False, id="desktop_ini"),
            pytest.param("custom_ignore_this.pdf", False, id="custom_pattern"),
            pytest.param("stfolder.pdf", True, id="similar_to_ignored"),
            pytest.param("my_document.pdf", True, id="normal_with_underscore"),
        ],
    )
    def test_file_filtering(
        self,
        consumer_filter: ConsumerFilter,
        tmp_path: Path,
        filename: str,
        should_accept: bool,  # noqa: FBT001
    ) -> None:
        """Test filter correctly accepts or rejects files."""
        test_file = tmp_path / filename
        test_file.touch()
        assert consumer_filter(Change.added, str(test_file)) is should_accept

    @pytest.mark.parametrize(
        ("dirname", "should_accept"),
        [
            pytest.param(".stfolder", False, id="syncthing_stfolder"),
            pytest.param(".stversions", False, id="syncthing_stversions"),
            pytest.param("@eaDir", False, id="synology_eadir"),
            pytest.param(".Spotlight-V100", False, id="macos_spotlight"),
            pytest.param(".Trashes", False, id="macos_trashes"),
            pytest.param("__MACOSX", False, id="macos_archive"),
            pytest.param(".localized", False, id="macos_localized"),
            pytest.param("documents", True, id="normal_dir"),
            pytest.param("invoices", True, id="normal_dir_2"),
        ],
    )
    def test_directory_filtering(
        self,
        consumer_filter: ConsumerFilter,
        tmp_path: Path,
        dirname: str,
        should_accept: bool,  # noqa: FBT001
    ) -> None:
        """Test filter correctly accepts or rejects directories."""
        test_dir = tmp_path / dirname
        test_dir.mkdir()
        assert consumer_filter(Change.added, str(test_dir)) is should_accept

    def test_default_patterns_are_valid_regex(self) -> None:
        """Test that default patterns are valid regex."""
        for pattern in ConsumerFilter.DEFAULT_IGNORE_PATTERNS:
            re.compile(pattern)

    def test_custom_ignore_dirs(self, tmp_path: Path) -> None:
        """Test filter respects custom ignore_dirs."""
        filter_obj = ConsumerFilter(
            supported_extensions=frozenset({".pdf"}),
            ignore_dirs=["custom_ignored_dir"],
        )

        # Custom ignored directory should be rejected
        custom_dir = tmp_path / "custom_ignored_dir"
        custom_dir.mkdir()
        assert filter_obj(Change.added, str(custom_dir)) is False

        # Normal directory should be accepted
        normal_dir = tmp_path / "normal_dir"
        normal_dir.mkdir()
        assert filter_obj(Change.added, str(normal_dir)) is True

        # Default ignored directories should still be ignored
        stfolder = tmp_path / ".stfolder"
        stfolder.mkdir()
        assert filter_obj(Change.added, str(stfolder)) is False


class TestConsumerFilterDefaults:
    """Tests for ConsumerFilter with default settings."""

    def test_filter_with_mocked_extensions(
        self,
        tmp_path: Path,
        mocker: MockerFixture,
    ) -> None:
        """Test filter works when using mocked extensions from parser."""
        mocker.patch(
            "documents.management.commands.document_consumer.get_supported_file_extensions",
            return_value={".pdf", ".png"},
        )
        filter_obj = ConsumerFilter()
        test_file = tmp_path / "document.pdf"
        test_file.touch()
        assert filter_obj(Change.added, str(test_file)) is True


class TestConsumeFile:
    """Tests for the _consume_file function."""

    def test_consume_queues_file(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
    ) -> None:
        """Test _consume_file queues a valid file."""
        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)

        result = _consume_file(
            filepath=target,
            consumption_dir=consumption_dir,
            subdirs_as_tags=False,
        )

        assert result is True
        mock_consume_file_delay.apply_async.assert_called_once()
        call_args = mock_consume_file_delay.apply_async.call_args
        consumable_doc = call_args.kwargs["kwargs"]["input_doc"]
        assert isinstance(consumable_doc, ConsumableDocument)
        assert consumable_doc.original_file == target
        assert consumable_doc.source == DocumentSource.ConsumeFolder

    def test_consume_nonexistent_file(
        self,
        consumption_dir: Path,
        mock_consume_file_delay: MagicMock,
    ) -> None:
        """Test _consume_file handles nonexistent files gracefully."""
        result = _consume_file(
            filepath=consumption_dir / "nonexistent.pdf",
            consumption_dir=consumption_dir,
            subdirs_as_tags=False,
        )
        assert result is False
        mock_consume_file_delay.apply_async.assert_not_called()

    def test_consume_directory(
        self,
        consumption_dir: Path,
        mock_consume_file_delay: MagicMock,
    ) -> None:
        """Test _consume_file ignores directories."""
        subdir = consumption_dir / "subdir"
        subdir.mkdir()

        result = _consume_file(
            filepath=subdir,
            consumption_dir=consumption_dir,
            subdirs_as_tags=False,
        )
        assert result is False
        mock_consume_file_delay.apply_async.assert_not_called()

    def test_consume_with_permission_error(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        mocker: MockerFixture,
    ) -> None:
        """Test _consume_file handles permission errors."""
        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)

        mocker.patch.object(Path, "is_file", side_effect=PermissionError("denied"))
        result = _consume_file(
            filepath=target,
            consumption_dir=consumption_dir,
            subdirs_as_tags=False,
        )
        assert result is False
        mock_consume_file_delay.apply_async.assert_not_called()

    def test_consume_with_apply_async_failure(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
    ) -> None:
        """Test _consume_file reports failure when apply_async raises."""
        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)

        mock_consume_file_delay.apply_async.side_effect = Exception("broker down")

        result = _consume_file(
            filepath=target,
            consumption_dir=consumption_dir,
            subdirs_as_tags=False,
        )
        assert result is False

    def test_consume_with_tags_error(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        mocker: MockerFixture,
    ) -> None:
        """Test _consume_file handles errors during tag creation"""
        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)

        mocker.patch(
            "documents.management.commands.document_consumer._tags_from_path",
            side_effect=DatabaseError("Something happened"),
        )

        result = _consume_file(
            filepath=target,
            consumption_dir=consumption_dir,
            subdirs_as_tags=True,
        )
        assert result is True
        mock_consume_file_delay.apply_async.assert_called_once()
        call_args = mock_consume_file_delay.apply_async.call_args
        overrides = call_args.kwargs["kwargs"]["overrides"]
        assert overrides.tag_ids is None


@pytest.mark.django_db
class TestTagsFromPath:
    """Tests for the _tags_from_path function."""

    def test_creates_tags_from_subdirectories(self, consumption_dir: Path) -> None:
        """Test tags are created for each subdirectory."""
        subdir = consumption_dir / "Invoice" / "2024"
        subdir.mkdir(parents=True)
        target = subdir / "document.pdf"
        target.touch()

        tag_ids = _tags_from_path(target, consumption_dir)

        assert len(tag_ids) == 2
        assert Tag.objects.filter(name="Invoice").exists()
        assert Tag.objects.filter(name="2024").exists()

    def test_reuses_existing_tags(self, consumption_dir: Path) -> None:
        """Test existing tags are reused (case-insensitive)."""
        existing_tag = Tag.objects.create(name="existing")

        subdir = consumption_dir / "EXISTING"
        subdir.mkdir(parents=True)
        target = subdir / "document.pdf"
        target.touch()

        tag_ids = _tags_from_path(target, consumption_dir)

        assert len(tag_ids) == 1
        assert existing_tag.pk in tag_ids
        assert Tag.objects.filter(name__iexact="existing").count() == 1

    def test_no_tags_for_root_file(self, consumption_dir: Path) -> None:
        """Test no tags created for files directly in consumption dir."""
        target = consumption_dir / "document.pdf"
        target.touch()

        tag_ids = _tags_from_path(target, consumption_dir)

        assert len(tag_ids) == 0


@pytest.mark.management
class TestCommandValidation:
    """Tests for command argument validation."""

    def test_raises_for_missing_consumption_dir(
        self,
        settings: Settings,
    ) -> None:
        """Test command raises error when directory is not provided."""
        settings.CONSUMPTION_DIR = None
        with pytest.raises(CommandError, match="not configured"):
            cmd = Command()
            cmd.handle(directory=None, oneshot=True, testing=False)

    def test_raises_for_nonexistent_directory(self, tmp_path: Path) -> None:
        """Test command raises error for nonexistent directory."""
        nonexistent = tmp_path / "nonexistent"

        with pytest.raises(CommandError, match="does not exist"):
            cmd = Command()
            cmd.handle(directory=str(nonexistent), oneshot=True, testing=False)

    def test_raises_for_file_instead_of_directory(self, sample_pdf: Path) -> None:
        """Test command raises error when path is a file, not directory."""
        with pytest.raises(CommandError, match="not a directory"):
            cmd = Command()
            cmd.handle(directory=str(sample_pdf), oneshot=True, testing=False)


@pytest.mark.management
@pytest.mark.usefixtures("mock_supported_extensions")
class TestCommandOneshot:
    """Tests for oneshot mode."""

    def test_processes_existing_files(
        self,
        consumption_dir: Path,
        scratch_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        settings: Settings,
    ) -> None:
        """Test oneshot mode processes existing files."""
        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)

        settings.SCRATCH_DIR = scratch_dir
        settings.CONSUMER_IGNORE_PATTERNS = []

        cmd = Command()
        cmd.handle(directory=str(consumption_dir), oneshot=True, testing=False)

        mock_consume_file_delay.apply_async.assert_called_once()

    def test_processes_recursive(
        self,
        consumption_dir: Path,
        scratch_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        settings: Settings,
    ) -> None:
        """Test oneshot mode processes files recursively."""
        subdir = consumption_dir / "subdir"
        subdir.mkdir()
        target = subdir / "document.pdf"
        shutil.copy(sample_pdf, target)

        settings.SCRATCH_DIR = scratch_dir
        settings.CONSUMER_RECURSIVE = True
        settings.CONSUMER_IGNORE_PATTERNS = []

        cmd = Command()
        cmd.handle(directory=str(consumption_dir), oneshot=True, testing=False)

        mock_consume_file_delay.apply_async.assert_called_once()

    def test_ignores_unsupported_extensions(
        self,
        consumption_dir: Path,
        scratch_dir: Path,
        mock_consume_file_delay: MagicMock,
        settings: Settings,
    ) -> None:
        """Test oneshot mode ignores unsupported file extensions."""
        target = consumption_dir / "document.xyz"
        target.write_bytes(b"content")

        settings.SCRATCH_DIR = scratch_dir
        settings.CONSUMER_IGNORE_PATTERNS = []

        cmd = Command()
        cmd.handle(directory=str(consumption_dir), oneshot=True, testing=False)

        mock_consume_file_delay.apply_async.assert_not_called()


class ConsumerThread(Thread):
    """Thread wrapper for running the consumer command with proper cleanup."""

    def __init__(
        self,
        consumption_dir: Path,
        scratch_dir: Path,
        *,
        recursive: bool = False,
        subdirs_as_tags: bool = False,
        polling_interval: float = 0,
        stability_delay: float = 0.1,
        rescan_interval: float | None = None,
    ) -> None:
        super().__init__()
        self.consumption_dir = consumption_dir
        self.scratch_dir = scratch_dir
        self.recursive = recursive
        self.subdirs_as_tags = subdirs_as_tags
        self.polling_interval = polling_interval
        self.stability_delay = stability_delay
        self.cmd = Command()
        if rescan_interval is not None:
            self.cmd.rescan_interval_s = rescan_interval
        self.cmd.stop_flag.clear()
        # Non-daemon ensures finally block runs and connections are closed
        self.daemon = False
        self.exception: Exception | None = None

    def run(self) -> None:
        try:
            # Use override_settings to avoid polluting global settings
            # which would affect other tests running on the same worker
            with override_settings(
                SCRATCH_DIR=self.scratch_dir,
                CONSUMER_RECURSIVE=self.recursive,
                CONSUMER_SUBDIRS_AS_TAGS=self.subdirs_as_tags,
                CONSUMER_POLLING_INTERVAL=self.polling_interval,
                CONSUMER_STABILITY_DELAY=self.stability_delay,
                CONSUMER_IGNORE_PATTERNS=[],
            ):
                self.cmd.handle(
                    directory=str(self.consumption_dir),
                    oneshot=False,
                    testing=True,
                )
        except Exception as e:
            self.exception = e
        finally:
            # Close database connections created in this thread
            db.connections.close_all()

    def stop(self) -> None:
        self.cmd.stop_flag.set()

    def stop_and_wait(self, timeout: float = 5.0) -> None:
        """Stop the thread and wait for it to finish, with cleanup."""
        self.stop()
        self.join(timeout=timeout)
        if self.is_alive():
            # Thread didn't stop in time - this is a test failure
            raise RuntimeError(
                f"Consumer thread did not stop within {timeout}s timeout",
            )


@pytest.fixture
def start_consumer(
    consumption_dir: Path,
    scratch_dir: Path,
    mock_supported_extensions: MagicMock,
) -> Generator[Callable[..., ConsumerThread], None, None]:
    """Start a consumer thread and ensure cleanup."""
    threads: list[ConsumerThread] = []

    def _start(**kwargs) -> ConsumerThread:
        thread = ConsumerThread(consumption_dir, scratch_dir, **kwargs)
        threads.append(thread)
        thread.start()
        sleep(2.0)  # Give thread time to start
        return thread

    try:
        yield _start
    finally:
        # Cleanup all threads that were started
        for thread in threads:
            thread.stop_and_wait()

        failed_threads = []
        for thread in threads:
            thread.join(timeout=5.0)
            if thread.is_alive():
                failed_threads.append(thread)

        # Clean up any Tags created by threads (they bypass test transaction isolation)
        Tag.objects.all().delete()

        db.connections.close_all()

        if failed_threads:
            pytest.fail(
                f"{len(failed_threads)} consumer thread(s) did not stop within timeout",
            )


@pytest.mark.management
@pytest.mark.django_db
class TestCommandWatch:
    """Integration tests for the watch loop."""

    def test_detects_new_file(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """Test watch mode detects and consumes new files."""
        thread = start_consumer()

        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)

        wait_for_mock_call(mock_consume_file_delay.apply_async, timeout_s=2.0)

        if thread.exception:
            raise thread.exception

        mock_consume_file_delay.apply_async.assert_called()

    def test_detects_moved_file(
        self,
        consumption_dir: Path,
        scratch_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """Test watch mode detects moved/renamed files."""
        temp_location = scratch_dir / "temp.pdf"
        shutil.copy(sample_pdf, temp_location)

        thread = start_consumer()

        sleep(0.5)

        target = consumption_dir / "document.pdf"
        shutil.move(temp_location, target)

        wait_for_mock_call(mock_consume_file_delay.apply_async, timeout_s=2.0)

        if thread.exception:
            raise thread.exception

        mock_consume_file_delay.apply_async.assert_called()

    def test_handles_slow_write(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """Test watch mode waits for slow writes to complete."""
        pdf_bytes = sample_pdf.read_bytes()

        thread = start_consumer(stability_delay=0.2)

        target = consumption_dir / "document.pdf"
        with target.open("wb") as f:
            for i in range(0, len(pdf_bytes), 100):
                f.write(pdf_bytes[i : i + 100])
                f.flush()
                sleep(0.05)

        wait_for_mock_call(mock_consume_file_delay.apply_async, timeout_s=2.0)

        if thread.exception:
            raise thread.exception

        mock_consume_file_delay.apply_async.assert_called()

    def test_scanner_placeholder_is_not_consumed_while_empty(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """
        GIVEN:
            - A scanner which creates a zero byte placeholder and only writes
              the page some time later (GH discussion #13969)
        WHEN:
            - The placeholder sits untouched well past the stability delay
            - The scanner then writes the real content
        THEN:
            - The empty placeholder is never queued, as it could only fail
              with "Unsupported mime type inode/x-empty"
            - The file is queued exactly once, when the content lands
        """
        thread = start_consumer(stability_delay=0.2)

        target = consumption_dir / "scan.pdf"
        target.write_bytes(b"")  # the scanner's placeholder

        # Well past the stability delay: the old behaviour queued it here.
        sleep_past_stability(thread, windows=5)
        if thread.exception:
            raise thread.exception
        assert mock_consume_file_delay.apply_async.call_count == 0

        shutil.copy(sample_pdf, target)  # the scanner finishes the page

        assert wait_for_mock_call(
            mock_consume_file_delay.apply_async,
            timeout_s=5.0,
        )
        if thread.exception:
            raise thread.exception

        assert mock_consume_file_delay.apply_async.call_count == 1
        queued_doc = mock_consume_file_delay.apply_async.call_args.kwargs["kwargs"][
            "input_doc"
        ]
        assert queued_doc.original_file.name == "scan.pdf"

    def test_ignores_macos_files(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """Test watch mode ignores macOS system files."""
        thread = start_consumer()

        (consumption_dir / ".DS_Store").write_bytes(b"test")
        (consumption_dir / "._document.pdf").write_bytes(b"test")
        shutil.copy(sample_pdf, consumption_dir / "valid.pdf")

        wait_for_mock_call(mock_consume_file_delay.apply_async, timeout_s=2.0)

        if thread.exception:
            raise thread.exception

        assert mock_consume_file_delay.apply_async.call_count == 1
        call_args = mock_consume_file_delay.apply_async.call_args.kwargs["kwargs"][
            "input_doc"
        ]
        assert call_args.original_file.name == "valid.pdf"

    def test_ignores_event_for_already_queued_file(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """
        A stray filesystem event (NAS metadata touch, AV scan, etc.) on a
        file that has already been queued and is awaiting consumption must
        not cause it to be queued a second time (GH #13511). The task is
        mocked, so the file is never removed from disk, mirroring a
        long-running OCR job still holding it.
        """
        thread = start_consumer()

        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)

        wait_for_mock_call(mock_consume_file_delay.apply_async, timeout_s=2.0)

        if thread.exception:
            raise thread.exception

        assert mock_consume_file_delay.apply_async.call_count == 1

        # Simulate a stray event on the still-present, already-queued file.
        target.touch()
        sleep(0.5)

        if thread.exception:
            raise thread.exception

        assert mock_consume_file_delay.apply_async.call_count == 1

    @pytest.mark.django_db
    @pytest.mark.usefixtures("mock_supported_extensions")
    def test_stop_flag_stops_consumer(
        self,
        consumption_dir: Path,
        scratch_dir: Path,
        mock_consume_file_delay: MagicMock,
    ) -> None:
        """Test stop flag properly stops the consumer."""
        thread = ConsumerThread(consumption_dir, scratch_dir)
        try:
            thread.start()
            sleep(0.3)
            assert thread.is_alive()
        finally:
            thread.stop_and_wait(timeout=5.0)
            # Clean up any Tags created by the thread
            Tag.objects.all().delete()

        assert not thread.is_alive()


@pytest.mark.management
@pytest.mark.django_db
class TestCommandWatchPolling:
    """Tests for polling mode."""

    def test_polling_mode_works(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """
        Test polling mode detects files.

        Uses active waiting with timeout to handle CI delays and polling timing.
        """
        # Use shorter polling interval for faster test
        thread = start_consumer(polling_interval=0.5, stability_delay=0.1)

        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)

        # Actively wait for consumption
        # Polling needs: interval (0.5s) + stability (0.1s) + next poll (0.5s) + margin
        wait_for_mock_call(mock_consume_file_delay.apply_async, timeout_s=5.0)

        if thread.exception:
            raise thread.exception

        mock_consume_file_delay.apply_async.assert_called()


@pytest.mark.management
@pytest.mark.django_db
class TestCommandWatchRecursive:
    """Tests for recursive watching."""

    def test_recursive_detects_nested_files(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """Test recursive mode detects files in subdirectories."""
        subdir = consumption_dir / "level1" / "level2"
        subdir.mkdir(parents=True)

        thread = start_consumer(recursive=True)

        target = subdir / "document.pdf"
        shutil.copy(sample_pdf, target)

        wait_for_mock_call(mock_consume_file_delay.apply_async, timeout_s=2.0)

        if thread.exception:
            raise thread.exception

        mock_consume_file_delay.apply_async.assert_called()

    def test_subdirs_as_tags(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
        mocker: MockerFixture,
    ) -> None:
        """Test subdirs_as_tags creates tags from directory names."""
        # Mock _tags_from_path to avoid database operations in the consumer thread
        mock_tags = mocker.patch(
            "documents.management.commands.document_consumer._tags_from_path",
            return_value=[1, 2],
        )

        subdir = consumption_dir / "Invoices" / "2024"
        subdir.mkdir(parents=True)

        thread = start_consumer(recursive=True, subdirs_as_tags=True)

        target = subdir / "document.pdf"
        shutil.copy(sample_pdf, target)

        wait_for_mock_call(mock_consume_file_delay.apply_async, timeout_s=2.0)

        if thread.exception:
            raise thread.exception

        mock_consume_file_delay.apply_async.assert_called()
        mock_tags.assert_called()
        call_args = mock_consume_file_delay.apply_async.call_args
        overrides = call_args.kwargs["kwargs"]["overrides"]
        assert overrides.tag_ids is not None
        assert len(overrides.tag_ids) == 2


@pytest.mark.management
@pytest.mark.django_db
class TestCommandWatchEdgeCases:
    """Tests for edge cases and error handling."""

    def test_handles_deleted_before_stable(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """Test handles files deleted before becoming stable."""
        thread = start_consumer(stability_delay=0.3)

        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)
        sleep(0.1)
        target.unlink()

        sleep(0.5)

        if thread.exception:
            raise thread.exception

        mock_consume_file_delay.apply_async.assert_not_called()

    @pytest.mark.usefixtures("mock_supported_extensions")
    def test_handles_task_exception(
        self,
        consumption_dir: Path,
        scratch_dir: Path,
        sample_pdf: Path,
        mocker: MockerFixture,
    ) -> None:
        """Test handles exceptions from consume task gracefully."""
        mock_task = mocker.patch(
            "documents.management.commands.document_consumer.consume_file",
        )
        mock_task.apply_async.side_effect = Exception("Task error")

        thread = ConsumerThread(consumption_dir, scratch_dir)
        try:
            thread.start()
            sleep(0.3)

            target = consumption_dir / "document.pdf"
            shutil.copy(sample_pdf, target)
            sleep(0.5)

            # Consumer should still be running despite the exception
            assert thread.is_alive()
        finally:
            thread.stop_and_wait(timeout=5.0)
            # Clean up any Tags created by the thread
            Tag.objects.all().delete()


class TestRescanExistingFiles:
    """
    Unit tests for the rescan safety net.

    Each ``watch()`` recreation silently adopts the current directory contents
    as its baseline, so a file appearing between one batch and the next
    watcher's baseline is never reported and would sit in the consume directory
    forever. ``_rescan_existing_files`` re-injects such files into the
    stability tracker as a periodic safety net (see GH issue #13011).
    """

    @pytest.fixture
    def pdf_only_filter(self) -> ConsumerFilter:
        return ConsumerFilter(
            supported_extensions=frozenset({".pdf"}),
            ignore_patterns=[],
        )

    def _rescan(
        self,
        directory: Path,
        consumer_filter: ConsumerFilter,
        tracker: FileStabilityTracker,
        queued: set[Path],
        *,
        recursive: bool = False,
    ) -> None:
        Command()._rescan_existing_files(
            directory=directory,
            recursive=recursive,
            consumer_filter=consumer_filter,
            tracker=tracker,
            queued=queued,
        )

    def test_tracks_stranded_file(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        pdf_only_filter: ConsumerFilter,
    ) -> None:
        """A supported on-disk file the watcher never reported gets tracked."""
        target = consumption_dir / "stranded.pdf"
        shutil.copy(sample_pdf, target)
        tracker = FileStabilityTracker(stability_delay=0.1)

        self._rescan(consumption_dir, pdf_only_filter, tracker, set())

        assert tracker.is_tracking(target) is True
        assert tracker.pending_count == 1

    def test_skips_already_tracked_file(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        pdf_only_filter: ConsumerFilter,
    ) -> None:
        """A file already being tracked by the watcher is not double-tracked."""
        target = consumption_dir / "tracked.pdf"
        shutil.copy(sample_pdf, target)
        tracker = FileStabilityTracker(stability_delay=0.1)
        tracker.track(target, Change.added)

        self._rescan(consumption_dir, pdf_only_filter, tracker, set())

        assert tracker.pending_count == 1

    def test_skips_queued_file(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        pdf_only_filter: ConsumerFilter,
    ) -> None:
        """A file already queued and awaiting consumption is not re-tracked."""
        target = consumption_dir / "inflight.pdf"
        shutil.copy(sample_pdf, target)
        tracker = FileStabilityTracker(stability_delay=0.1)
        queued = {target.resolve()}

        self._rescan(consumption_dir, pdf_only_filter, tracker, queued)

        assert tracker.pending_count == 0

    def test_prunes_vanished_queued_paths(
        self,
        consumption_dir: Path,
        pdf_only_filter: ConsumerFilter,
    ) -> None:
        """Queued paths no longer on disk are dropped so the name can recur."""
        gone = (consumption_dir / "gone.pdf").resolve()
        tracker = FileStabilityTracker(stability_delay=0.1)
        queued = {gone}

        self._rescan(consumption_dir, pdf_only_filter, tracker, queued)

        assert gone not in queued

    def test_skips_unsupported_extension(
        self,
        consumption_dir: Path,
        pdf_only_filter: ConsumerFilter,
    ) -> None:
        """Files filtered out by the consumer filter are not tracked."""
        (consumption_dir / "notes.xyz").write_bytes(b"content")
        tracker = FileStabilityTracker(stability_delay=0.1)

        self._rescan(consumption_dir, pdf_only_filter, tracker, set())

        assert tracker.pending_count == 0

    def test_recursive_respects_flag(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        pdf_only_filter: ConsumerFilter,
    ) -> None:
        """Nested files are only found when recursive scanning is enabled."""
        subdir = consumption_dir / "nested"
        subdir.mkdir()
        target = subdir / "deep.pdf"
        shutil.copy(sample_pdf, target)

        shallow = FileStabilityTracker(stability_delay=0.1)
        self._rescan(consumption_dir, pdf_only_filter, shallow, set())
        assert shallow.pending_count == 0

        deep = FileStabilityTracker(stability_delay=0.1)
        self._rescan(consumption_dir, pdf_only_filter, deep, set(), recursive=True)
        assert deep.is_tracking(target) is True


class TestProcessExistingFilesQueued:
    """Tests that startup processing reports which paths it queued."""

    @pytest.mark.usefixtures("mock_supported_extensions")
    def test_returns_queued_paths(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        settings: Settings,
    ) -> None:
        """The set returned seeds the rescan's queued set, avoiding re-queue."""
        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)
        settings.CONSUMER_IGNORE_PATTERNS = []

        queued = Command()._process_existing_files(
            directory=consumption_dir,
            recursive=False,
            subdirs_as_tags=False,
            consumer_filter=ConsumerFilter(ignore_patterns=[]),
        )

        assert target.resolve() in queued


@pytest.mark.management
@pytest.mark.django_db
class TestCommandRetryAfterQueueFailure:
    """
    Regression test for GH #13923.

    A file whose ``apply_async`` publish fails (e.g. broker briefly down)
    must not be marked as queued, so the periodic rescan retries it once
    the broker recovers, instead of stranding it until the consumer
    process is restarted.
    """

    def test_watch_loop_retries_failed_publish_on_rescan(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """A publish failure from the watch loop is retried by the rescan."""
        apply_async = mock_consume_file_delay.apply_async

        def fail_first_call(*args: object, **kwargs: object) -> None:
            if apply_async.call_count == 1:
                raise Exception("broker down")

        apply_async.side_effect = fail_first_call

        thread = start_consumer(stability_delay=0.1, rescan_interval=0.3)

        target = consumption_dir / "document.pdf"
        shutil.copy(sample_pdf, target)

        deadline = monotonic() + 5.0
        while apply_async.call_count < 2 and monotonic() < deadline:
            sleep(0.1)

        if thread.exception:
            raise thread.exception

        assert apply_async.call_count >= 2, (
            "Expected the failed publish to be retried by the rescan, "
            f"but apply_async was only called {apply_async.call_count} time(s)"
        )


@pytest.mark.management
@pytest.mark.django_db
class TestCommandRescanRecovery:
    """End-to-end test that the rescan recovers files the watcher misses."""

    def test_rescan_consumes_file_the_watcher_never_reports(
        self,
        consumption_dir: Path,
        sample_pdf: Path,
        mock_consume_file_delay: MagicMock,
        start_consumer: Callable[..., ConsumerThread],
    ) -> None:
        """
        Isolate the rescan path: a long polling interval guarantees the
        watcher cannot report the file within the test window, so only the
        periodic rescan can consume it.
        """
        # poll interval far longer than the test window -> watcher stays silent
        thread = start_consumer(
            polling_interval=30.0,
            stability_delay=0.1,
            rescan_interval=0.5,
        )

        # created after startup, so _process_existing_files did not see it
        target = consumption_dir / "stranded.pdf"
        shutil.copy(sample_pdf, target)

        wait_for_mock_call(mock_consume_file_delay.apply_async, timeout_s=5.0)

        if thread.exception:
            raise thread.exception

        mock_consume_file_delay.apply_async.assert_called()
        call_args = mock_consume_file_delay.apply_async.call_args.kwargs["kwargs"][
            "input_doc"
        ]
        assert call_args.original_file.name == "stranded.pdf"