"""Run a deterministic quota fixture with real, local SQLite transactions.

No HTTP requests, threads, cloud storage or production database are exercised.
Two connections are advanced in a deliberate order; lock contention is not tested.
"""
import json
import sqlite3
import tempfile
from pathlib import Path


def connect(path):
    connection = sqlite3.connect(path, isolation_level=None)
    connection.execute("PRAGMA foreign_keys = ON")
    return connection


def initialize(path, limit=5, used=4):
    with connect(path) as db:
        db.executescript("""
          CREATE TABLE quotas (
            tenant TEXT PRIMARY KEY, limit_value INTEGER NOT NULL,
            used INTEGER NOT NULL CHECK (used >= 0 AND used <= limit_value)
          );
          CREATE TABLE projects (
            id TEXT PRIMARY KEY, tenant TEXT NOT NULL, name TEXT NOT NULL,
            active INTEGER NOT NULL DEFAULT 1, UNIQUE (tenant, name)
          );
          CREATE TABLE operations (
            tenant TEXT NOT NULL, request_key TEXT NOT NULL,
            name TEXT NOT NULL, project_id TEXT NOT NULL,
            PRIMARY KEY (tenant, request_key)
          );
        """)
        for tenant in ("a", "b"):
            db.execute("INSERT INTO quotas VALUES (?, ?, ?)", (tenant, limit, used))
            for i in range(used):
                db.execute("INSERT INTO projects VALUES (?, ?, ?, 1)",
                           (f"{tenant}-seed-{i}", tenant, f"seed-{i}"))


def state(db, tenant="a"):
    used, limit = db.execute("SELECT used, limit_value FROM quotas WHERE tenant=?", (tenant,)).fetchone()
    active = db.execute("SELECT count(*) FROM projects WHERE tenant=? AND active=1", (tenant,)).fetchone()[0]
    return {"used": used, "limit": limit, "active_projects": active}


def create(db, tenant, request_key, name, fail_after_reservation=False):
    db.execute("BEGIN IMMEDIATE")
    try:
        prior = db.execute("SELECT name, project_id FROM operations WHERE tenant=? AND request_key=?",
                           (tenant, request_key)).fetchone()
        if prior:
            db.execute("ROLLBACK")
            return {"outcome": "replay" if prior[0] == name else "payload_conflict",
                    "project_id": prior[1]}
        reserved = db.execute("UPDATE quotas SET used=used+1 WHERE tenant=? AND used<limit_value", (tenant,))
        if reserved.rowcount != 1:
            db.execute("ROLLBACK")
            return {"outcome": "quota_denied"}
        if fail_after_reservation:
            raise RuntimeError("injected application failure")
        project_id = f"{tenant}-request-{request_key}"
        db.execute("INSERT INTO projects VALUES (?, ?, ?, 1)", (project_id, tenant, name))
        db.execute("INSERT INTO operations VALUES (?, ?, ?, ?)", (tenant, request_key, name, project_id))
        db.execute("COMMIT")
        return {"outcome": "created", "project_id": project_id}
    except (RuntimeError, sqlite3.IntegrityError) as error:
        db.execute("ROLLBACK")
        return {"outcome": "rolled_back", "error": type(error).__name__}
    except BaseException:
        db.execute("ROLLBACK")
        raise


def delete(db, tenant, project_id):
    db.execute("BEGIN IMMEDIATE")
    try:
        changed = db.execute("UPDATE projects SET active=0 WHERE id=? AND tenant=? AND active=1",
                             (project_id, tenant))
        if changed.rowcount == 1:
            db.execute("UPDATE quotas SET used=used-1 WHERE tenant=?", (tenant,))
        db.execute("COMMIT")
        return changed.rowcount
    except BaseException:
        db.execute("ROLLBACK")
        raise


def run():
    cases = []
    with tempfile.TemporaryDirectory() as temporary:
        root = Path(temporary)
        # Unsafe policy: the checks commit before either insert happens.
        path = root / "unsafe.db"
        initialize(path)
        a, b = connect(path), connect(path)
        observations = [state(a)["active_projects"], state(b)["active_projects"]]
        for db, key in ((a, "first"), (b, "second")):
            db.execute("INSERT INTO projects VALUES (?, 'a', ?, 1)", (key, key))
        final_count = state(a)["active_projects"]
        assert observations == [4, 4] and final_count == 6
        cases.append({"case": "separate_count_then_insert", "observed_counts": observations,
                      "limit": 5, "final_active_projects": final_count, "overshoot": 1})
        a.close(); b.close()

        path = root / "guarded.db"
        initialize(path)
        a, b = connect(path), connect(path)
        first, second = create(a, "a", "first", "new-project"), create(b, "a", "second", "other-project")
        assert first["outcome"] == "created" and second["outcome"] == "quota_denied"
        assert state(a) == {"used": 5, "limit": 5, "active_projects": 5}
        cases.append({"case": "conditional_reservation", "attempts": [first, second], "state": state(a)})
        replay = create(b, "a", "first", "new-project")
        assert replay == {"outcome": "replay", "project_id": first["project_id"]} and state(a)["used"] == 5
        cases.append({"case": "same_operation_replay", "result": replay, "state": state(a)})
        conflict = create(b, "a", "first", "changed-name")
        assert conflict["outcome"] == "payload_conflict" and state(a)["used"] == 5
        cases.append({"case": "same_key_changed_payload", "result": conflict, "state": state(a)})
        a.close(); b.close()

        for key, name, failure in (("rollback_after_reservation", "new-project", True),
                                   ("rollback_on_duplicate_name", "seed-0", False)):
            path = root / f"{key}.db"
            initialize(path)
            db = connect(path)
            result = create(db, "a", "failed", name, failure)
            assert result["outcome"] == "rolled_back" and state(db) == {"used": 4, "limit": 5, "active_projects": 4}
            assert db.execute("SELECT count(*) FROM operations").fetchone()[0] == 0
            cases.append({"case": key, "result": result, "state": state(db)})
            db.close()

        path = root / "release.db"
        initialize(path)
        db = connect(path)
        releases = [delete(db, "a", "a-seed-0"), delete(db, "a", "a-seed-0")]
        assert releases == [1, 0] and state(db) == {"used": 3, "limit": 5, "active_projects": 3}
        cases.append({"case": "duplicate_delete", "changed_rows": releases, "state": state(db)})
        db.close()

        path = root / "tenants.db"
        initialize(path, limit=1, used=0)
        db = connect(path)
        outcomes = [create(db, tenant, "same-key", "same-name") for tenant in ("a", "b")]
        assert all(v["outcome"] == "created" for v in outcomes)
        assert state(db, "a")["used"] == state(db, "b")["used"] == 1
        cases.append({"case": "tenant_scoped_operation", "outcomes": outcomes,
                      "states": {tenant: state(db, tenant) for tenant in ("a", "b")}})
        db.close()
    assert len(cases) == 8
    return {"method": "Deterministic, ordered SQLite transactions; no simultaneous threads or lock test",
            "summary": {"case_count": 8, "unsafe_final_active": 6, "guarded_final_active": 5,
                        "last_slot_limit": 5, "guarded_creates": 1, "guarded_denials": 1}, "cases": cases}


if __name__ == "__main__":
    results = run()
    assert results == run(), "Repeated execution must preserve the exact fixture output"
    Path(__file__).with_name("results.json").write_text(json.dumps(results, indent=2) + "\n")
    print(json.dumps(results["summary"], sort_keys=True))
    print("PASS: eight cases, each assertion and repeated-run equality")