import unittest

from tests.fakes import BOT, FakeGitea, FakeModel, config, diff_for, gitea_file, review_data
from tests.test_flow import run
from tests.test_line_comments import APP, app_gitea
from werkrevisor import codex
from werkrevisor.verdict import Verdict

AUTOMERGE = {"yellow_gate": "approval", "human_reviewers": ["alice"], "automerge": True, "require_wip_transition": True}


OUTSIDE = "other.py"


def hint_run(gitea=None, issues=(APP,), start=40, **gate):
    data = review_data(issues=list(issues), header="[low] Naming")
    for issue in data["review"]["key_issues_to_review"]:
        issue.update(start_line=start, end_line=start + 1 if start else 0)
    gitea = gitea or app_gitea()
    outcome, _ = run(gitea, FakeModel(data=data, prompt_files=[APP]), cfg=config(gate={**AUTOMERGE, **gate}))
    return outcome, gitea


DOC = "docs/guide.md"
LONG_LINE = "A sentence that runs on and on " * 6
WRITING_BODY = "**Writing rule: line longer than 128 characters (180)** (werkrevisor writing rule)"


def writing_run(gitea=None):
    gitea = gitea or FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["Short.", LONG_LINE]))
    gitea.raw_files[DOC] = f"Short.\n{LONG_LINE}\n".encode()
    outcome, _ = run(gitea, FakeModel(prompt_files=[DOC]), cfg=config(gate=AUTOMERGE))
    return outcome, gitea


class HintsHoldTheMergeTest(unittest.TestCase):
    def test_a_writing_rule_violation_is_a_line_comment_that_holds_the_merge(self):
        outcome, gitea = writing_run()
        self.assertIs(outcome.result.verdict, Verdict.GREEN)
        self.assertEqual([r[0] for r in gitea.created_reviews], ["APPROVED"])
        [comment] = gitea.review_comments[0]
        self.assertEqual((comment["path"], comment["new_position"]), (DOC, 2))
        self.assertIn("line longer than 128 characters", comment["body"])
        self.assertEqual(gitea.scheduled_merges, [])
        self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)

    def test_a_resolved_writing_rule_comment_releases_the_merge(self):
        outcome, gitea = writing_run()
        self.assertIs(outcome.result.verdict, Verdict.GREEN)
        self.assertEqual(gitea.scheduled_merges, [])
        for comments in gitea.line_comments.values():
            for c in comments:
                c["resolver"] = {"login": "claude-code"}
        outcome, gitea = writing_run(gitea)
        self.assertEqual(outcome.result.llm_status, "reused")
        self.assertEqual(len(gitea.scheduled_merges), 1)

    def test_a_codex_line_comment_holds_no_merge(self):
        gitea = FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["Short."]))
        gitea.raw_files[DOC] = b"Short.\n"
        gitea.create_review(7, "COMMENT", "Codex review\n" + codex.COMMENT_MARKER, gitea.heads[0],
                            [{"path": DOC, "new_position": 1, "old_position": 0,
                              "body": "**[high] Lost grant** (Codex review)\n" + codex.COMMENT_MARKER}])
        outcome, _ = run(gitea, FakeModel(prompt_files=[DOC]), cfg=config(gate=AUTOMERGE))
        self.assertIs(outcome.result.verdict, Verdict.GREEN)
        self.assertEqual(len(gitea.scheduled_merges), 1)

    def test_a_refused_writing_rule_comment_still_holds_the_merge(self):
        gitea = FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["Short.", LONG_LINE]))
        gitea.refuse_line_comments = 500
        outcome, gitea = writing_run(gitea)
        self.assertEqual(gitea.scheduled_merges, [])
        self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)

    def test_an_open_writing_rule_comment_is_reused(self):
        gitea = FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["Short.", LONG_LINE]))
        gitea.reviews.append({"id": 7, "state": "APPROVED", "commit_id": "c" * 40, "user": {"login": BOT}})
        gitea.line_comments[7] = [{"id": 77, "user": {"login": BOT}, "path": DOC, "position": 2, "resolver": None,
                                   "body": WRITING_BODY, "html_url": "https://x/77"}]
        outcome, gitea = writing_run(gitea)
        self.assertEqual(gitea.review_comments, [[]])
        self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)

    def test_a_finding_never_takes_over_a_writing_rule_comment(self):
        gitea = app_gitea()
        gitea.reviews.append({"id": 7, "state": "APPROVED", "commit_id": "c" * 40, "user": {"login": BOT}})
        gitea.line_comments[7] = [{"id": 77, "user": {"login": BOT}, "path": APP, "position": 41, "resolver": None,
                                   "body": WRITING_BODY, "html_url": "https://x/77"}]
        outcome, gitea = hint_run(gitea)
        self.assertEqual(len(gitea.review_comments[0]), 1)
        self.assertEqual([c for c, _ in gitea.edited_comments if c == 77], [])

    def test_a_green_review_with_open_hints_approves_but_schedules_no_merge(self):
        gitea = app_gitea()
        gitea.scheduled_merges.append((7, "merge", False))
        outcome, gitea = hint_run(gitea)
        self.assertIs(outcome.result.verdict, Verdict.GREEN)
        self.assertEqual([r[0] for r in gitea.created_reviews], ["APPROVED"])
        self.assertEqual(len(gitea.review_comments[0]), 1)
        # The merge scheduled for an earlier head would still merge this one.
        self.assertEqual((gitea.scheduled_merges, gitea.cancelled_merges), ([], [7]))
        self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)
        self.assertIn("merge waits for 1 open hint(s)", gitea.edited_comments[-1][1])

    def test_a_hint_without_a_line_asks_for_an_answer_in_a_pr_comment(self):
        """Scenario#95.01"""
        outcome, gitea = hint_run(start=0)
        self.assertEqual(gitea.review_comments, [[]])
        self.assertEqual(gitea.scheduled_merges, [])
        self.assertIn("merge waits for 1 open hint(s): answer the 1 hint(s) that stand only in the result comment "
                      "in a PR comment, then restart the review or push a commit", gitea.edited_comments[-1][1])
        self.assertNotIn("line comment and resolve", outcome.result.gate_note)

    def test_line_comments_and_hints_without_a_line_are_named_apart(self):
        """Scenario#95.02"""
        gitea = app_gitea()
        gitea.refuse_line_comments = True
        gitea.line_comments[7] = [{"id": 77, "user": {"login": BOT}, "path": APP, "position": 2, "resolver": None,
                                   "body": WRITING_BODY, "html_url": "https://x/77"}]
        gitea.reviews.append({"id": 7, "state": "APPROVED", "commit_id": "c" * 40, "user": {"login": BOT}})
        outcome, gitea = hint_run(gitea)
        self.assertIn("merge waits for 2 open hint(s): reply to each line comment and resolve it; "
                      "answer the 1 hint(s) that stand only in the result comment in a PR comment, "
                      "then restart the review or push a commit", outcome.result.gate_note)

    def test_a_restart_releases_a_hint_without_a_line(self):
        outcome, gitea = hint_run(start=0)
        outcome, gitea = hint_run(gitea, start=0)
        self.assertEqual(outcome.result.llm_status, "reused")
        self.assertEqual(len(gitea.scheduled_merges), 1)

    def test_a_hint_outside_the_diff_holds_no_merge(self):
        """Scenario#97.01"""
        outcome, gitea = hint_run(issues=[OUTSIDE])
        self.assertIs(outcome.result.verdict, Verdict.GREEN)
        self.assertEqual(gitea.review_comments, [[]])
        self.assertIn(f"`{OUTSIDE}:40-41`", gitea.created_comments[-1])
        self.assertEqual(len(gitea.scheduled_merges), 1)

    def test_a_restarted_review_schedules_the_merge_once_all_hints_are_resolved(self):
        outcome, gitea = hint_run()
        self.assertEqual(gitea.scheduled_merges, [])
        for comments in gitea.line_comments.values():
            for c in comments:
                c["resolver"] = {"login": "claude-code"}
        outcome, gitea = hint_run(gitea)
        self.assertEqual(outcome.result.llm_status, "reused")
        self.assertEqual(len(gitea.scheduled_merges), 1)
        self.assertIn("merge scheduled", outcome.result.gate_note)
        self.assertIn("review restarted", gitea.created_comments[-1])
        self.assertIn("merge scheduled for when the checks succeed", gitea.created_comments[-1])

    def test_an_open_reply_to_a_resolved_hint_does_not_hold_the_merge(self):
        outcome, gitea = hint_run()
        [(review_id, [hint])] = gitea.line_comments.items()
        hint["resolver"] = {"login": "claude-code"}
        gitea.line_comments[review_id].append({**hint, "id": 1, "user": {"login": "claude-code"}, "resolver": None})
        outcome, gitea = hint_run(gitea)
        self.assertEqual(len(gitea.scheduled_merges), 1)

    def test_a_scheduled_merge_withdraws_the_review_request_of_an_earlier_round(self):
        gitea = app_gitea()
        gitea.requested = ["alice", "carol"]
        outcome, gitea = hint_run(gitea, hints_hold_merge=False)
        self.assertEqual(len(gitea.scheduled_merges), 1)
        # Gitea blocks a merge while an official review request is open.
        self.assertEqual(gitea.requested, ["carol"])
        self.assertIn("review request of alice withdrawn", outcome.result.gate_note)

    def test_a_request_listed_only_as_review_is_withdrawn_too(self):
        gitea = app_gitea()
        gitea.reviews.append({"id": 9, "state": "REQUEST_REVIEW", "user": {"login": "alice"}, "commit_id": ""})
        outcome, gitea = hint_run(gitea, hints_hold_merge=False)
        self.assertEqual(gitea.withdrawn, ["alice"])

    def test_a_merge_not_scheduled_keeps_the_review_request(self):
        gitea = app_gitea()
        gitea.requested = ["alice"]
        gitea.branch["required_approvals"] = 0
        outcome, gitea = hint_run(gitea, hints_hold_merge=False)
        self.assertIn("merge not scheduled", outcome.result.gate_note)
        self.assertEqual(gitea.requested, ["alice"])

    def test_a_held_merge_keeps_the_review_request(self):
        gitea = app_gitea()
        gitea.requested = ["alice"]
        outcome, gitea = hint_run(gitea)
        self.assertEqual(gitea.requested, ["alice"])

    def test_a_failed_withdrawal_is_named_and_keeps_the_review_alive(self):
        gitea = app_gitea()
        gitea.requested = ["alice"]
        gitea.fail_review_request = True
        outcome, gitea = hint_run(gitea, hints_hold_merge=False)
        self.assertEqual(outcome.result.exit_code, 0)
        self.assertIn("review request of alice not withdrawn", outcome.result.gate_note)

    def test_an_open_reply_to_an_open_hint_is_no_second_hint(self):
        outcome, gitea = hint_run()
        [(review_id, [hint])] = gitea.line_comments.items()
        gitea.line_comments[review_id].append({**hint, "id": 1, "user": {"login": "claude-code"}})
        outcome, gitea = hint_run(gitea)
        self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)

    def test_a_later_edit_shortens_the_result_once_the_line_comments_exist(self):
        gitea = app_gitea()
        edit, calls = gitea.edit_issue_comment, []

        def first_fails(comment_id, body):
            calls.append(comment_id)
            if len(calls) == 1:
                raise OSError("HTTP 502")
            return edit(comment_id, body)
        gitea.edit_issue_comment = first_fails
        outcome, gitea = hint_run(gitea)
        body = gitea.comments[-1]["body"]
        self.assertEqual(len(gitea.review_comments[0]), 1)
        self.assertIn("as a line comment", body)
        self.assertIn("merge waits for 1 open hint(s)", body)

    def test_a_restart_with_open_hints_comments_that_the_merge_still_waits(self):
        outcome, gitea = hint_run()
        outcome, gitea = hint_run(gitea)
        self.assertEqual(gitea.scheduled_merges, [])
        self.assertIn("merge waits for 1 open hint(s)", gitea.created_comments[-1])

    def test_hints_refused_as_line_comments_hold_the_merge_too(self):
        gitea = app_gitea()
        gitea.refuse_line_comments = True
        outcome, gitea = hint_run(gitea)
        self.assertIn("line_comments_refused", [r.code for r in outcome.result.reasons])
        self.assertEqual(gitea.scheduled_merges, [])
        self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)
        outcome, gitea = hint_run(gitea)
        self.assertEqual(len(gitea.scheduled_merges), 1)

    def test_a_failed_cancel_is_named(self):
        from werkrevisor.gitea import GiteaError
        gitea = app_gitea()

        def refuse(number):
            raise GiteaError("DELETE merge failed with HTTP 403", 403)
        gitea.cancel_scheduled_merge = refuse
        outcome, gitea = hint_run(gitea)
        self.assertIn("a merge scheduled earlier could not be cancelled: DELETE merge failed with HTTP 403",
                      gitea.edited_comments[-1][1])
        # The failed gate check stops the merge that Gitea still has scheduled.
        self.assertEqual(outcome.result.exit_code, 10)

    def test_unreadable_hints_hold_the_merge(self):
        gitea = app_gitea()

        def unreadable(number, review_id):
            raise OSError("HTTP 502")
        outcome, gitea = hint_run(gitea)
        for comments in gitea.line_comments.values():
            for c in comments:
                c["resolver"] = {"login": "claude-code"}
        gitea.list_review_comments = unreadable
        outcome, gitea = hint_run(gitea)
        self.assertEqual(gitea.scheduled_merges, [])
        self.assertEqual(outcome.result.exit_code, 10)
        self.assertIn("the open hints could not be read: HTTP 502", outcome.result.gate_note)

    def test_a_restart_comments_only_a_change_of_the_merge(self):
        outcome, gitea = hint_run()
        for comments in gitea.line_comments.values():
            for c in comments:
                c["resolver"] = {"login": "claude-code"}
        hint_run(gitea)
        comments = len(gitea.created_comments)
        outcome, gitea = hint_run(gitea)
        self.assertIn("merge already scheduled", outcome.result.gate_note)
        self.assertEqual(len(gitea.created_comments), comments)

    def test_the_hold_can_be_switched_off(self):
        outcome, gitea = hint_run(hints_hold_merge=False)
        self.assertIs(outcome.result.verdict, Verdict.GREEN)
        self.assertEqual(len(gitea.scheduled_merges), 1)
        self.assertEqual(gitea.cancelled_merges, [])


class HintsHoldMergeConfigTest(unittest.TestCase):
    def test_the_switch_must_be_a_boolean(self):
        from werkrevisor.config import ConfigError
        with self.assertRaisesRegex(ConfigError, r"\[gate\] hints_hold_merge must be true or false"):
            config(gate={**AUTOMERGE, "hints_hold_merge": "false"})


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