Gitea-Tools/tests/test_reconciliation_workflow.py

"""Tests for already-landed reconciliation workflow helpers (#301)."""
import sys
import unittest
from unittest.mock import patch

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))

import reconciliation_workflow
from review_proofs import assess_reconcile_workflow_source


class TestReconciliationAssessment(unittest.TestCase):
    def test_already_landed_open_pr(self):
        with patch(
            "reconciliation_workflow.is_head_ancestor_of_ref",
            return_value=True,
        ):
            assessment = reconciliation_workflow.assess_open_pr_reconciliation(
                pr={
                    "number": 278,
                    "state": "open",
                    "title": "Fix (Closes #263)",
                    "body": "",
                    "head": {"ref": "feat/x", "sha": "abc" * 13 + "a"},
                    "base": {"ref": "master"},
                },
                project_root="/tmp/repo",
                remote="prgs",
                target_branch="master",
                target_fetch={
                    "success": True,
                    "target_ref": "prgs/master",
                    "target_branch_sha": "deadbeef",
                    "git_fetch_command": "git fetch prgs master",
                },
            )
        self.assertTrue(assessment["reconciliation_allowed"])
        self.assertEqual(
            assessment["eligibility_class"],
            reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
        )

    def test_not_landed_pr(self):
        with patch(
            "reconciliation_workflow.is_head_ancestor_of_ref",
            return_value=False,
        ):
            assessment = reconciliation_workflow.assess_open_pr_reconciliation(
                pr={
                    "number": 1,
                    "state": "open",
                    "title": "WIP",
                    "body": "",
                    "head": {"ref": "feat/y", "sha": "fff" * 13 + "f"},
                    "base": {"ref": "master"},
                },
                project_root="/tmp/repo",
                remote="prgs",
                target_branch="master",
                target_fetch={
                    "success": True,
                    "target_ref": "prgs/master",
                    "target_branch_sha": "deadbeef",
                },
            )
        self.assertFalse(assessment["reconciliation_allowed"])


class TestReconciliationPlan(unittest.TestCase):
    def test_full_close_when_close_pr_allowed(self):
        plan = reconciliation_workflow.resolve_reconciliation_plan(
            assessment={
                "reconciliation_allowed": True,
                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
            },
            capabilities={
                "close_pr": True,
                "comment_pr": True,
                "close_issue": True,
                "comment_issue": True,
            },
        )
        self.assertEqual(
            plan["outcome"],
            reconciliation_workflow.OUTCOME_FULL_RECONCILE,
        )
        self.assertTrue(plan["close_pr_allowed"])
        self.assertFalse(plan["review_merge_allowed"])

    def test_comment_then_stop_without_close(self):
        plan = reconciliation_workflow.resolve_reconciliation_plan(
            assessment={
                "reconciliation_allowed": True,
                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
            },
            capabilities={
                "close_pr": False,
                "comment_pr": True,
            },
        )
        self.assertEqual(
            plan["outcome"],
            reconciliation_workflow.OUTCOME_PARTIAL_COMMENT,
        )

    def test_recovery_handoff_without_comment_or_close(self):
        plan = reconciliation_workflow.resolve_reconciliation_plan(
            assessment={
                "reconciliation_allowed": True,
                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
            },
            capabilities={"close_pr": False, "comment_pr": False},
        )
        self.assertEqual(
            plan["outcome"],
            reconciliation_workflow.OUTCOME_RECOVERY_HANDOFF,
        )

    def test_not_landed_no_action(self):
        plan = reconciliation_workflow.resolve_reconciliation_plan(
            assessment={
                "reconciliation_allowed": False,
                "eligibility_class": reconciliation_workflow.ELIGIBILITY_NOT_LANDED,
                "reasons": ["not ancestor"],
            },
            capabilities={"close_pr": True},
        )
        self.assertEqual(
            plan["outcome"],
            reconciliation_workflow.OUTCOME_NOT_LANDED,
        )


class TestReconcileWorkflowSource(unittest.TestCase):
    def test_valid_report_passes(self):
        report = (
            "Task mode: reconcile-landed-pr\n"
            "Workflow source: workflows/reconcile-landed-pr.md\n"
        )
        result = assess_reconcile_workflow_source(report)
        self.assertTrue(result["complete"])

    def test_missing_workflow_fails(self):
        result = assess_reconcile_workflow_source("Task mode: reconcile-landed-pr")
        self.assertFalse(result["complete"])


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