Gitea-Tools/tests/test_issue_747_sliding_lease_ttl.py

"""Tests for the 10-minute sliding TTL on reviewer and merger PR leases (#747).

The lease ledger previously minted a fixed 120-minute expiry and derived
staleness from separate 30/60-minute activity bands. A dead session therefore
held a PR for up to two hours. These tests pin the sliding-window contract:
acquisition mints a 10-minute expiry, every heartbeat slides it forward, and an
expired lease is immediately reclaimable with no intermediate waiting tier.
"""

import sys
import unittest
from datetime import datetime, timedelta, timezone

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

import reviewer_pr_lease as leases


def _body(
    *,
    session_id: str = "session-a",
    pr_number: int = 747,
    phase: str = "claimed",
    last_activity: datetime | None = None,
    expires_at: datetime | None = None,
    ttl_minutes: int | None = None,
) -> str:
    kwargs = {}
    if ttl_minutes is not None:
        kwargs["ttl_minutes"] = ttl_minutes
    return leases.format_lease_body(
        repo="Scaled-Tech-Consulting/Gitea-Tools",
        pr_number=pr_number,
        issue_number=747,
        reviewer_identity="rev1",
        profile="prgs-reviewer",
        session_id=session_id,
        worktree="branches/review-pr747",
        phase=phase,
        candidate_head="a" * 40,
        target_branch="master",
        target_branch_sha="b" * 40,
        last_activity=last_activity,
        expires_at=expires_at,
        **kwargs,
    )


def _comment(**kwargs) -> dict:
    return {"id": 1, "body": _body(**kwargs), "user": {"login": "rev1"}}


def _minutes_ago(minutes: int) -> datetime:
    return datetime.now(timezone.utc) - timedelta(minutes=minutes)


class TestSlidingTTLConstant(unittest.TestCase):
    """AC6: one named constant per lease kind, no duplicated literals."""

    def test_ttl_is_ten_minutes(self):
        self.assertEqual(leases.LEASE_TTL_MINUTES, 10)

    def test_renewal_window_is_separately_named(self):
        self.assertEqual(leases.LEASE_RENEWAL_MINUTES, 10)


class TestAcquisitionTTL(unittest.TestCase):
    """AC1 / AC2: reviewer and merger acquisition both mint now + 10 minutes."""

    def test_acquire_mints_ten_minute_expiry(self):
        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
        lease = leases.parse_lease_comment(_body(last_activity=now))
        expires = leases._parse_timestamp(lease["expires_at"])
        self.assertEqual(expires, now + timedelta(minutes=10))

    def test_merger_acquisition_shares_the_same_window(self):
        # Merger acquisition funnels through the same lease-body formatter, so
        # the reviewer TTL is the merger TTL by construction.
        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
        lease = leases.parse_lease_comment(_body(phase="merging", last_activity=now))
        expires = leases._parse_timestamp(lease["expires_at"])
        self.assertEqual(expires, now + timedelta(minutes=10))


class TestHeartbeatSlides(unittest.TestCase):
    """AC3: a heartbeat slides expires_at to now + 10 minutes."""

    def test_heartbeat_slides_expiry_forward(self):
        acquired = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
        beat = acquired + timedelta(minutes=7)
        first = leases.parse_lease_comment(_body(last_activity=acquired))
        renewed = leases.parse_lease_comment(_body(last_activity=beat))

        first_expiry = leases._parse_timestamp(first["expires_at"])
        renewed_expiry = leases._parse_timestamp(renewed["expires_at"])

        self.assertEqual(renewed_expiry, beat + timedelta(minutes=10))
        self.assertGreater(renewed_expiry, first_expiry)

    def test_renewal_window_is_independently_tunable(self):
        # The renewal amount must not be hardwired to the acquisition TTL;
        # format_lease_body accepts an explicit window.
        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
        lease = leases.parse_lease_comment(_body(last_activity=now, ttl_minutes=3))
        expires = leases._parse_timestamp(lease["expires_at"])
        self.assertEqual(expires, now + timedelta(minutes=3))


class TestFreshnessBands(unittest.TestCase):
    """AC5: expiry is the only gate; no intermediate reclaim tier."""

    def test_fresh_lease_is_active(self):
        lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(1)))
        self.assertEqual(leases.classify_lease_freshness(lease), "active")

    def test_idle_past_half_ttl_warns_before_expiry(self):
        lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(6)))
        self.assertEqual(leases.classify_lease_freshness(lease), "stale_warning")

    def test_lease_expires_after_ten_idle_minutes(self):
        lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(11)))
        self.assertEqual(leases.classify_lease_freshness(lease), "expired")

    def test_no_separate_reclaimable_tier_remains(self):
        # The old 60-minute reclaim band sat between "stale" and "expired" and
        # blocked acquisition. Under a sliding TTL an idle lease is already
        # expired, so the tier must not reappear at any idle duration.
        for minutes in (11, 30, 65, 121, 600):
            lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(minutes)))
            self.assertEqual(
                leases.classify_lease_freshness(lease),
                "expired",
                f"idle {minutes}m should be expired, not a waiting tier",
            )


class TestExpiredLeaseIsImmediatelyReclaimable(unittest.TestCase):
    """AC5: another session takes over an expired lease with no extra wait."""

    def setUp(self):
        leases.clear_session_lease()

    def _acquire_against(self, comments: list[dict]) -> dict:
        return leases.assess_acquire_lease(
            comments,
            pr_number=747,
            reviewer_identity="rev2",
            profile="prgs-reviewer",
            session_id="session-b",
            repo="Scaled-Tech-Consulting/Gitea-Tools",
            issue_number=747,
            worktree="branches/review-pr747-b",
            candidate_head="c" * 40,
            target_branch="master",
            target_branch_sha="d" * 40,
        )

    def test_expired_foreign_lease_does_not_block_acquisition(self):
        comments = [_comment(session_id="dead-session", last_activity=_minutes_ago(11))]
        result = self._acquire_against(comments)
        self.assertTrue(result["acquire_allowed"], result["reasons"])

    def test_live_foreign_lease_still_blocks_acquisition(self):
        comments = [_comment(session_id="live-session", last_activity=_minutes_ago(2))]
        result = self._acquire_against(comments)
        self.assertFalse(result["acquire_allowed"])
        self.assertTrue(any("already has active" in r for r in result["reasons"]))


class TestRemainingTimeReporting(unittest.TestCase):
    """AC7: diagnostics can distinguish 'held and live' from 'held and dying'."""

    def test_seconds_remaining_on_live_lease(self):
        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
        lease = leases.parse_lease_comment(_body(last_activity=now))
        remaining = leases.lease_seconds_remaining(lease, now=now + timedelta(minutes=4))
        self.assertEqual(remaining, 360)

    def test_seconds_remaining_is_zero_when_expired(self):
        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
        lease = leases.parse_lease_comment(_body(last_activity=now))
        remaining = leases.lease_seconds_remaining(lease, now=now + timedelta(minutes=30))
        self.assertEqual(remaining, 0)

    def test_seconds_remaining_is_none_without_parsable_expiry(self):
        self.assertIsNone(leases.lease_seconds_remaining({"expires_at": "not-a-time"}))


class TestLegacyLeaseRows(unittest.TestCase):
    """AC9: leases minted under the old 120-minute TTL still evaluate."""

    def test_legacy_two_hour_expiry_is_honoured_until_it_passes(self):
        now = datetime.now(timezone.utc)
        legacy = leases.parse_lease_comment(
            _body(last_activity=now - timedelta(minutes=90), expires_at=now + timedelta(minutes=30))
        )
        # Still inside its originally minted window: not expired, but idle long
        # enough to warn.
        self.assertEqual(leases.classify_lease_freshness(legacy), "stale_warning")

    def test_legacy_row_past_its_own_expiry_is_expired(self):
        now = datetime.now(timezone.utc)
        legacy = leases.parse_lease_comment(
            _body(last_activity=now - timedelta(minutes=180), expires_at=now - timedelta(minutes=60))
        )
        self.assertEqual(leases.classify_lease_freshness(legacy), "expired")


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