"""Actual Linux process/lock tests on a pinned standalone source module.

Usage: python lease-probe.py /path/to/data_lease.py
Fixtures are temporary SQLite databases; no agent or backend server is started.
Worker exit 7 is this probe's contention code, not the application's exit code.
"""
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import platform
import selectors
import signal
import sqlite3
import subprocess
import sys
import tempfile

PIN = "cca66e7a5028b23f352a24671e749afb6933da76"
DIGEST = "f06b113020e533f12d38c4ddde8794659204e1669d9c884ef8499fde753ce7c6"
source = Path(sys.argv[1]).resolve()
assert hashlib.sha256(source.read_bytes()).hexdigest() == DIGEST
assert sys.platform.startswith("linux"), "This replay tests Linux flock and POSIX signals"
spec = importlib.util.spec_from_file_location("lease_source", source)
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)

if len(sys.argv) > 2 and sys.argv[2] == "--worker":
    try:
        lease = module.acquire(Path(sys.argv[3]))
    except module.DataDirInUse as exc:
        print(json.dumps({"status": "in_use", "holder": exc.holder}), flush=True)
        sys.exit(7)
    try:
        print(json.dumps({"status": "held", "pid": os.getpid(),
                          "inode": os.fstat(lease._fd).st_ino}), flush=True)
        if sys.argv[4] == "hold":
            sys.stdin.readline()
    finally:
        lease.release()
    sys.exit(0)

workers = []


def args(database, mode):
    return [sys.executable, str(Path(__file__).resolve()), str(source),
            "--worker", str(database), mode]


def start(database):
    child = subprocess.Popen(args(database, "hold"), stdin=subprocess.PIPE,
                             stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    workers.append(child)
    with selectors.DefaultSelector() as selector:
        selector.register(child.stdout, selectors.EVENT_READ)
        assert selector.select(timeout=10), "worker did not report readiness"
    record = json.loads(child.stdout.readline())
    assert record["status"] == "held", record
    assert child.poll() is None
    return child, record


def once(database, expected):
    result = subprocess.run(args(database, "once"), capture_output=True, text=True, timeout=10)
    assert result.returncode == expected, (result.returncode, result.stdout, result.stderr)
    return json.loads(result.stdout)


def stop(child):
    if child.poll() is None:
        child.stdin.write("release\n")
        child.stdin.flush()
        child.wait(timeout=10)
        assert child.returncode == 0


def database(root, name):
    path = root / name
    with sqlite3.connect(path) as db:
        db.execute("CREATE TABLE fixture (id INTEGER PRIMARY KEY)")
    return path


print(f"Python {platform.python_version()}; source {PIN}; Linux flock")
try:
    with tempfile.TemporaryDirectory(prefix="duet-lease-probe-") as tmp:
        root = Path(tmp)
        db = database(root, "shared.db")
        holder, record = start(db)
        denied = once(db, 7)
        assert denied["holder"]["pid"] == holder.pid
        print("same database: contender exit=7; live holder PID reported")

        alias = root / "alias.db"
        alias.symlink_to(db)
        assert once(alias, 7)["holder"]["pid"] == holder.pid
        print("symlink spelling: contender exit=7; same resolved lease")

        other = database(root, "other.db")
        assert once(other, 0)["status"] == "held"
        print("different database in same directory: contender exit=0")

        # Metadata is diagnostic. Corrupt it without replacing the locked inode.
        lease_path = module.lease_path(db)
        lease_path.write_text("not valid JSON\n")
        denied = once(db, 7)
        assert denied["holder"] == {}
        print("corrupt holder metadata: contender exit=7; holder details empty")
        stop(holder)

        for sig in (signal.SIGTERM, signal.SIGKILL):
            holder, record = start(db)
            holder.send_signal(sig)
            holder.wait(timeout=10)
            assert holder.returncode == -int(sig)
            stale = json.loads(lease_path.read_text())
            assert stale["pid"] == holder.pid
            assert once(db, 0)["status"] == "held"
            print(f"{sig.name}: holder exit={holder.returncode}; old metadata remains; next acquire exit=0")

        first = module.acquire(db)
        second = module.acquire(db)
        assert first is second
        first.release()
        once(db, 7)
        second.release()
        once(db, 0)
        print("same-process sharing: first release keeps lock; last release frees it")

        # Controlled misuse in a temporary directory: unlink the live lease.
        old, old_record = start(db)
        old_inode = lease_path.stat().st_ino
        assert old_inode == old_record["inode"]
        lease_path.unlink()
        new, new_record = start(db)
        assert old.poll() is None and new.poll() is None
        assert new_record["inode"] != old_inode
        assert lease_path.stat().st_ino == new_record["inode"]
        assert once(db, 7)["holder"]["pid"] == new.pid
        print("unlink live lease: two live holders; distinct lease inodes; third contender exit=7")
        stop(old)
        once(db, 7)  # The old process ending does not release the new inode's lock.
        stop(new)
        once(db, 0)
        print("split-lock cleanup: stopping old holder leaves new lock held; stopping both frees it")
finally:
    for child in workers:
        if child.poll() is None:
            child.kill()
            child.wait(timeout=10)
        for stream in (child.stdin, child.stdout, child.stderr):
            stream.close()
print("PASS: contention, aliases, diagnostics, two signals, reference counts, and inode replacement")