"""SQLite retry-key constraint fixture. No network calls or external effects.
Run with Python 3.9+: python3 agent-idempotency-null-key.py [--test]
This tests admission constraints, not exactly-once remote execution.
"""
import sqlite3
import sys
import unittest

WEAK = """CREATE TABLE operations (
    tenant TEXT, operation_id TEXT, payload TEXT NOT NULL,
    UNIQUE (tenant, operation_id)
)"""
FIXED = """CREATE TABLE operations (
    tenant TEXT NOT NULL CHECK (length(trim(tenant)) > 0),
    operation_id TEXT NOT NULL CHECK (length(trim(operation_id)) > 0),
    payload TEXT NOT NULL,
    UNIQUE (tenant, operation_id)
)"""


def database(schema):
    db = sqlite3.connect(":memory:")
    db.execute(schema)
    return db


def insert(db, tenant, operation_id, payload="same-request"):
    db.execute("INSERT INTO operations VALUES (?, ?, ?)",
               (tenant, operation_id, payload))
    db.commit()


def require_key(value):
    # An application-owned ASCII identifier contract; no coercion or trimming.
    if (type(value) is not str or not 1 <= len(value) <= 128
            or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-"
                   for c in value)):
        raise ValueError("key must be 1-128 ASCII letters, digits, underscores or hyphens")
    return value


def admit(db, tenant, operation_id, payload="same-request"):
    insert(db, require_key(tenant), require_key(operation_id), payload)


def count(db):
    return db.execute("SELECT count(*) FROM operations").fetchone()[0]


def demo():
    weak = database(WEAK)
    fixed = database(FIXED)
    try:
        insert(weak, "tenant-a", None)
        insert(weak, "tenant-a", None)
        print(f"nullable_key_rows={count(weak)}")
        try:
            insert(fixed, "tenant-a", None)
        except sqlite3.IntegrityError:
            print("fixed_missing_key=rejected")
        admit(fixed, "tenant-a", "op-1")
        try:
            admit(fixed, "tenant-a", "op-1")
        except sqlite3.IntegrityError:
            print("fixed_duplicate_key=rejected")
        print(f"fixed_rows={count(fixed)}")
    finally:
        weak.close()
        fixed.close()


class Tests(unittest.TestCase):
    def make(self, schema=FIXED):
        db = database(schema)
        self.addCleanup(db.close)
        return db

    def test_nullable_operation_id_allows_duplicates(self):
        db = self.make(WEAK)
        insert(db, "tenant-a", None)
        insert(db, "tenant-a", None)
        self.assertEqual(count(db), 2)

    def test_nullable_tenant_allows_duplicates(self):
        db = self.make(WEAK)
        insert(db, None, "op-1")
        insert(db, None, "op-1")
        self.assertEqual(count(db), 2)

    def test_ordinary_text_primary_key_allows_null(self):
        db = self.make("CREATE TABLE operations (operation_id TEXT PRIMARY KEY)")
        db.execute("INSERT INTO operations VALUES (NULL)")
        db.execute("INSERT INTO operations VALUES (NULL)")
        self.assertEqual(count(db), 2)

    def test_check_alone_does_not_reject_null(self):
        db = self.make("CREATE TABLE operations (operation_id TEXT CHECK(length(operation_id)>0))")
        db.execute("INSERT INTO operations VALUES (NULL)")
        self.assertEqual(count(db), 1)

    def test_fixed_rejects_null_in_each_key_column(self):
        for tenant, operation_id in [(None, "op-1"), ("tenant-a", None), (None, None)]:
            with self.subTest(tenant=tenant, operation_id=operation_id):
                db = self.make()
                with self.assertRaises(sqlite3.IntegrityError):
                    insert(db, tenant, operation_id)
                self.assertEqual(count(db), 0)

    def test_fixed_rejects_empty_and_space_only_keys(self):
        for bad in ["", " ", "   "]:
            for tenant, operation_id in [(bad, "op-1"), ("tenant-a", bad)]:
                db = self.make()
                with self.assertRaises(sqlite3.IntegrityError):
                    insert(db, tenant, operation_id)

    def test_valid_retry_is_blocked(self):
        db = self.make()
        admit(db, "tenant-a", "op-1")
        with self.assertRaises(sqlite3.IntegrityError):
            admit(db, "tenant-a", "op-1")
        self.assertEqual(count(db), 1)

    def test_tenants_have_separate_namespaces(self):
        db = self.make()
        admit(db, "tenant-a", "op-1")
        admit(db, "tenant-b", "op-1")
        self.assertEqual(count(db), 2)

    def test_changed_payload_cannot_replace_existing_request(self):
        db = self.make()
        admit(db, "tenant-a", "op-1", "original")
        with self.assertRaises(sqlite3.IntegrityError):
            admit(db, "tenant-a", "op-1", "changed")
        self.assertEqual(db.execute("SELECT payload FROM operations").fetchone()[0], "original")

    def test_application_rejects_noncanonical_keys(self):
        for bad in [None, True, 7, b"op-1", "", " op-1", "op-1 ", "\t", "\n", "é", "x" * 129]:
            with self.subTest(bad=bad):
                with self.assertRaises(ValueError):
                    require_key(bad)
        self.assertEqual(require_key("Op_1-a"), "Op_1-a")

    def test_ignore_is_not_successful_admission(self):
        db = self.make()
        cursor = db.execute("INSERT OR IGNORE INTO operations VALUES (?, ?, ?)",
                            ("tenant-a", None, "request"))
        self.assertEqual(cursor.rowcount, 0)
        self.assertEqual(count(db), 0)

    def test_replace_overwrites_request(self):
        db = self.make()
        admit(db, "tenant-a", "op-1", "original")
        db.execute("INSERT OR REPLACE INTO operations VALUES (?, ?, ?)",
                   ("tenant-a", "op-1", "changed"))
        self.assertEqual(count(db), 1)
        self.assertEqual(db.execute("SELECT payload FROM operations").fetchone()[0], "changed")


if __name__ == "__main__":
    if sys.argv[1:] == ["--test"]:
        unittest.main(argv=[sys.argv[0]], verbosity=2)
    elif not sys.argv[1:]:
        demo()
    else:
        raise SystemExit("usage: agent-idempotency-null-key.py [--test]")