"""A PDF waits to be told whose it is; a photograph does not.

The rule this file covers exists because of what WhatsApp does, not because of a
preference. Forwarding a document offers no caption box, so the message arrives
with the field absent — six real forwards on 31 July 2026 all did — and the
identifier has to come as the next message instead. Reading a PDF's caption
would therefore only ever pick up a caption that came from somewhere else,
which is worse than reading none.

Images are untouched. An image can carry a caption, its contents are what
identify the client, and requiring a second message for every photograph of a
CNIC would make the working half of the workflow worse to serve the broken half.
"""

from __future__ import annotations

import pytest

from app.whatsapp.inbox import InboxFilter
from app.whatsapp.messages import InboundMessage, MediaReference, MessageType
from app.intake import InMemoryOutbox
from app.whatsapp.pending import InMemoryPendingPdfs, PendingPdf
from app.whatsapp.webhook import InboundQueue
from tests.intake_support import RecordingCms

OWNER = "923049637232"


@pytest.fixture
def wired(tmp_path):
    """A container whose engine records what it was asked to start."""
    started: list = []

    class FakeContainer:
        inbound_queue = InboundQueue()
        incoming_dir = tmp_path
        inbox = InboxFilter()
        pending_pdfs = InMemoryPendingPdfs()
        model = None

        # Since ADR-0010 nothing proceeds until the CMS has a record of it, so a
        # container without these processes nothing at all.
        cms = RecordingCms()
        intake_outbox = InMemoryOutbox()

        class whatsapp:
            @staticmethod
            def download_media(media, destination):
                destination.write_bytes(
                    b"\xff\xd8\xff\xe0 jpeg"
                    if destination.suffix.lower() in {".jpg", ".jpeg"}
                    else b"%PDF-1.4 statement"
                )

                return destination

        class engine:
            @staticmethod
            def start(workflow, context):
                started.append((workflow.name, context))

                class Run:
                    id = "r1"

                    class state:
                        value = "awaiting_approval"

                return Run()

    return FakeContainer, started


def pdf(*, message_id="pdf-1", filename="Account Statement - 12-Jul-26.pdf", caption=""):
    return InboundMessage(
        provider_message_id=message_id,
        sender=OWNER,
        type=MessageType.DOCUMENT,
        text=caption,
        media=MediaReference(
            handle="h", mime_type="application/pdf", filename=filename, size_bytes=69100
        ),
    )


def image(*, message_id="img-1", caption="", filename="cnic.jpg"):
    return InboundMessage(
        provider_message_id=message_id,
        sender=OWNER,
        type=MessageType.IMAGE,
        text=caption,
        media=MediaReference(
            handle="h", mime_type="image/jpeg", filename=filename, size_bytes=1024
        ),
    )


def text(body, *, message_id="txt-1"):
    return InboundMessage(
        provider_message_id=message_id, sender=OWNER, type=MessageType.TEXT, text=body
    )


def drain(container):
    from app import __main__ as entry

    entry._process_inbox(container)


class TestAPdfWaits:
    def test_a_pdf_alone_starts_nothing(self, wired):
        container, started = wired
        container.inbound_queue.put(pdf())

        drain(container)

        assert started == []
        assert len(container.pending_pdfs) == 1

    def test_it_is_downloaded_while_it_waits(self, wired, tmp_path):
        """Fetched now, filed later.

        The media handle is the provider's and does not keep. Waiting for the
        identifier before fetching would mean the file expires precisely when
        somebody finally says who it belongs to.
        """
        container, _ = wired
        container.inbound_queue.put(pdf())

        drain(container)

        assert [p.suffix for p in tmp_path.iterdir()] == [".pdf"]

    def test_the_next_message_files_it(self, wired):
        container, started = wired
        container.inbound_queue.put(pdf())
        drain(container)

        container.inbound_queue.put(text("File No: TP-2026-00125"))
        drain(container)

        assert len(started) == 1
        assert started[0][1]["file_number"] == "TP-2026-00125"
        assert len(container.pending_pdfs) == 0

    def test_a_cnic_names_it_just_as_well(self, wired):
        container, started = wired
        container.inbound_queue.put(pdf())
        drain(container)

        container.inbound_queue.put(text("CNIC: 35202-1234567-1"))
        drain(container)

        assert started[0][1]["cnic"] == "3520212345671"

    def test_a_bank_statement_still_reaches_the_workflow_that_never_opens_it(self, wired):
        """The property that matters most, preserved through the change.

        The filename is what routes it, and the routing happens while it is
        being queued rather than when it is named — so the decision not to open
        the file is taken before anything could open it.
        """
        container, started = wired
        container.inbound_queue.put(pdf())
        drain(container)
        container.inbound_queue.put(text("File No: TP-1"))
        drain(container)

        assert started[0][0] == "bank_statement_intake"
        assert started[0][1]["document_type"] == "bank_statement"
        assert started[0][1]["classification"]["was_read"] is False


class TestTheCaptionIsNotRead:
    def test_a_caption_naming_a_client_is_ignored(self, wired):
        """Even a correct one.

        A forwarded PDF has no caption of its own, so any caption present came
        from whatever the document was forwarded from — another client's chat,
        most dangerously. Using it would file this document against them.
        """
        container, started = wired
        container.inbound_queue.put(pdf(caption="File No: TP-9999"))
        drain(container)

        assert started == []

        container.inbound_queue.put(text("File No: TP-1111"))
        drain(container)

        assert started[0][1]["file_number"] == "TP-1111"

    def test_a_caption_naming_a_type_does_not_classify_it(self, wired):
        container, started = wired
        container.inbound_queue.put(pdf(caption="Salary Slip", filename="scan001.pdf"))
        drain(container)
        container.inbound_queue.put(text("File No: TP-1"))
        drain(container)

        assert started[0][1]["classification"]["method"] != "caption"
        assert started[0][1]["document_type"] != "salary_slip"


class TestSeveralPdfsQueue:
    def test_the_oldest_is_named_first(self, wired):
        container, started = wired

        for n in ("A", "B", "C"):
            container.inbound_queue.put(pdf(message_id=f"pdf-{n}"))

        drain(container)

        assert len(container.pending_pdfs) == 3

        container.inbound_queue.put(text("File No: TP-001", message_id="t1"))
        drain(container)

        assert started[0][1]["source"]["message_id"] == "pdf-A"

        container.inbound_queue.put(text("CNIC: 35202-1234567-1", message_id="t2"))
        drain(container)

        assert started[1][1]["source"]["message_id"] == "pdf-B"
        assert started[1][1]["cnic"] == "3520212345671"
        assert len(container.pending_pdfs) == 1

    def test_one_text_names_one_document(self, wired):
        # Not "the text applies to everything waiting" — two statements filed
        # against one client because one message arrived is the kind of error
        # nobody finds until an audit.
        container, started = wired
        container.inbound_queue.put(pdf(message_id="pdf-A"))
        container.inbound_queue.put(pdf(message_id="pdf-B"))
        drain(container)

        container.inbound_queue.put(text("File No: TP-001"))
        drain(container)

        assert len(started) == 1
        assert len(container.pending_pdfs) == 1


class TestNeitherIdentifierIsPresent:
    def test_it_is_filed_for_a_person_rather_than_guessed(self, wired):
        """Never guess. The document still moves — to a human."""
        container, started = wired
        container.inbound_queue.put(pdf())
        drain(container)

        container.inbound_queue.put(text("thanks, that's the one"))
        drain(container)

        assert len(started) == 1
        assert started[0][1]["file_number"] is None
        assert started[0][1]["cnic"] is None
        # Consumed, not left waiting for a better message that is not coming.
        assert len(container.pending_pdfs) == 0

    def test_text_with_nothing_waiting_is_still_just_talk(self, wired):
        container, started = wired
        container.inbound_queue.put(text("File No: TP-001"))

        drain(container)

        assert started == []


class TestAnImageDoesNotWait:
    def test_it_is_processed_at_once(self, wired):
        container, started = wired
        container.inbound_queue.put(image(caption="file 1420"))

        drain(container)

        assert len(started) == 1
        assert len(container.pending_pdfs) == 0

    def test_its_caption_is_still_read(self, wired):
        container, started = wired
        container.inbound_queue.put(image(caption="file 1420"))

        drain(container)

        assert started[0][1]["file_number"] == "1420"

    def test_it_does_not_consume_a_waiting_pdf(self, wired):
        """The two paths must not interfere.

        A photograph arriving between a PDF and its identifier is an ordinary
        thing to happen, and it must not take the PDF's place in the queue.
        """
        container, started = wired
        container.inbound_queue.put(pdf(message_id="pdf-A"))
        drain(container)

        container.inbound_queue.put(image(message_id="img-1", caption="file 99"))
        drain(container)

        assert len(container.pending_pdfs) == 1

        container.inbound_queue.put(text("File No: TP-001"))
        drain(container)

        filed = [c["source"]["message_id"] for _, c in started]

        assert "pdf-A" in filed


class TestAPdfNobodyEverNames:
    def test_it_goes_for_review_once_the_wait_expires(self, wired, monkeypatch):
        from datetime import UTC, datetime, timedelta

        from app import __main__ as entry

        container, started = wired
        container.pending_pdfs.add(
            PendingPdf(
                message_id="old-1",
                path=str(container.incoming_dir / "old.pdf"),
                sender=OWNER,
                received_at=datetime.now(UTC) - timedelta(days=3),
                file_name="Account Statement.pdf",
                mime_type="application/pdf",
                size_bytes=100,
                classification={"workflow": "bank_statement_intake", "filing_type": "bank_statement"},
            )
        )

        drain(container)

        assert len(started) == 1
        assert started[0][1]["file_number"] is None
        assert len(container.pending_pdfs) == 0

    def test_one_still_inside_the_window_keeps_waiting(self, wired):
        from datetime import UTC, datetime, timedelta

        container, started = wired
        container.pending_pdfs.add(
            PendingPdf(
                message_id="recent-1",
                path=str(container.incoming_dir / "recent.pdf"),
                sender=OWNER,
                received_at=datetime.now(UTC) - timedelta(minutes=5),
                classification={"workflow": "document_intake"},
            )
        )

        drain(container)

        assert started == []
        assert len(container.pending_pdfs) == 1

    def test_the_window_is_configurable(self, monkeypatch):
        from app import __main__ as entry

        monkeypatch.setenv("TAXPILOT_PDF_METADATA_WAIT_MINUTES", "30")

        assert entry._pdf_metadata_wait().total_seconds() == 30 * 60

    def test_a_nonsense_window_falls_back_rather_than_crashing(self, monkeypatch):
        from app import __main__ as entry

        monkeypatch.setenv("TAXPILOT_PDF_METADATA_WAIT_MINUTES", "soon")

        assert entry._pdf_metadata_wait().total_seconds() == 24 * 60 * 60


class TestTheQueueItself:
    def test_it_is_first_in_first_out(self):
        from datetime import UTC, datetime

        queue = InMemoryPendingPdfs()
        now = datetime.now(UTC)

        for name in ("a", "b", "c"):
            queue.add(PendingPdf(message_id=name, path=f"/{name}", sender=OWNER, received_at=now))

        assert [queue.take_oldest().message_id for _ in range(3)] == ["a", "b", "c"]

    def test_a_redelivered_message_is_not_queued_twice(self):
        """Providers retry. Two entries would consume two identifiers."""
        from datetime import UTC, datetime

        queue = InMemoryPendingPdfs()
        entry = PendingPdf(message_id="dup", path="/x", sender=OWNER, received_at=datetime.now(UTC))

        queue.add(entry)
        queue.add(entry)

        assert len(queue) == 1

    def test_taking_from_an_empty_queue_is_not_an_error(self):
        assert InMemoryPendingPdfs().take_oldest() is None


class TestTheStrategyDecidesNotTheFormat:
    """The hybrid rule: what happens is a property of the type, not the file.

    A PDF is not "the kind of thing that waits" and an image is not "the kind of
    thing that is read". A CNIC carries its client's identity on its face
    whether it arrives as a photograph or a scan; a bank statement does not,
    whichever way it arrives.
    """

    def test_a_type_that_carries_its_own_identity_is_read_at_once(self, wired):
        from app.documents import registry

        container, started = wired
        container.inbound_queue.put(image(filename="cnic-front.jpg", caption=""))

        drain(container)

        assert registry.get("cnic_front").strategy is registry.ProcessingStrategy.OCR_REQUIRED
        assert len(container.pending_pdfs) == 0, "An OCR_REQUIRED type must never wait."
        assert len(started) == 1

    def test_a_bank_statement_waits_even_as_an_image(self, wired):
        """The point of the change, stated as a test.

        Under the old rule this was a photograph and photographs were read. The
        type says otherwise, and the type wins.
        """
        container, started = wired
        container.inbound_queue.put(image(filename="Bank Statement July.jpg"))

        drain(container)

        assert started == []
        assert len(container.pending_pdfs) == 1

    def test_an_unknown_pdf_reaches_a_person_rather_than_waiting(self, wired):
        """It was held unread, and that cost more than it saved.

        A scanner's filename names nothing, so ordinary paperwork stopped and
        waited for a message nobody knew to send — four such PDFs sat unseen on
        the live installation — while OCR read comparable documents two thousand
        characters at a time. Now it is read and reaches a person either way.
        """
        container, started = wired
        container.inbound_queue.put(pdf(filename="CamScanner 07-06-2026 09.33.pdf"))

        drain(container)

        assert len(container.pending_pdfs) == 0, "an unrecognised PDF must not be held"
        assert len(started) == 1

    def test_reading_it_does_not_mean_taking_it_on_trust(self, wired):
        """The envelope checks apply on this path too.

        The size cap, the type list and the magic bytes lived on the holding
        path alone. Sending unknown PDFs past it without moving them would have
        fetched and opened a file merely claiming to be a PDF, which is how
        "read this document" becomes "run an unknown binary through a parser".
        """
        container, started = wired

        class NotReallyAPdf:
            @staticmethod
            def download_media(media, destination):
                destination.write_bytes(b"MZ\x90\x00 this is not a pdf")

                return destination

        container.whatsapp = NotReallyAPdf
        container.inbound_queue.put(pdf(filename="scan0001.pdf"))

        drain(container)

        assert started == [], "a file that is not what it claims must never be read"
        assert container.inbox.forwards_rejected == 1

    def test_an_unknown_image_is_read_at_once(self, wired):
        # The other half of the same decision. A photograph nearly always shows
        # something that names itself, and making somebody type a file number
        # for every snapshot would be a worse workflow than the one that works.
        container, started = wired
        container.inbound_queue.put(image(filename="IMG_4821.jpg"))

        drain(container)

        assert len(container.pending_pdfs) == 0
        assert len(started) == 1

    def test_a_caption_cannot_turn_a_readable_document_into_a_waiting_one(self, wired):
        """Whether a document may be opened is settled without the caption.

        Otherwise a caption reading "Bank Statement for file 1420" would classify
        the document into the one type whose rule is that captions are ignored,
        and the identifier in that very caption would be discarded and a second
        message demanded for something already answered.
        """
        container, started = wired
        container.inbound_queue.put(image(filename="IMG_4821.jpg", caption="Bank statement for file 1420"))

        drain(container)

        assert len(container.pending_pdfs) == 0
        assert started[0][1]["file_number"] == "1420"


class TestEveryTypeDeclaresHowItIsHandled:
    def test_no_type_is_left_without_a_strategy(self):
        from app.documents import registry

        for filing in registry.filing_types():
            assert isinstance(filing.strategy, registry.ProcessingStrategy), filing.slug

    def test_reading_cannot_disagree_with_the_strategy(self):
        """They were two fields for one release and could contradict each other.

        A type could declare "never read me" beside a strategy that reads, and
        nothing would catch it. One has to be the truth.
        """
        from app.documents import registry

        for filing in registry.filing_types():
            assert filing.reads_document == filing.strategy.reads_document

    def test_the_one_type_that_is_never_opened_says_so_once(self):
        from app.documents import registry

        never_read = [t.slug for t in registry.filing_types() if not t.reads_document]

        assert never_read == ["bank_statement"]

    def test_the_router_carries_the_strategy_to_the_caller(self):
        from app.documents import registry, router
        from app.documents.classifier import Classification

        route = router.route(Classification("bank_statement", confidence=0.95))

        assert route.strategy is registry.ProcessingStrategy.USER_METADATA_REQUIRED
        assert route.reads_document is False

    def test_an_unclassified_document_has_no_strategy_at_all(self):
        """None, not a default.

        Giving "unknown" its own strategy would freeze a guess into the routing
        table. It is a state the caller resolves and re-decides.
        """
        from app.documents import router
        from app.documents.classifier import UNKNOWN, Classification

        assert router.route(Classification(UNKNOWN, confidence=0.0)).strategy is None


class TestReadingFirstAndAskingOnlyIfItFailed:
    """OCR_WITH_METADATA_FALLBACK — the hybrid half.

    The document is read because it usually names its own client. When it does
    not, the answer is not a proposal with an empty client field: there is a
    person on the other end of the conversation who knows, so ask them.
    """

    def test_a_reading_that_named_a_client_proposes_as_before(self):
        from app.workflow.document_intake import nobody_named_it

        ctx = {"may_wait_for_metadata": True, "identify": {"unambiguous": True}}

        assert nobody_named_it(ctx) is False

    def test_a_reading_that_named_nobody_stops_before_proposing(self):
        from app.workflow.document_intake import nobody_named_it

        ctx = {"may_wait_for_metadata": True, "identify": {"unambiguous": False, "matches": []}}

        assert nobody_named_it(ctx) is True

    def test_several_candidates_also_counts_as_naming_nobody(self):
        # Three people is not an answer. Asking is better than making a reviewer
        # choose between candidates the sender could disambiguate in one word.
        from app.workflow.document_intake import nobody_named_it

        ctx = {"may_wait_for_metadata": True,
               "identify": {"unambiguous": False, "matches": [{"id": 1}, {"id": 2}, {"id": 3}]}}

        assert nobody_named_it(ctx) is True

    def test_the_second_pass_proposes_even_when_still_unresolved(self):
        """The rule that stops a document waiting for ever.

        An identifier was supplied and still named nobody. Asking again for
        something already answered would leave the document in the queue
        indefinitely, so it goes to a person instead.
        """
        from app.workflow.document_intake import nobody_named_it

        ctx = {"identify": {"unambiguous": False, "matches": []}}  # flag absent

        assert nobody_named_it(ctx) is False

    def test_a_type_that_must_never_be_read_never_reaches_this_at_all(self):
        # Belt and braces: USER_METADATA_REQUIRED is held before any run starts,
        # so the flag is never set for it.
        from app.workflow.document_intake import nobody_named_it

        assert nobody_named_it({"identify": {"unambiguous": False}}) is False


class TestAStepCanDeclineToRun:
    """`skip_when`, the engine mechanism the fallback rests on."""

    def _engine(self, calls):
        from app.tools.base import Tool, ToolResult
        from app.tools.registry import ToolRegistry
        from app.workflow.engine import WorkflowEngine

        class Noted(Tool):
            name = "noted"
            description = "records that it ran"
            parameters: dict = {}

            def run(self, **kwargs):
                calls.append(kwargs)

                return ToolResult.success({"ran": True})

        registry = ToolRegistry()
        registry.register(Noted())

        return WorkflowEngine(registry)

    def test_a_step_whose_predicate_says_skip_does_not_run(self):
        from app.workflow.engine import Step, Workflow
        from app.workflow.state import StepState

        calls: list = []
        workflow = Workflow(name="w", steps=(
            Step(name="only", tool="noted", skip_when=lambda ctx: True),
        ))

        run = self._engine(calls).start(workflow, {})

        assert calls == []
        assert run.steps[0].state is StepState.SKIPPED
        # Never attempted, so the retry figures do not read as work that failed.
        assert run.steps[0].attempts == 0

    def test_a_step_whose_predicate_says_run_still_runs(self):
        from app.workflow.engine import Step, Workflow
        from app.workflow.state import StepState

        calls: list = []
        workflow = Workflow(name="w", steps=(
            Step(name="only", tool="noted", skip_when=lambda ctx: False),
        ))

        run = self._engine(calls).start(workflow, {})

        # The mechanism under test is the predicate, not the tool: what matters
        # is that the step was not skipped.
        assert run.steps[0].state is not StepState.SKIPPED

    def test_a_predicate_that_raises_does_not_skip(self):
        """A predicate that cannot decide is not grounds to skip.

        Running the step is what happened before this feature existed, and a
        broken predicate silently dropping the filing step would be the worst
        possible failure of it.
        """
        from app.workflow.engine import Step, Workflow

        from app.workflow.state import StepState

        def broken(ctx):
            raise RuntimeError("cannot decide")

        calls: list = []
        workflow = Workflow(name="w", steps=(Step(name="only", tool="noted", skip_when=broken),))

        run = self._engine(calls).start(workflow, {})

        assert run.steps[0].state is not StepState.SKIPPED


class TestATypeDecidesWhatFormatsItAccepts:
    """`allowed_file_types`, actually enforced.

    It sat in the registry for one milestone as configuration nothing read,
    which is worse than not having it: it looks enforced.
    """

    def test_a_type_accepts_the_formats_it_declares(self):
        from app.documents import registry

        assert registry.permits_format("bank_statement", ".pdf") is True
        assert registry.permits_format("bank_statement", "statement.jpg") is True

    def test_a_format_a_type_does_not_declare_is_refused(self):
        from app.documents import registry

        assert registry.permits_format("bank_statement", ".tiff") is False

    def test_an_unregistered_type_decides_nothing(self):
        # Permissive where it does not know. Refusing on ignorance would turn an
        # unrecognised type into a lost document.
        from app.documents import registry

        assert registry.permits_format("not_a_type", ".tiff") is True
        assert registry.permits_format(None, ".tiff") is True

    def test_a_file_with_no_extension_is_not_refused_for_it(self):
        from app.documents import registry

        assert registry.permits_format("bank_statement", "") is True
        assert registry.permits_format("bank_statement", "statement") is True

    def test_a_photographed_bank_statement_is_accepted_and_still_never_read(self):
        """The correction to an over-narrow value I set a milestone ago.

        A phone held over a printout is a normal way for a statement to arrive.
        Refusing it would lose real documents, and it would buy nothing: whether
        the file is opened is decided by the strategy, which does not care what
        format it is in.
        """
        from app.documents import registry

        statement = registry.get("bank_statement")

        assert "jpg" in statement.allowed_file_types
        assert statement.reads_document is False


class TestBusyIsNotStalled:
    """From production, 1 August 2026.

    The agent reported itself unhealthy while reading a client's document —
    which is the work it exists to do. The loop cannot come round while a job is
    inside it, and OCR is tens of seconds a page, so the tick went stale during
    entirely ordinary work. Anything acting on that signal would have restarted
    a working agent mid-read.
    """

    def _daemon(self):
        from app.runtime.daemon import Daemon

        d = Daemon()
        d.started_at = datetime_now()
        d.last_tick_at = datetime_now() - _minutes(5)  # long stale

        return d

    def test_a_long_job_in_flight_is_alive(self):
        daemon = self._daemon()
        daemon.running_job = "inbox"
        daemon.running_since = datetime_now() - _minutes(2)

        assert daemon.is_ticking() is True

    def test_the_same_daemon_with_no_job_running_is_stalled(self):
        # The control: it is only the in-flight job that makes this alive, not
        # a weakened staleness rule.
        assert self._daemon().is_ticking() is False

    def test_a_job_running_far_too_long_is_not_an_excuse_for_ever(self):
        daemon = self._daemon()
        daemon.running_job = "inbox"
        daemon.running_since = datetime_now() - _minutes(30)

        assert daemon.is_ticking() is False

    def test_liveness_says_what_it_is_busy_with(self):
        from app.runtime.health import liveness

        daemon = self._daemon()
        daemon.running_job = "inbox"
        daemon.running_since = datetime_now() - _minutes(1)

        report = liveness(daemon)
        detail = report.to_dict()["components"][0]["detail"]

        assert "inbox" in detail
        assert report.is_ready is True

    def test_a_job_that_raised_does_not_leave_the_loop_looking_busy(self):
        """Cleared in `finally`.

        Otherwise one failing job would suppress the stall signal permanently —
        losing exactly the alarm this whole mechanism exists to preserve.
        """
        from app.runtime.daemon import Daemon

        daemon = Daemon()
        daemon.started_at = datetime_now()
        job = daemon.add("boom", lambda: (_ for _ in ()).throw(RuntimeError("no")), 1)

        daemon._run_job(job)

        assert daemon.running_job is None
        assert daemon.running_since is None


def datetime_now():
    from datetime import UTC, datetime

    return datetime.now(UTC)


def _minutes(n):
    from datetime import timedelta

    return timedelta(minutes=n)


class TestTheQueueIsVisible:
    """A depth that climbs and never falls must be visible as a number.

    Otherwise "identifiers stopped arriving" looks exactly like "a quiet
    evening", and the documents sitting in the queue are invisible until
    somebody asks after one.
    """

    def test_the_depth_is_reported(self, wired):
        from datetime import UTC, datetime

        from app.runtime.server import _pending_pdf_series

        container, _ = wired
        container.pending_pdfs.add(
            PendingPdf(message_id="m1", path="/x.pdf", sender=OWNER, received_at=datetime.now(UTC))
        )

        assert "taxpilot_pending_pdfs 1" in _pending_pdf_series(container)

    def test_a_queue_that_cannot_be_read_does_not_break_the_scrape(self):
        from app.runtime.server import _pending_pdf_series

        class Broken:
            @property
            def pending_pdfs(self):
                raise RuntimeError("database is gone")

        assert _pending_pdf_series(Broken()) == ""


class TestEveryRouteInCountsTheDocument:
    """A forward that arrives must move the same numbers however it got here.

    Three paths take a document in — held for an identifier, forwarded with one,
    and read straight away — and only the first two counted. The third is the
    ordinary path for a photograph, so the operations console reported fewer
    forwards than had arrived, and the gap was invisible: a document read
    successfully looks exactly like nothing went wrong.
    """

    def test_an_image_read_straight_away_is_counted(self, wired):
        container, started = wired
        container.inbound_queue.put(image(caption="here you go"))

        drain(container)

        assert len(started) == 1
        assert container.inbox.forwards_accepted == 1
        assert container.inbox.forwards_failed == 0

    def test_a_download_that_fails_on_that_path_is_counted_too(self, wired):
        container, _ = wired

        class Refusing:
            @staticmethod
            def download_media(media, destination):
                raise RuntimeError("the provider refused the media")

        container.whatsapp = Refusing
        container.inbound_queue.put(image(caption="here you go"))

        drain(container)

        # Failed, not accepted: "it refused everything" and "it could not fetch
        # anything" send an operator to different places.
        assert container.inbox.forwards_failed == 1
        assert container.inbox.forwards_accepted == 0