import copy
import os
import tempfile
import unittest

from tests.fakes import FakeGitea
from werkrevisor import codex, reviewreport

HEAD = "6d1842aa2111a50cbb0614aa840ed1cb8a75985c"
DRY_RUN = f"""<!-- werkrevisor:{HEAD} -->
<!-- werkrevisor-verdict:YELLOW -->
## 🟡 werkrevisor: YELLOW / HUMAN_REVIEW

Change kind: `mixed`.

### Reasons
- **YELLOW** risky paths changed: build.gradle.kts

<sub>
Usage: model `m`, cost 0.17 USD (reported).<br>
Reviewed head `{HEAD}`.
</sub>

{{"verdict": "YELLOW"}}
"""
FINDING = {"severity": "medium", "file": "tools/x", "line": 148, "title": "Claims collapse", "explanation": "Why."}


def record(answer, exit_code=0):
    return {"model": "gpt-6-astra", "effort": "high", "exit_code": exit_code, "seconds": 72.2,
            "usage": {"input_tokens": 311095, "cached_input_tokens": 256000, "output_tokens": 2143,
                      "reasoning_output_tokens": 643}, "commands": [], "errors": [], "answer": answer}


class ReviewReportTest(unittest.TestCase):
    def test_the_werkrevisor_part_has_no_markers(self):
        title, body, usage = reviewreport.werkrevisor_part(DRY_RUN)
        self.assertEqual(title, "🟡 YELLOW / HUMAN_REVIEW")
        self.assertIn("#### Reasons", body)
        self.assertNotIn("<!--", body)
        self.assertNotIn('"verdict"', body)
        self.assertEqual(usage, "model `m`, cost 0.17 USD (reported).")
        self.assertIsNone(reviewreport.werkrevisor_part("werkrevisor: configuration error"))

    def test_the_report_holds_both_reviews_and_asks_for_a_check(self):
        answer = {"critical": True, "critical_reason": "Changes shared test infrastructure.", "security_effect": "none", "fails_closed": False, "hardening_tested": False, "findings": [FINDING]}
        report = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN),
                                      reviewreport.codex_part(record(answer)))
        self.assertTrue(report.startswith("## AI review of `6d1842aa` (werkrevisor and Codex)"))
        self.assertIn("check every finding in the code before posting", report)
        self.assertIn("### werkrevisor: 🟡 YELLOW / HUMAN_REVIEW", report)
        self.assertIn("Codex rates the PR as critical. Changes shared test infrastructure.\n\n"
                      "Security effect: none; fails closed: no; a test of the closed case: no.", report)
        self.assertIn("1. **[medium] Claims collapse** at `tools/x:148`", report)
        self.assertIn("input tokens 311095 (256000 cached)", report)

    def test_a_failed_review_is_named_not_left_out(self):
        report = reviewreport.compose(HEAD, None, reviewreport.codex_part(record({"unparsed": "x"})),
                                      "docker: not found", "quota exceeded")
        self.assertIn("### werkrevisor\n\nThe review did not complete: docker: not found", report)
        self.assertIn("### Codex\n\nThe review did not complete: quota exceeded", report)
        self.assertIsNone(reviewreport.codex_part(record({"critical": False, "critical_reason": "", "security_effect": "none", "fails_closed": False, "hardening_tested": False, "findings": []},
                                                         exit_code=1)))
        self.assertTrue(codex.valid_answer({"critical": False, "critical_reason": "", "security_effect": "none", "fails_closed": False, "hardening_tested": False, "findings": []}))

    def test_an_empty_reason_leaves_no_colon(self):
        text, _ = reviewreport.codex_part(record({"critical": False, "critical_reason": " ", "security_effect": "none", "fails_closed": False, "hardening_tested": False, "findings": []}))
        self.assertTrue(text.startswith("Codex rates the PR as not critical.\n"))

    def test_each_error_is_cut_on_its_own(self):
        self.assertEqual(codex.error_text(["a" * 600, "first", "b" * 10 + "end"], limit=5), "first; bbend")

    def test_reviews_of_different_commits_are_flagged(self):
        other = "f" * 40
        self.assertEqual(reviewreport.reviewed_head(DRY_RUN), HEAD)
        werkrevisor = reviewreport.werkrevisor_part(DRY_RUN)
        self.assertNotIn("Warning", reviewreport.compose(HEAD, werkrevisor, None, werkrevisor_head=HEAD))
        self.assertIn("> **Warning:** werkrevisor reviewed `6d1842aa`, Codex `ffffffff`",
                      reviewreport.compose(other, werkrevisor, None, werkrevisor_head=HEAD))

    def test_a_failed_codex_run_keeps_the_head_and_warns_of_no_mismatch(self):
        report = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="timeout",
                                      werkrevisor_head=None)
        self.assertTrue(report.startswith("## AI review of `6d1842aa`"))
        self.assertNotIn("Warning", report)

    def test_the_draft_links_the_diagrams_next_to_it(self):
        """Scenario#129.02"""
        with tempfile.TemporaryDirectory() as tmp:
            source, target = os.path.join(tmp, "drawn"), os.path.join(tmp, "draft")
            os.makedirs(source)
            os.makedirs(target)
            for name in ("pr-lens-flow.svg", "pr-lens-containers.svg", "notes.txt"):
                open(os.path.join(source, name), "w").write("<svg/>")
            open(os.path.join(target, "pr-lens-old.svg"), "w").write("<svg/>")

            names = reviewreport.copy_diagrams(source, target)
            report = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x",
                                          diagrams=names)

            self.assertEqual(names, ["pr-lens-containers.svg", "pr-lens-flow.svg"])
            self.assertEqual(sorted(os.listdir(target)), names)
            self.assertIn("### Diagrams", report)
            self.assertIn("![pr-lens-flow.svg](pr-lens-flow.svg)", report)
            self.assertLess(report.index("### Diagrams"), report.index("### Codex"))

    def test_a_run_without_new_diagrams_shows_none_of_an_earlier_run(self):
        """Scenario#129.02"""
        with tempfile.TemporaryDirectory() as tmp:
            source, target = os.path.join(tmp, "drawn"), os.path.join(tmp, "draft")
            os.makedirs(source)
            os.makedirs(target)
            for directory in (source, target):
                open(os.path.join(directory, "pr-lens-earlier.svg"), "w").write("<svg/>")

            reviewreport.clear_diagrams(source)
            names = reviewreport.copy_diagrams(source, target)

            self.assertEqual((names, os.listdir(source), os.listdir(target)), ([], [], []))
            self.assertNotIn("### Diagrams", reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None,
                                                                   diagrams=names))

    def test_without_diagrams_the_draft_has_no_section(self):
        with tempfile.TemporaryDirectory() as tmp:
            self.assertEqual(reviewreport.copy_diagrams(os.path.join(tmp, "none"), tmp), [])
        self.assertNotIn("### Diagrams", reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None))


class AttachTest(unittest.TestCase):
    def setUp(self):
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        self.dir = tmp.name
        open(os.path.join(self.dir, "pr-lens-flow.svg"), "w").write("<svg/>")
        report = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x",
                                      diagrams=["pr-lens-flow.svg"])
        self.checked = report.replace(report.split("\n")[2] + "\n", "")
        self.gitea = FakeGitea()

    def test_a_checked_draft_is_posted_with_its_diagrams(self):
        """Scenario#130.01"""
        url = reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)

        comment = self.gitea.comments[-1]
        self.assertEqual(url, comment["html_url"])
        self.assertIn(f"<!-- hs-review:{HEAD} -->", comment["body"])
        self.assertIn("![pr-lens-flow.svg](https://git.example.org/attachments/1-pr-lens-flow.svg)", comment["body"])
        self.assertNotIn("Attach the SVGs", comment["body"])
        self.assertEqual([a["name"] for a in comment["assets"]], ["pr-lens-flow.svg"])

    def test_an_unchecked_draft_is_refused(self):
        """Scenario#130.02"""
        draft = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x")
        with self.assertRaisesRegex(reviewreport.AttachError, "not checked"):
            reviewreport.attach(draft, self.dir, self.gitea, 7, HEAD)
        self.assertEqual(self.gitea.writes, [])

    def test_a_draft_of_another_head_is_refused(self):
        """Scenario#130.02"""
        with self.assertRaisesRegex(reviewreport.AttachError, "run hs-review again"):
            reviewreport.attach(self.checked, self.dir, self.gitea, 7, "f" * 40)
        self.assertEqual(self.gitea.writes, [])

    def test_a_second_post_for_the_same_head_replaces_the_report(self):
        """Scenario#139.02"""
        first = reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)

        second = reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="Finding 2 does not hold.")

        self.assertEqual(second, first)
        self.assertEqual(len(self.gitea.comments), 1)
        comment = self.gitea.comments[0]
        self.assertTrue(comment["body"].startswith(f"<!-- hs-review:{HEAD} -->\n"))
        self.assertIn("Finding 2 does not hold.", comment["body"])
        self.assertEqual([a["id"] for a in comment["assets"]], [2])
        self.assertIn("attachments/2-pr-lens-flow.svg", comment["body"])
        self.assertNotIn("attachments/1-pr-lens-flow.svg", comment["body"])

        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="A third reason.")
        self.assertEqual([a["id"] for a in self.gitea.comments[0]["assets"]], [3])
        self.assertIn("attachments/3-pr-lens-flow.svg", self.gitea.comments[0]["body"])

    def test_a_replacement_keeps_a_file_added_by_hand(self):
        """Scenario#139.04"""
        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
        self.gitea.create_comment_asset(1, "server.log", b"log", "text/plain")

        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="Another reason.")

        self.assertEqual(sorted(a["name"] for a in self.gitea.comments[0]["assets"]), ["pr-lens-flow.svg", "server.log"])

    def test_a_replacement_keeps_an_svg_linked_by_hand(self):
        """Scenario#139.04"""
        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
        custom = self.gitea.create_comment_asset(1, "pr-lens-custom.svg", b"<svg/>", "image/svg+xml")
        comment = self.gitea.comments[0]
        comment["body"] += f"\n![custom]({custom['browser_download_url']})"

        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="Another reason.")

        self.assertIn("pr-lens-custom.svg", [a["name"] for a in comment["assets"]])

    def test_a_replacement_keeps_an_old_svg_that_the_reason_links(self):
        """Scenario#139.03"""
        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
        url = self.gitea.comments[0]["assets"][0]["browser_download_url"]

        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked=f"See the [old diagram]({url}).")
        self.assertIn(url, [a["browser_download_url"] for a in self.gitea.comments[0]["assets"]])

        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="No link.")

        self.assertNotIn(url, [a["browser_download_url"] for a in self.gitea.comments[0]["assets"]])

    def test_an_edit_applied_despite_a_lost_response_keeps_its_svgs(self):
        """Scenario#139.03"""
        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
        edit = self.gitea.edit_issue_comment

        def lost(comment_id, body):
            edit(comment_id, body)
            raise OSError("connection reset")
        self.gitea.edit_issue_comment = lost

        with self.assertRaises(OSError):
            reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="Another reason.")
        comment = self.gitea.comments[0]
        self.assertIn("attachments/2-pr-lens-flow.svg", comment["body"])
        self.assertIn(2, [a["id"] for a in comment["assets"]])

    def test_a_failed_replacement_keeps_the_report_and_its_svgs(self):
        """Scenario#139.03"""
        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
        before = copy.deepcopy(self.gitea.comments)

        def refused(*args):
            raise OSError("HTTP 500")
        self.gitea.edit_issue_comment = refused

        with self.assertRaises(OSError):
            reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="Another reason.")
        self.assertEqual(self.gitea.comments, before)

    def test_a_report_of_someone_else_is_neither_replaced_nor_folded(self):
        """Scenario#139.04"""
        self.gitea.comments.append({"id": 1, "user": {"login": "other"}, "html_url": "u",
                                    "body": f"<!-- hs-review:{HEAD} -->\n## AI review"})

        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)

        self.assertEqual(len(self.gitea.comments), 2)
        self.assertEqual(self.gitea.comments[0]["body"], f"<!-- hs-review:{HEAD} -->\n## AI review")

    def test_a_report_of_a_new_head_folds_the_earlier_reports(self):
        """Scenario#139.05"""
        old = "a" * 40
        earlier = self.checked.replace(f"`{HEAD[:8]}`", f"`{old[:8]}`", 1)
        reviewreport.attach(earlier, self.dir, self.gitea, 7, old)

        url = reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)

        first, second = self.gitea.comments
        self.assertEqual(second["html_url"], url)
        self.assertTrue(first["body"].startswith(f"<!-- hs-review:{old} -->\n<!-- hs-review-folded -->\n<details>"))
        self.assertIn(f'superseded by <a href="{url}">the review of <code>{HEAD[:8]}</code></a>', first["body"])
        self.assertTrue(first["body"].endswith("</details>"))
        self.assertNotIn("<details>", second["body"])

        writes = list(self.gitea.writes)
        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="Reason.")
        self.assertEqual(self.gitea.comments[0]["body"].count("<details>"), 1)
        self.assertEqual(self.gitea.writes.count("edit") - writes.count("edit"), 1)

    def test_a_fold_keeps_an_edit_made_meanwhile(self):
        """Scenario#139.05"""
        old = "a" * 40
        reviewreport.attach(self.checked.replace(f"`{HEAD[:8]}`", f"`{old[:8]}`", 1), self.dir, self.gitea, 7, old)
        listed = self.gitea.list_issue_comments
        self.gitea.list_issue_comments = lambda number: copy.deepcopy(listed(number))
        create = self.gitea.create_issue_comment

        def edited_meanwhile(number, body):
            self.gitea.comments[0]["body"] += "\nAdded by hand."
            return create(number, body)
        self.gitea.create_issue_comment = edited_meanwhile

        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)

        self.assertIn("<details>", self.gitea.comments[0]["body"])
        self.assertIn("Added by hand.", self.gitea.comments[0]["body"])

    def test_a_replaced_report_folds_the_reports_of_other_heads(self):
        """Scenario#139.05"""
        new = "b" * 40
        later = self.checked.replace(f"`{HEAD[:8]}`", f"`{new[:8]}`", 1)
        first = reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
        reviewreport.attach(later, self.dir, self.gitea, 7, new)

        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="Back at this commit.")

        replaced, other = self.gitea.comments
        self.assertNotIn("<details>", replaced["body"])
        self.assertIn(f'superseded by <a href="{first}">the review of <code>{HEAD[:8]}</code></a>', other["body"])

    def test_a_failed_upload_leaves_no_comment_and_allows_a_retry(self):
        """Scenario#130.01"""
        create = self.gitea.create_comment_asset

        def refused(*args):
            raise OSError("HTTP 413")
        self.gitea.create_comment_asset = refused

        with self.assertRaises(OSError):
            reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
        self.assertEqual(self.gitea.comments, [])

        self.gitea.create_comment_asset = create
        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
        self.assertEqual(len(self.gitea.comments), 1)

    def test_a_finding_that_quotes_the_draft_line_is_no_refusal(self):
        """Scenario#130.02"""
        quoted = self.checked.replace("### Codex", "> **Draft:** quoted in a finding\n\n### Codex")

        reviewreport.attach(quoted, self.dir, self.gitea, 7, HEAD)

        self.assertIn("quoted in a finding", self.gitea.comments[-1]["body"])

    def test_a_reason_follows_the_report_as_assessment(self):
        """Scenario#139.01"""
        draft = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x",
                                     diagrams=["pr-lens-flow.svg"])
        reason = "Finding 1 is fixed in a1b2c3d.\n\nFinding 2 does not hold: the check runs in the repository."

        reviewreport.attach(draft, self.dir, self.gitea, 7, HEAD, checked=reason)

        body = self.gitea.comments[-1]["body"]
        self.assertNotIn("> **Draft:**", body)
        self.assertIn("### Assessment of the findings\n\nFinding 1 is fixed in a1b2c3d.\n\n"
                      "Finding 2 does not hold: the check runs in the repository.\n", body)
        self.assertLess(body.index("### Codex"), body.index("### Assessment of the findings"))
        self.assertLess(body.index("### Assessment of the findings"), body.index("<sub>"))
        self.assertTrue(body.rstrip().endswith("</sub>"))
        self.assertNotIn("\n\n\n", body)

    def test_a_reason_that_quotes_the_draft_line_stays_whole(self):
        """Scenario#139.01"""
        draft = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x")

        reviewreport.attach(draft, self.dir, self.gitea, 7, HEAD,
                            checked="> **Draft:** the remaining finding is already covered.\nSecond line.")

        body = self.gitea.comments[-1]["body"]
        self.assertIn("### Assessment of the findings\n\n> **Draft:** the remaining finding is already covered.\n"
                      "Second line.\n", body)

    def test_an_empty_reason_is_refused_before_any_write(self):
        """Scenario#136.02"""
        draft = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x")
        for reason in ("", "  \n\t\n"):
            with self.subTest(reason=reason), self.assertRaisesRegex(reviewreport.AttachError, "needs a reason"):
                reviewreport.attach(draft, self.dir, self.gitea, 7, HEAD, checked=reason)
        self.assertEqual(self.gitea.writes, [])

    def test_a_reason_also_follows_a_draft_line_deleted_by_hand(self):
        """Scenario#139.01"""
        reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="All points hold; fixed.")

        body = self.gitea.comments[-1]["body"]
        self.assertIn("### Assessment of the findings\n\nAll points hold; fixed.", body)
        self.assertLess(body.index("### Codex"), body.index("### Assessment of the findings"))

    def test_a_missing_svg_is_refused_before_any_write(self):
        os.remove(os.path.join(self.dir, "pr-lens-flow.svg"))
        with self.assertRaisesRegex(reviewreport.AttachError, "pr-lens-flow.svg"):
            reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
        self.assertEqual(self.gitea.writes, [])


class ErrorTextTest(unittest.TestCase):
    def test_the_cause_of_a_failure_comes_first(self):
        """Scenario#131.01"""
        stderr = ("werkrevisor: operator config from /werkrevisor/config/hsadmin-ng.toml\n"
                  "werkrevisor: cannot read the PR: GET /repos/o/r/pulls/343 failed: The handshake operation timed out\n")

        text = reviewreport.error_text(stderr)
        report = reviewreport.compose(HEAD, None, None, werkrevisor_error=text, codex_error="x")

        self.assertTrue(text.startswith("werkrevisor: cannot read the PR"))
        self.assertIn("The review did not complete: werkrevisor: cannot read the PR", report)
        self.assertNotIn("operator config", report)

    def test_a_gitea_request_waits_90_seconds(self):
        """Scenario#131.02"""
        from werkrevisor.gitea import GiteaClient
        self.assertEqual(GiteaClient("https://g.example.org", "o", "r", "t")._timeout, 90)


class TitleTest(unittest.TestCase):
    URL = "https://git.example.org/team/app/pulls/343"

    def test_the_title_links_the_pr_and_keeps_the_commit(self):
        """Scenario#134.02"""
        report = reviewreport.compose(HEAD, None, None, codex_error="x", pr=(343, self.URL))

        self.assertTrue(report.startswith(f"## AI review of [PR#343]({self.URL}) (`6d1842aa`)\n"))

    def test_attach_reads_the_commit_from_the_new_and_the_old_title(self):
        """Scenario#134.02"""
        for pr in ((343, self.URL), None):
            with self.subTest(pr=pr):
                draft = reviewreport.compose(HEAD, None, None, codex_error="x", pr=pr)
                checked = "\n".join(line for line in draft.split("\n") if not line.startswith("> **Draft:**"))
                gitea = FakeGitea()

                reviewreport.attach(checked, tempfile.gettempdir(), gitea, 7, HEAD)

                self.assertEqual(len(gitea.comments), 1)
                with self.assertRaisesRegex(reviewreport.AttachError, "run hs-review again"):
                    reviewreport.attach(checked, tempfile.gettempdir(), FakeGitea(), 7, "f" * 40)