from __future__ import annotations

import json
import tempfile
import threading
import unittest
from dataclasses import replace
from pathlib import Path
from unittest.mock import patch

from latexprep.author_records import AuthorRecordOptions, check_author_records
from latexprep.errors import PreparationError
from latexprep.option_config import read_option_groups, read_options
from latexprep.pdf_structure_checks import (
    HeadingExpectation,
    StructureOptions,
    inspect_pdf_structure,
)
from latexprep.pdf_tool_output import MAX_INSPECTION_PAGES
from latexprep.resource_budget import ResourceBudget
from tests.pdf_fakes import DetailRunner


def structured_pdf_objects() -> dict[str, object]:
    def entry(value: dict[str, object]) -> dict[str, object]:
        return {"value": value}

    objects = {
        "trailer": entry({"/Root": "1 0 R"}),
        "obj:1 0 R": entry({"/Type": "/Catalog", "/Pages": "2 0 R", "/StructTreeRoot": "4 0 R"}),
        "obj:2 0 R": entry({"/Type": "/Pages", "/Kids": ["3 0 R"], "/Count": 1}),
        "obj:3 0 R": entry(
            {"/Type": "/Page", "/Parent": "2 0 R", "/StructParents": 0, "/Annots": ["10 0 R"]}
        ),
        "obj:4 0 R": entry({"/Type": "/StructTreeRoot", "/K": ["5 0 R"], "/ParentTree": "11 0 R"}),
        "obj:5 0 R": entry(
            {
                "/Type": "/StructElem",
                "/S": "/Document",
                "/P": "4 0 R",
                "/Pg": "3 0 R",
                "/K": ["6 0 R", "9 0 R", "12 0 R"],
            }
        ),
        "obj:6 0 R": entry(
            {"/Type": "/StructElem", "/S": "/Table", "/P": "5 0 R", "/K": ["7 0 R"]}
        ),
        "obj:7 0 R": entry({"/Type": "/StructElem", "/S": "/TR", "/P": "6 0 R", "/K": ["8 0 R"]}),
        "obj:8 0 R": entry({"/Type": "/StructElem", "/S": "/TH", "/P": "7 0 R", "/K": 0}),
        "obj:9 0 R": entry(
            {
                "/Type": "/StructElem",
                "/S": "/Figure",
                "/P": "5 0 R",
                "/K": 1,
                "/Alt": "u:Measured result chart",
            }
        ),
        "obj:10 0 R": entry(
            {
                "/Type": "/Annot",
                "/Subtype": "/Link",
                "/StructParent": 1,
                "/A": {"/S": "/URI", "/URI": "u:https://example.org"},
            }
        ),
        "obj:11 0 R": entry({"/Nums": [0, ["8 0 R", "9 0 R", "12 0 R"], 1, "12 0 R"]}),
        "obj:12 0 R": entry(
            {
                "/Type": "/StructElem",
                "/S": "/Link",
                "/P": "5 0 R",
                "/K": [2, {"/Type": "/OBJR", "/Pg": "3 0 R", "/Obj": "10 0 R"}],
            }
        ),
    }
    return {
        "version": 2,
        "pages": [{"object": "3 0 R", "pageposfrom1": 1}],
        "qpdf": [{"jsonversion": 2}, objects],
    }


def object_value(document: dict[str, object], number: int) -> dict[str, object]:
    return document["qpdf"][1][f"obj:{number} 0 R"]["value"]


class AuthorRecordTests(unittest.TestCase):
    def check(self, preamble: str, options: AuthorRecordOptions | None = None):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root / "main.tex").write_text(
                "\\documentclass{article}\n"
                + preamble
                + "\n\\begin{document}Body.\\end{document}\n"
            )
            return check_author_records(
                root,
                "main.tex",
                options or AuthorRecordOptions(required_author_fields=("affiliation", "email")),
            )

    def test_each_author_needs_its_own_fields(self) -> None:
        complete = (
            "\\author{Ada}\\affiliation{Lab A}\\email{ada@example.org}\n"
            "\\author{Grace}\\affiliation{Lab B}\\email{grace@example.org}"
        )
        passed = self.check(complete)[0]
        self.assertEqual(passed.code, "MAN201")
        self.assertEqual(passed.status, "passed")
        self.assertEqual(len(passed.details["records"]), 2)
        failed = self.check(complete.replace("\\email{grace@example.org}", ""))[0]
        self.assertEqual(failed.status, "failed")
        self.assertEqual(failed.details["violations"][0]["author_index"], 2)
        self.assertEqual(failed.details["violations"][0]["missing_fields"], ["email"])
        custom = self.check(
            "\\newcommand{\\person}[1]{#1}\n"
            "\\person{Ada}\\contact{ada@example.org}\n% \\person{Ignored}\n",
            AuthorRecordOptions(author_command="person", required_author_fields=("contact",)),
        )[0]
        self.assertEqual(custom.status, "passed")
        self.assertEqual(len(custom.details["records"]), 1)
        self.assertEqual(self.check("\\email{orphan@example.org}\\author{Ada}")[0].status, "failed")

    def test_a_declaration_of_the_author_command_is_not_a_record_however_it_is_written(
        self,
    ) -> None:
        for declaration in (
            "\\newcommand{\\person}[1]{#1}",
            "\\NewDocumentCommand{\\person}{m}{#1}",
            "\\DeclareDocumentCommand{\\person}{m}{#1}",
        ):
            with self.subTest(declaration=declaration):
                result = self.check(
                    declaration + "\\person{Ada}\\contact{ada@example.org}",
                    AuthorRecordOptions(
                        author_command="person", required_author_fields=("contact",)
                    ),
                )[0]
                self.assertEqual(result.status, "passed")
                self.assertEqual(len(result.details["records"]), 1)

    def test_grouped_and_dynamic_records_are_inconclusive(self) -> None:
        for source in (
            "\\author{Ada}\\affiliation{Lab}\\box{\\email{ada@example.org}}",
            "\\author{\\hiddenname}\\affiliation{Lab}\\email{ada@example.org}",
            "\\iftrue\\author{Ada}\\affiliation{Lab}\\email{ada@example.org}\\fi",
        ):
            with self.subTest(source=source):
                self.assertEqual(self.check(source)[0].status, "inconclusive")

    def test_defined_conditional_names_do_not_block_author_records(self) -> None:
        result = self.check(
            "\\newif\\ifanonymous\\anonymousfalse\n"
            "\\author{Ada}\\affiliation{Lab A}\\email{ada@example.org}"
        )[0]
        self.assertEqual(result.status, "passed", result.details["uncertainty"])

    def test_failed_author_records_report_counts_and_locations(self) -> None:
        result = self.check("\\author{Ada}\\affiliation{Lab A}\n\\author{Grace}")[0]
        self.assertEqual(result.status, "failed")
        self.assertEqual(
            result.message,
            "Incomplete author records (2 of 2): main.tex:2 (missing \\email), "
            "main.tex:3 (missing \\affiliation, \\email).",
        )

    def test_a_record_with_two_problems_is_counted_once(self) -> None:
        options = AuthorRecordOptions(required_author_fields=("email",))
        result = self.check("\\author{}", options)[0]
        self.assertEqual(result.status, "failed")
        self.assertEqual(
            result.message,
            "Incomplete author records (1 of 1): main.tex:2 (empty or placeholder name; "
            "missing \\email).",
        )
        self.assertEqual(result.details["violation_count"], 1)
        self.assertEqual(len(result.details["violations"]), 1)
        self.assertEqual(result.details["violations"][0]["author_index"], 1)

    def test_missing_author_records_and_fields_before_the_first_author_are_named(self) -> None:
        none = self.check("\\title{No authors}")[0]
        self.assertEqual(none.status, "failed")
        self.assertNotIn("1 of 1", none.message)
        self.assertIn("No literal \\author author record was found in the preamble", none.message)
        self.assertIn("structure_checks.author_command", none.message)
        first = self.check(
            "\\affiliation{Lab}\\author{Ada}",
            AuthorRecordOptions(required_author_fields=("affiliation",)),
        )[0]
        self.assertEqual(
            first.message,
            "Incomplete author records (1 of 1): main.tex:2 (missing \\affiliation); "
            "Fields placed before the first \\author: \\affiliation at main.tex:2.",
        )

    def test_long_failure_lists_name_the_first_records_and_count_the_rest(self) -> None:
        authors = "".join(f"\\author{{Author {number}}}\n" for number in range(1, 6))
        result = self.check(authors, AuthorRecordOptions(required_author_fields=("email",)))[0]
        self.assertEqual(result.details["violation_count"], 5)
        self.assertTrue(result.message.startswith("Incomplete author records (5 of 5): main.tex:2"))
        self.assertTrue(result.message.endswith("(+2 more)."), result.message)

    def test_option_schema_is_explicit_and_frozen(self) -> None:
        parsed = read_options(
            StructureOptions,
            {"heading_expectations": [{"page": 1, "title": "Methods", "number": "2"}]},
            "structure_checks",
        )
        self.assertEqual(parsed.heading_expectations, (HeadingExpectation(1, "Methods", "2"),))
        for values in (
            {"require_structure_tree": 1},
            {
                "heading_expectations": (
                    HeadingExpectation(1, "Methods", "2"),
                    HeadingExpectation(1, "Methods", "3"),
                )
            },
        ):
            with self.subTest(values=values), self.assertRaises(PreparationError):
                StructureOptions(**values)
        with self.assertRaises(PreparationError):
            HeadingExpectation(True, "Methods", "2")

    def test_author_record_options_are_validated_on_their_own(self) -> None:
        parsed = read_options(AuthorRecordOptions, {"required_author_fields": ["email"]}, "x")
        self.assertEqual(parsed, AuthorRecordOptions(required_author_fields=("email",)))
        self.assertEqual(AuthorRecordOptions().author_command, "author")
        for values in (
            {"author_command": "\\author"},
            {"required_author_fields": ("email", "email")},
            {"required_author_fields": ("author",)},
        ):
            with self.subTest(values=values), self.assertRaises(PreparationError):
                AuthorRecordOptions(**values)

    def test_structure_options_hold_no_author_record_options(self) -> None:
        for name in ("author_command", "required_author_fields"):
            with self.assertRaises(TypeError):
                StructureOptions(**{name: ()})

    def test_one_table_fills_both_option_groups_and_reports_every_valid_key(self) -> None:
        structure, authors = read_option_groups(
            [StructureOptions, AuthorRecordOptions],
            {"check_table_structure": True, "required_author_fields": ["email"]},
            "structure_checks",
        )
        self.assertTrue(structure.check_table_structure)
        self.assertEqual(authors.required_author_fields, ("email",))
        with self.assertRaisesRegex(
            PreparationError,
            r"Unknown structure_checks setting\(s\): author_comand "
            r"\(did you mean author_command\?\)\. "
            r"Valid structure_checks settings: author_command, ",
        ):
            read_option_groups(
                [StructureOptions, AuthorRecordOptions], {"author_comand": "x"}, "structure_checks"
            )
        with self.assertRaisesRegex(PreparationError, "structure_checks must be a settings table"):
            read_option_groups([StructureOptions, AuthorRecordOptions], 3, "structure_checks")

    def test_an_option_cannot_belong_to_two_groups_of_one_table(self) -> None:
        with self.assertRaisesRegex(TypeError, "is an option of both"):
            read_option_groups([StructureOptions, StructureOptions], {}, "structure_checks")


class PdfStructureTests(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        self.root = Path(self.temporary.name)
        self.pdf = self.root / "document.pdf"
        self.pdf.write_bytes(b"%PDF-controlled-test")
        self.count = 0

    async def asyncTearDown(self) -> None:
        self.temporary.cleanup()

    async def check(self, options: StructureOptions, document=None, **outputs):
        self.count += 1
        runner = DetailRunner(
            info="Pages: 1\n",
            objects=json.dumps(document or structured_pdf_objects()),
            text="2 Methods\nBody.\f",
            **outputs,
        )
        result = await inspect_pdf_structure(
            self.pdf, self.root / f"inspect-{self.count}", runner, options
        )
        return result

    async def test_heading_expectations_match_differ_and_remain_uncertain(self) -> None:
        options = StructureOptions(heading_expectations=(HeadingExpectation(1, "Methods", "2"),))
        passed = (await self.check(options))[0]
        self.assertEqual(passed.code, "PDF401")
        self.assertEqual(passed.status, "passed")
        wrong = replace(options, heading_expectations=(HeadingExpectation(1, "Methods", "3"),))
        self.assertEqual((await self.check(wrong))[0].status, "failed")
        outside = replace(options, heading_expectations=(HeadingExpectation(2, "Methods", "2"),))
        self.assertEqual((await self.check(outside))[0].status, "inconclusive")
        for text in ("\f", "2 Methods\n2 Methods\f", "2 Methods without page boundary"):
            self.count += 1
            result = await inspect_pdf_structure(
                self.pdf,
                self.root / f"inspect-{self.count}",
                DetailRunner(info="Pages: 1\n", text=text),
                options,
            )
            self.assertEqual(result[0].status, "inconclusive")

    async def test_tagged_tables_have_rows_and_cells(self) -> None:
        options = StructureOptions(check_table_structure=True)
        passed = (await self.check(options))[0]
        self.assertEqual(passed.code, "PDF402")
        self.assertEqual(passed.status, "passed")
        self.assertEqual(passed.details["tables"][0]["header_cells"], 1)
        document = structured_pdf_objects()
        object_value(document, 8)["/S"] = "/P"
        self.assertEqual((await self.check(options, document))[0].status, "failed")
        document = structured_pdf_objects()
        object_value(document, 6)["/K"] = []
        self.assertEqual((await self.check(options, document))[0].status, "failed")

    async def test_link_tags_match_real_page_annotations(self) -> None:
        options = StructureOptions(check_link_structure=True)
        passed = (await self.check(options))[0]
        self.assertEqual(passed.code, "PDF403")
        self.assertEqual(passed.status, "passed")
        for change in (
            "association",
            "target",
            "empty_uri",
            "empty_destination",
            "wrong_page",
            "repeated",
        ):
            document = structured_pdf_objects()
            if change == "association":
                object_value(document, 12)["/K"] = 2
            elif change == "target":
                object_value(document, 10).pop("/A")
            elif change == "empty_uri":
                object_value(document, 10)["/A"] = {"/S": "/URI", "/URI": "u:"}
            elif change == "empty_destination":
                object_value(document, 10)["/A"] = {"/S": "/GoTo", "/D": None}
            elif change == "wrong_page":
                object_value(document, 12)["/K"] = [
                    2,
                    {"/Type": "/OBJR", "/Pg": "90 0 R", "/Obj": "10 0 R"},
                ]
            else:
                object_value(document, 3)["/Annots"] = ["10 0 R", "10 0 R"]
            with self.subTest(change=change):
                self.assertEqual((await self.check(options, document))[0].status, "failed")
        for destination, status in (
            (["3 0 R", "/Fit"], "passed"),
            (["90 0 R", "/Fit"], "failed"),
            ("u:named-destination", "inconclusive"),
        ):
            document = structured_pdf_objects()
            object_value(document, 10)["/A"] = {"/S": "/GoTo", "/D": destination}
            with self.subTest(destination=destination):
                self.assertEqual((await self.check(options, document))[0].status, status)

    async def test_figure_alt_presence_and_unreadable_strings(self) -> None:
        options = StructureOptions(require_figure_alt=True)
        passed = (await self.check(options))[0]
        self.assertEqual(passed.code, "PDF404")
        self.assertEqual(passed.status, "passed")
        for alt, status in ((None, "failed"), ("u:  ", "failed"), ("b:FEFF0041", "inconclusive")):
            document = structured_pdf_objects()
            object_value(document, 9)["/Alt"] = alt
            self.assertEqual((await self.check(options, document))[0].status, status)
        document = structured_pdf_objects()
        object_value(document, 9)["/S"] = "/Picture"
        self.assertEqual((await self.check(options, document))[0].status, "inconclusive")
        object_value(document, 4)["/RoleMap"] = {"/Picture": "/Figure"}
        self.assertEqual((await self.check(options, document))[0].status, "passed")

    async def test_structure_references_reject_duplicate_missing_and_invalid_mcid_links(
        self,
    ) -> None:
        options = StructureOptions(require_structure_tree=True, check_structure_references=True)
        passed = (await self.check(options))[0]
        self.assertEqual(passed.code, "PDF405")
        self.assertEqual(passed.status, "passed")
        self.assertEqual(passed.details["marked_content_count"], 3)
        for mutation in (
            "duplicate",
            "missing_mcid",
            "missing_parent",
            "wrong_parent",
            "invalid_page",
        ):
            document = structured_pdf_objects()
            if mutation == "duplicate":
                object_value(document, 9)["/K"] = [1, 1]
            elif mutation == "missing_mcid":
                object_value(document, 9)["/K"] = {"/Type": "/MCR", "/Pg": "3 0 R"}
            elif mutation == "missing_parent":
                object_value(document, 3).pop("/StructParents")
            elif mutation == "wrong_parent":
                object_value(document, 11)["/Nums"] = [0, ["9 0 R", "9 0 R", "12 0 R"], 1, "12 0 R"]
            else:
                object_value(document, 9)["/Pg"] = "90 0 R"
            with self.subTest(mutation=mutation):
                self.assertEqual((await self.check(options, document))[0].status, "failed")

    async def test_repeated_unsupported_roles_are_reported_once(self) -> None:
        document = structured_pdf_objects()
        # Several elements with one unmapped role must not repeat the recorded uncertainty.
        object_value(document, 5)["/K"] = ["6 0 R", "9 0 R", "12 0 R", "13 0 R", "14 0 R"]
        objects = document["qpdf"][1]
        for number in (13, 14):
            objects[f"obj:{number} 0 R"] = {
                "value": {"/Type": "/StructElem", "/S": "/Picture", "/P": "5 0 R"}
            }
        for options in (
            StructureOptions(check_table_structure=True),
            StructureOptions(require_structure_tree=True, check_structure_references=True),
            StructureOptions(check_link_structure=True),
        ):
            with self.subTest(options=options):
                finding = (await self.check(options, document))[0]
                self.assertEqual(finding.status, "inconclusive")
                self.assertEqual(
                    finding.details["uncertainty"],
                    ["unmapped or unsupported PDF structure role"],
                )

    async def test_the_structure_tree_is_read_in_a_worker_thread_that_sees_cancellation(
        self,
    ) -> None:
        seen: list[int] = []
        with patch(
            "latexprep.pdf_structure_tree.cancellation_point",
            side_effect=lambda: seen.append(threading.get_ident()),
        ):
            result = await self.check(StructureOptions(check_table_structure=True))
        self.assertEqual(result[0].status, "passed")
        self.assertTrue(seen)
        # Off the event loop's thread, where a cancellation request can reach the traversal.
        self.assertNotIn(threading.get_ident(), seen)

    async def test_link_violations_carry_a_readable_reason(self) -> None:
        document = structured_pdf_objects()
        object_value(document, 10).pop("/A")
        finding = (await self.check(StructureOptions(check_link_structure=True), document))[0]
        self.assertEqual(finding.status, "failed")
        reasons = [item["reason"] for item in finding.details["violations"]]
        self.assertIn("link annotation has no destination or action", reasons)

    async def test_heading_expectations_can_name_pages_up_to_the_inspection_limit(self) -> None:
        pages = 301
        options = StructureOptions(
            heading_expectations=(HeadingExpectation(pages, "Appendix", "A"),)
        )
        runner = DetailRunner(info=f"Pages: {pages}\n", text="\f" * (pages - 1) + "A Appendix\f")
        result = await inspect_pdf_structure(self.pdf, self.root / "late-heading", runner, options)
        self.assertEqual(result[0].status, "passed")
        HeadingExpectation(MAX_INSPECTION_PAGES, "Appendix", "A")
        with self.assertRaisesRegex(PreparationError, str(MAX_INSPECTION_PAGES)):
            HeadingExpectation(MAX_INSPECTION_PAGES + 1, "Appendix", "A")

    async def test_missing_or_malformed_structure_and_tools_are_not_passes(self) -> None:
        options = StructureOptions(require_structure_tree=True, require_figure_alt=True)
        document = structured_pdf_objects()
        object_value(document, 1).pop("/StructTreeRoot")
        result = {finding.code: finding for finding in await self.check(options, document)}
        self.assertEqual(result["PDF405"].status, "failed")
        self.assertEqual(result["PDF404"].status, "inconclusive")
        document = structured_pdf_objects()
        object_value(document, 5)["/K"] = ["5 0 R"]
        self.assertTrue(
            all(f.status == "inconclusive" for f in await self.check(options, document))
        )
        document = structured_pdf_objects()
        object_value(document, 11)["/Nums"] = [part for key in range(10001) for part in (key, [])]
        self.assertTrue(
            all(f.status == "inconclusive" for f in await self.check(options, document))
        )
        options = replace(options, heading_expectations=(HeadingExpectation(1, "Methods", "2"),))
        self.count += 1
        runner = DetailRunner(info="Pages: 1\n", text="2 Methods\f")
        runner.missing = "qpdf"
        result = await inspect_pdf_structure(
            self.pdf, self.root / f"inspect-{self.count}", runner, options
        )
        findings = {finding.code: finding for finding in result}
        self.assertEqual(findings["PDF401"].status, "passed")
        self.assertEqual(findings["PDF405"].status, "inconclusive")
        self.assertEqual(findings["PDF404"].status, "inconclusive")

    async def test_structure_probes_keep_frozen_inputs_and_shared_resource_lease(self) -> None:
        budget = ResourceBudget(2, 256)
        runner = DetailRunner(info="Pages: 1\n", objects=json.dumps(structured_pdf_objects()))
        runner.expected_budget = budget
        options = StructureOptions(require_structure_tree=True, check_table_structure=True)
        results = await inspect_pdf_structure(
            self.pdf, self.root / "leased", runner, options, budget
        )
        self.assertTrue(all(finding.status == "passed" for finding in results))
        inputs = [
            paths for command, _, paths in runner.calls if command[-1] not in {"-v", "--version"}
        ]
        self.assertTrue(inputs)
        self.assertTrue(all(len(paths) == 1 for paths in inputs))
        self.assertEqual(len({paths[0] for paths in inputs}), 1)
        self.assertNotEqual(inputs[0][0], self.pdf)


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