test_decide.py

"""Gap detection and the accept / install / cloud decision (tec_lite.md §6.3)."""

from __future__ import annotations

import os
import tempfile
import unittest
import zipfile
from pathlib import Path
from unittest.mock import patch

from _helpers import TINY_PNG, write_test_pptx, write_test_xlsx

from deckflow_extract import decide, probes, registry
from deckflow_extract.adapters import pptx_python as pptx_adapter

PDF_ENGINES = {engine.name: engine for engine in registry.engines_for("pdf")}


# Roughly 200 characters per page, i.e. comfortably above the "scanned" floor.
BODY = "这是一段足够长的正文内容，用来模拟真实文档的文字密度。" * 8


def assess(**overrides):
    defaults = {
        "format_id": "pdf",
        "engine": PDF_ENGINES["pypdf"],
        "markdown": BODY,
        "element_stats": {"text": 5, "image": 0, "table": 0, "code": 0, "list": 0},
        "asset_count": 0,
        "detected": {},
        "origin": "/tmp/report.pdf",
        "out": "/tmp/report.parse",
    }
    defaults.update(overrides)
    return decide.assess(**defaults)


class SuspectedTablesTest(unittest.TestCase):
    def test_whitespace_aligned_rows_are_flagged_as_heuristic(self) -> None:
        markdown = "季度    收入    成本\nQ1      100     60\nQ2      120     70\n"
        count, locators = decide.suspected_tables(markdown)
        self.assertGreaterEqual(count, 1)
        self.assertTrue(locators)

    def test_real_gfm_tables_are_not_double_counted(self) -> None:
        markdown = "| a | b |\n| --- | --- |\n| 1 | 2 |\n"
        self.assertEqual(decide.suspected_tables(markdown)[0], 0)

    def test_code_blocks_are_ignored(self) -> None:
        markdown = "```\ncol1    col2    col3\nval1    val2    val3\n```\n"
        self.assertEqual(decide.suspected_tables(markdown)[0], 0)


class GapSeverityTest(unittest.TestCase):
    def test_missing_tables_are_major_and_drive_an_upgrade(self) -> None:
        payload = assess(markdown=BODY * 6, detected={"pages": 10, "tables": 12, "images": 0})

        gap = next(item for item in payload["gaps"] if item["kind"] == "tables")
        self.assertEqual(gap["severity"], "major")
        self.assertEqual((gap["detected"], gap["extracted"]), (12, 0))
        self.assertEqual(gap["confidence"], "exact")
        self.assertNotEqual(payload["decision"]["recommended"], "accept")

    def test_one_or_two_missing_images_stay_minor_and_silent(self) -> None:
        payload = assess(detected={"pages": 3, "images": 2}, asset_count=1)

        gap = next(item for item in payload["gaps"] if item["kind"] == "images")
        self.assertEqual(gap["severity"], "minor")
        self.assertEqual(payload["decision"]["recommended"], "accept")
        self.assertEqual([r["action"] for r in payload["recommendations"]], ["accept"])

    def test_charts_are_reported_but_never_recommended(self) -> None:
        payload = assess(detected={"pages": 1, "charts": 2})

        gap = next(item for item in payload["gaps"] if item["kind"] == "charts")
        self.assertEqual(gap["resolvable_by"], [])
        self.assertEqual([r["action"] for r in payload["recommendations"]], ["accept"])

    def test_empty_output_is_blocking_not_merely_degraded(self) -> None:
        payload = assess(markdown="", detected={"pages": 12}, element_stats={"text": 0})

        gap = next(item for item in payload["gaps"] if item["kind"] == "text")
        self.assertEqual(gap["severity"], "blocking")
        self.assertFalse(payload["decision"]["usable"])
        self.assertTrue(payload["blocking"])

    def test_a_short_but_complete_document_is_still_usable(self) -> None:
        payload = assess(markdown="| a | b |\n| --- | --- |\n| 1 | x |\n", detected={})
        self.assertTrue(payload["decision"]["usable"])
        self.assertEqual(payload["decision"]["recommended"], "accept")


class DecisionMatrixTest(unittest.TestCase):
    """API key present/absent x upgrade policy -> which option leads."""

    def setUp(self) -> None:
        self.detected = {"pages": 2, "tables": 12, "images": 8}

    def _actions(self, **overrides):
        payload = assess(detected=self.detected, **overrides)
        return payload, [item["action"] for item in payload["recommendations"]]

    def test_install_leads_when_no_api_key_is_configured(self) -> None:
        with patch.dict(os.environ, {}, clear=False):
            os.environ.pop("DECKFLOW_API_KEY", None)
            payload, actions = self._actions()

        self.assertEqual(payload["decision"]["recommended"], "install")
        self.assertIn("cloud", actions)  # still offered, just not first
        cloud = next(item for item in payload["recommendations"] if item["action"] == "cloud")
        self.assertFalse(cloud["available"])
        self.assertIn("auth:cloud", cloud["blocked_by"])

    def test_accept_is_always_offered_alongside_the_upgrades(self) -> None:
        _, actions = self._actions()
        self.assertIn("accept", actions)

    def test_upgrade_never_hides_install_suggestions(self) -> None:
        _, actions = self._actions(upgrade="never")
        self.assertNotIn("install", actions)
        self.assertIn("accept", actions)

    def test_install_option_is_structured_without_embedding_cli_commands(self) -> None:
        payload, _ = self._actions()
        install = next(item for item in payload["recommendations"] if item["action"] == "install")

        self.assertEqual(install["capability"], "pdf")
        self.assertNotIn("commands", install)
        self.assertTrue(install["rerun_required"])
        self.assertEqual(install["cost"]["size_mb"], 56)
        self.assertTrue(install["cost"]["reversible"])

    def test_summary_names_the_evidence_not_just_the_upgrade(self) -> None:
        payload, _ = self._actions()
        install = next(item for item in payload["recommendations"] if item["action"] == "install")
        self.assertIn("表格", install["summary"])
        self.assertIn("0/12", install["summary"])


class ProbePrecisionTest(unittest.TestCase):
    """Counts labelled ``exact`` must match ground truth exactly."""

    def setUp(self) -> None:
        self.tmp = Path(tempfile.mkdtemp(prefix="df-test-"))

    @unittest.skipUnless(pptx_adapter.available()[0], "python-pptx needed to build the fixture")
    def test_pptx_counts_are_exact(self) -> None:
        source = self.tmp / "deck.pptx"
        write_test_pptx(source)

        counts = probes.container_counts("pptx", source)

        self.assertEqual(counts["slides"], 1)
        self.assertEqual(counts["tables"], 1)
        self.assertEqual(counts["images"], 1)
        self.assertEqual(counts["notes"], 1)
        self.assertGreater(counts["text_runs"], 0)

    def test_xlsx_counts_are_exact(self) -> None:
        source = self.tmp / "book.xlsx"
        write_test_xlsx(source)

        counts = probes.container_counts("xlsx", source)

        self.assertEqual(counts["sheets"], 2)
        self.assertEqual(counts["charts"], 0)

    def test_iwork_counts_the_unreadable_text_blocks(self) -> None:
        source = self.tmp / "deck.key"
        with zipfile.ZipFile(source, "w") as archive:
            for index in range(5):
                archive.writestr(f"Index/Slide-{index}.iwa", b"\x00")
            for index in range(3):
                archive.writestr(f"Data/image-{index}.png", TINY_PNG)

        counts = probes.container_counts("key", source)

        self.assertEqual(counts["text_blocks"], 5)
        self.assertEqual(counts["images"], 3)
        self.assertNotIn("slides", counts)  # never guessed

    def test_a_probe_failure_never_breaks_the_run(self) -> None:
        broken = self.tmp / "broken.pptx"
        broken.write_bytes(b"not-a-zip")
        self.assertEqual(probes.container_counts("pptx", broken), {})


class UnrecoverableTest(unittest.TestCase):
    def test_a_vectorised_deck_gets_no_promise_of_a_fix(self) -> None:
        """Zero text runs in the source: no engine can recover what was never stored."""
        payload = assess(
            format_id="pptx",
            engine=next(e for e in registry.engines_for("pptx") if e.name == "ooxml-pptx"),
            markdown="",
            element_stats={"text": 0},
            detected={"slides": 1, "text_runs": 0, "images": 0},
        )

        gap = next(item for item in payload["gaps"] if item["kind"] == "text")
        self.assertEqual(gap["severity"], "blocking")
        self.assertEqual(gap["resolvable_by"], [])
        self.assertEqual(payload["decision"]["recommended"], "input")
        self.assertIn("矢量", payload["decision"]["reason"])


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