"""Teaching fixture: fence stale MCP workers at the mutation boundary.
Run: python3 stale-worker-fence.py
No network calls. No files changed. In-memory state is not a production lock.
"""
from dataclasses import dataclass
from threading import Lock
import unittest


@dataclass(frozen=True)
class Lease:
    resource: str
    worker: str
    epoch: int
    expires_at: int


class Resource:
    def __init__(self):
        self._lock = Lock()
        self._epoch = 0
        self._lease = None
        self.value = "initial"

    def acquire(self, worker, now, ttl=10):
        # 'now' models an authoritative service clock, never a caller clock.
        with self._lock:
            if ttl <= 0:
                raise ValueError("positive TTL required")
            if self._lease and now < self._lease.expires_at:
                raise PermissionError("lease still held")
            self._epoch += 1
            self._lease = Lease("staging-route", worker, self._epoch, now + ttl)
            return self._lease

    def revoke(self):
        with self._lock:
            self._epoch += 1
            self._lease = None

    def write(self, lease, value, now):
        # Validate and mutate under the SAME lock. There is no remote write here.
        with self._lock:
            if (self._lease is None or lease != self._lease
                    or lease.epoch != self._epoch or now >= lease.expires_at):
                raise PermissionError("expired, revoked, or superseded lease")
            self.value = value
            return {"resource": lease.resource, "epoch": lease.epoch,
                    "worker": lease.worker, "outcome": "committed"}


class FenceTests(unittest.TestCase):
    def setUp(self):
        self.r = Resource()
        self.a = self.r.acquire("worker-A", now=0)

    def test_current_holder_writes(self):
        receipt = self.r.write(self.a, "A", now=1)
        self.assertEqual(receipt["epoch"], 1)
        self.assertEqual(self.r.value, "A")

    def test_live_lease_cannot_be_taken(self):
        with self.assertRaises(PermissionError):
            self.r.acquire("worker-B", now=9)

    def test_expiry_blocks_without_replacement(self):
        with self.assertRaises(PermissionError):
            self.r.write(self.a, "late", now=10)
        self.assertEqual(self.r.value, "initial")

    def test_old_worker_cannot_overwrite_successor(self):
        b = self.r.acquire("worker-B", now=10)
        self.r.write(b, "B", now=11)
        with self.assertRaises(PermissionError):
            self.r.write(self.a, "late A", now=12)
        self.assertEqual(self.r.value, "B")
        self.assertGreater(b.epoch, self.a.epoch)

    def test_takeover_fences_before_successors_first_write(self):
        self.r.acquire("worker-B", now=10)
        with self.assertRaises(PermissionError):
            self.r.write(self.a, "late A", now=11)
        self.assertEqual(self.r.value, "initial")

    def test_revocation_blocks_immediately(self):
        self.r.revoke()
        with self.assertRaises(PermissionError):
            self.r.write(self.a, "revoked", now=1)

    def test_wrong_resource_rejected(self):
        wrong = Lease("production-route", self.a.worker, self.a.epoch, 10)
        with self.assertRaises(PermissionError):
            self.r.write(wrong, "wrong target", now=1)

    def test_forged_future_epoch_rejected(self):
        forged = Lease(self.a.resource, self.a.worker, 1000, 1000)
        with self.assertRaises(PermissionError):
            self.r.write(forged, "forged", now=1)

    def test_wrong_worker_rejected(self):
        wrong = Lease(self.a.resource, "worker-B", self.a.epoch, 10)
        with self.assertRaises(PermissionError):
            self.r.write(wrong, "wrong identity", now=1)

    def test_reacquisition_uses_new_epoch(self):
        self.r.revoke()
        fresh = self.r.acquire("worker-A", now=1)
        self.assertGreater(fresh.epoch, self.a.epoch)
        self.r.write(fresh, "fresh", now=2)
        with self.assertRaises(PermissionError):
            self.r.write(self.a, "old", now=2)


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