#!/usr/bin/env python3
"""Synthetic referential-integrity regression. No agent or external effects."""
from __future__ import annotations

import sqlite3
import sys
import unittest

SCHEMA = """
CREATE TABLE approvals (
    approval_id TEXT NOT NULL PRIMARY KEY,
    revoked INTEGER NOT NULL DEFAULT 0 CHECK (revoked IN (0, 1))
);
CREATE TABLE actions (
    action_id TEXT NOT NULL PRIMARY KEY,
    approval_id TEXT NOT NULL REFERENCES approvals(approval_id)
);
"""


def require_foreign_keys(db):
    if db.in_transaction:
        raise RuntimeError("configure enforcement before BEGIN")
    db.execute("PRAGMA foreign_keys = ON")
    if db.execute("PRAGMA foreign_keys").fetchone() != (1,):
        raise RuntimeError("foreign key enforcement unavailable")


def connect(enforce=True):
    # Explicit SQL transactions; avoid dependence on Python's implicit BEGIN.
    db = sqlite3.connect(":memory:", isolation_level=None)
    try:
        if enforce:
            require_foreign_keys(db)
        else:
            # Deliberately reproduce the failure even on default-ON builds.
            db.execute("PRAGMA foreign_keys = OFF")
            if db.execute("PRAGMA foreign_keys").fetchone() != (0,):
                raise RuntimeError("cannot reproduce enforcement-OFF case")
        db.executescript(SCHEMA)
        return db
    except BaseException:
        db.close()
        raise


def insert_action(db, approval_id: str | None = "missing"):
    db.execute("INSERT INTO actions VALUES (?, ?)", ("action-1", approval_id))


def demo():
    weak = connect(False)
    safe = connect()
    try:
        insert_action(weak)
        print("disabled_orphan_rows=" + str(weak.execute("SELECT count(*) FROM actions").fetchone()[0]))
        print("integrity_check=" + weak.execute("PRAGMA integrity_check").fetchone()[0])
        print("foreign_key_violations=" + str(len(weak.execute("PRAGMA foreign_key_check").fetchall())))
        try:
            insert_action(safe)
        except sqlite3.IntegrityError:
            print("enabled_orphan=rejected")
        else:
            raise RuntimeError("orphan unexpectedly accepted")
        print("enabled_action_rows=" + str(safe.execute("SELECT count(*) FROM actions").fetchone()[0]))
    finally:
        weak.close()
        safe.close()


class ApprovalTests(unittest.TestCase):
    def db(self, enforce=True):
        db = connect(enforce)
        self.addCleanup(db.close)
        return db

    def test_off_accepts_orphan(self):
        db = self.db(False)
        insert_action(db)
        self.assertEqual(db.execute("SELECT count(*) FROM actions").fetchone(), (1,))

    def test_on_rejects_orphan_without_row(self):
        db = self.db()
        with self.assertRaises(sqlite3.IntegrityError):
            insert_action(db)
        self.assertEqual(db.execute("SELECT count(*) FROM actions").fetchone(), (0,))

    def test_existing_parent_accepts(self):
        db = self.db()
        db.execute("INSERT INTO approvals VALUES ('approved', 0)")
        insert_action(db, "approved")
        self.assertEqual(db.execute("SELECT count(*) FROM actions").fetchone(), (1,))

    def test_null_rejected(self):
        with self.assertRaises(sqlite3.IntegrityError):
            insert_action(self.db(), None)

    def test_parent_delete_rejected(self):
        db = self.db()
        db.execute("INSERT INTO approvals VALUES ('approved', 0)")
        insert_action(db, "approved")
        with self.assertRaises(sqlite3.IntegrityError):
            db.execute("DELETE FROM approvals")

    def test_pragma_inside_transaction_is_noop(self):
        db = self.db(False)
        db.execute("BEGIN")
        db.execute("PRAGMA foreign_keys = ON")
        self.assertEqual(db.execute("PRAGMA foreign_keys").fetchone(), (0,))
        insert_action(db)
        db.execute("COMMIT")
        self.assertEqual(len(db.execute("PRAGMA foreign_key_check").fetchall()), 1)

    def test_helper_refuses_active_transaction(self):
        db = self.db(False)
        db.execute("BEGIN")
        with self.assertRaises(RuntimeError):
            require_foreign_keys(db)
        db.execute("ROLLBACK")

    def test_connections_have_separate_settings(self):
        enabled, disabled = self.db(), self.db(False)
        self.assertEqual(enabled.execute("PRAGMA foreign_keys").fetchone(), (1,))
        self.assertEqual(disabled.execute("PRAGMA foreign_keys").fetchone(), (0,))

    def test_enabling_does_not_repair_old_rows(self):
        db = self.db(False)
        insert_action(db)
        require_foreign_keys(db)
        self.assertEqual(len(db.execute("PRAGMA foreign_key_check").fetchall()), 1)

    def test_integrity_check_does_not_detect_orphan(self):
        db = self.db(False)
        insert_action(db)
        self.assertEqual(db.execute("PRAGMA integrity_check").fetchall(), [("ok",)])
        self.assertEqual(len(db.execute("PRAGMA foreign_key_check").fetchall()), 1)

    def test_revoked_parent_still_satisfies_foreign_key(self):
        db = self.db()
        db.execute("INSERT INTO approvals VALUES ('revoked', 1)")
        insert_action(db, "revoked")
        self.assertEqual(db.execute("SELECT count(*) FROM actions").fetchone(), (1,))
        self.assertEqual(db.execute("PRAGMA foreign_key_check").fetchall(), [])


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-sqlite-approval.py [--test]")