"""Independent contract checks for the attached benchmark's generated queue."""

import importlib
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest

sys.path.insert(0, str(Path(sys.argv.pop(1)).resolve()))
Queue = importlib.import_module("durable_queue").Queue


class Contract(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.path = str(Path(self.directory.name) / "queue.sqlite")
        self.q = Queue(self.path)

    def tearDown(self):
        self.q.close()
        self.directory.cleanup()

    def test_priority_availability_and_idempotency(self):
        low = self.q.enqueue({"nested": [1, None, True]}, key="once", priority=0)
        self.assertEqual(self.q.enqueue("replacement", key="once", priority=500), low)
        high = self.q.enqueue("high", priority=3)
        later = self.q.enqueue("later", priority=9, available_at=10)
        first = self.q.claim("worker", now=0)
        self.assertEqual(first["id"], high)
        self.assertTrue(self.q.ack(high, first["token"], now=1))
        second = self.q.claim("worker", now=1)
        self.assertEqual(second["id"], low)
        self.assertEqual(second["payload"], {"nested": [1, None, True]})
        self.assertTrue(self.q.ack(low, second["token"], now=2))
        self.assertIsNone(self.q.claim("worker", now=9))
        self.assertEqual(self.q.claim("worker", now=10)["id"], later)
        self.assertEqual(self.q.enqueue(123, key="once"), low)

    def test_tokens_expire_and_exhaustion_is_terminal(self):
        job = self.q.enqueue(None, max_attempts=2)
        first = self.q.claim("one", now=10, lease_seconds=2)
        self.assertFalse(self.q.ack(job, first["token"], now=12))
        second = self.q.claim("two", now=12, lease_seconds=2)
        self.assertEqual(second["id"], job)
        self.assertNotEqual(first["token"], second["token"])
        self.assertEqual(second["attempts"], 2)
        self.assertFalse(self.q.fail(job, first["token"], now=12))
        self.assertFalse(self.q.heartbeat(job, first["token"], now=12))
        self.assertIsNone(self.q.claim("three", now=14))
        self.assertEqual(self.q.get(job)["state"], "dead")
        self.assertEqual(
            self.q.stats(), {"ready": 0, "leased": 0, "done": 0, "dead": 1}
        )

    def test_retry_delay_heartbeat_and_reopen(self):
        job = self.q.enqueue([1, "payload"])
        lease = self.q.claim("worker", now=0, lease_seconds=10)
        self.assertTrue(self.q.heartbeat(job, lease["token"], now=9, lease_seconds=20))
        self.assertIsNone(self.q.claim("other", now=15))
        self.assertTrue(self.q.fail(job, lease["token"], now=16, delay=10))
        self.assertIsNone(self.q.claim("other", now=25))
        self.q.close()
        self.q = Queue(self.path)
        retry = self.q.claim("worker", now=26)
        self.assertEqual(retry["id"], job)
        self.assertEqual(retry["attempts"], 2)
        self.assertTrue(self.q.ack(job, retry["token"], now=27))
        self.assertFalse(self.q.ack(job, retry["token"], now=27))
        self.assertEqual(self.q.stats()["done"], 1)

    def test_expiry_reaps_all_exhausted_jobs(self):
        ids = [self.q.enqueue(index, max_attempts=1) for index in range(5)]
        for _ in ids:
            self.q.claim("worker", now=0, lease_seconds=1)
        self.assertIsNone(self.q.claim("another", now=1))
        self.assertEqual(self.q.stats()["dead"], 5)

    def test_insertion_order_breaks_ties(self):
        ids = [self.q.enqueue(value) for value in [0, False, "", [], {}]]
        for expected in ids:
            lease = self.q.claim("worker", now=0)
            self.assertEqual(lease["id"], expected)
            self.assertTrue(self.q.ack(expected, lease["token"], now=0))

    def test_validation_does_not_insert_bad_jobs(self):
        for args in [
            {"max_attempts": 0},
            {"available_at": float("nan")},
            {"available_at": float("inf")},
        ]:
            with self.assertRaises((ValueError, TypeError)):
                self.q.enqueue("invalid", **args)
        for args in [
            {"owner": "", "now": 0},
            {"owner": "a", "now": 0, "lease_seconds": 0},
        ]:
            with self.assertRaises((ValueError, TypeError)):
                self.q.claim(**args)
        self.assertEqual(sum(self.q.stats().values()), 0)

    def test_claim_survives_abrupt_process_exit(self):
        job = self.q.enqueue("survives", max_attempts=2)
        source = "import json,os,sys;from durable_queue import Queue;q=Queue(sys.argv[1]);print(json.dumps(q.claim('dead-worker',now=0,lease_seconds=1)),flush=True);os._exit(0)"
        result = subprocess.run(
            [sys.executable, "-c", source, self.path],
            cwd=sys.path[0],
            capture_output=True,
            text=True,
            check=True,
            timeout=20,
        )
        token = json.loads(result.stdout)["token"]
        lease = self.q.claim("new-worker", now=1)
        self.assertEqual(lease["id"], job)
        self.assertFalse(self.q.ack(job, token, now=1))
        self.assertTrue(self.q.ack(job, lease["token"], now=1))


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