Gitea-Tools/tests/test_reconciliation_inventory_proof.py

"""Tests for reconciliation PR-inventory pagination proof (#308).

Already-landed reconciliation reports may not claim a complete queue
scan (or that all already-landed PRs were found) unless the report
carries pagination proof fields and the per-page evidence proves the
final page was reached or the page limit was not hit.
"""
import sys
import unittest
from pathlib import Path

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

from review_proofs import assess_reconciliation_inventory_proof  # noqa: E402


def _report(claim=True, fields=True, pages="1", counts="12",
            final_proof="is_final_page=true, has_more=false"):
    lines = []
    if claim:
        lines.append(
            "Inventory scope: complete queue scan; all already-landed "
            "PRs were found."
        )
    if fields:
        lines += [
            "- Requested page size: 50",
            f"- Pages fetched: {pages}",
            f"- Returned PR count per page: {counts}",
            f"- Final page proof: {final_proof}",
        ]
    return "\n".join(lines) + "\n"


def _page(page, returned, per_page=50, has_more=False, final=True,
          inventory_complete=None):
    d = {
        "page": page,
        "per_page": per_page,
        "returned_count": returned,
        "has_more": has_more,
        "is_final_page": final,
    }
    if inventory_complete is not None:
        d["inventory_complete"] = inventory_complete
    return d


class TestCompleteScanProven(unittest.TestCase):
    def test_first_page_below_limit_passes(self):
        result = assess_reconciliation_inventory_proof(
            _report(),
            pagination_evidence=[_page(1, 12)],
        )
        self.assertTrue(result["complete"])
        self.assertEqual(result["reasons"], [])

    def test_multiple_pages_to_final_passes(self):
        report = _report(
            pages="3", counts="50, 50, 12",
            final_proof="has_more=false on page 3",
        )
        evidence = [
            _page(1, 50, has_more=True, final=False),
            _page(2, 50, has_more=True, final=False),
            _page(3, 12),
        ]
        result = assess_reconciliation_inventory_proof(
            report, pagination_evidence=evidence
        )
        self.assertTrue(result["complete"])

    def test_no_completeness_claim_needs_no_proof(self):
        result = assess_reconciliation_inventory_proof(
            _report(claim=False, fields=False),
            pagination_evidence=None,
        )
        self.assertTrue(result["complete"])


class TestUnprovenScanFails(unittest.TestCase):
    def test_first_page_exactly_at_limit_fails(self):
        # 50 returned with per_page 50 and no final-page signal: the page
        # limit was hit, so completeness is unproven.
        evidence = [_page(1, 50, has_more=True, final=False)]
        result = assess_reconciliation_inventory_proof(
            _report(counts="50", final_proof="none"),
            pagination_evidence=evidence,
        )
        self.assertFalse(result["complete"])
        self.assertTrue(
            any("final page" in r.lower() for r in result["reasons"])
        )

    def test_missing_pagination_evidence_fails(self):
        result = assess_reconciliation_inventory_proof(
            _report(), pagination_evidence=None
        )
        self.assertFalse(result["complete"])
        self.assertTrue(
            any("pagination" in r.lower() for r in result["reasons"])
        )

    def test_missing_report_fields_fails(self):
        result = assess_reconciliation_inventory_proof(
            _report(fields=False),
            pagination_evidence=[_page(1, 12)],
        )
        self.assertFalse(result["complete"])
        self.assertTrue(
            any("requested page size" in r.lower()
                for r in result["reasons"])
        )

    def test_page_count_mismatch_fails(self):
        # Report claims 1 page; evidence shows 2 were fetched.
        evidence = [
            _page(1, 50, has_more=True, final=False),
            _page(2, 3),
        ]
        result = assess_reconciliation_inventory_proof(
            _report(pages="1", counts="50"),
            pagination_evidence=evidence,
        )
        self.assertFalse(result["complete"])
        self.assertTrue(
            any("pages fetched" in r.lower() for r in result["reasons"])
        )

    def test_per_page_count_mismatch_fails(self):
        evidence = [_page(1, 12)]
        result = assess_reconciliation_inventory_proof(
            _report(counts="11"),
            pagination_evidence=evidence,
        )
        self.assertFalse(result["complete"])
        self.assertTrue(
            any("count per page" in r.lower() for r in result["reasons"])
        )

    def test_evidence_page_missing_metadata_fails(self):
        result = assess_reconciliation_inventory_proof(
            _report(),
            pagination_evidence=[{"page": 1}],
        )
        self.assertFalse(result["complete"])


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