app.py

"""Minimal bucket mount probe for a Docker SDK Space following
https://huggingface.co/docs/hub/spaces-sdks-docker#permissions (UID 1000).

Runs the probe once at import time so results appear in startup logs,
then serves the output via a tiny http.server on port 7860 so the Space
has something to show. No Gradio dependency — keeps the container minimal
and the probe behavior identical to the Gradio SDK probe Space for an
apples-to-apples comparison.
"""

from __future__ import annotations

import http.server
import os
import pwd
import socketserver
import sqlite3
import subprocess
import traceback

try:
    import fcntl
except ImportError:
    fcntl = None


def _run_probe() -> str:
    lines = []
    lines.append("=" * 60)
    lines.append("BUCKET MOUNT PROBE — Docker SDK Space (UID 1000)")
    lines.append("=" * 60)

    try:
        uid = os.getuid()
        user = pwd.getpwuid(uid).pw_name
        lines.append(f"uid={uid} user={user} euid={os.geteuid()} gid={os.getgid()}")
    except Exception as e:
        lines.append(f"uid/user lookup failed: {e}")

    lines.append(f"SYSTEM={os.environ.get('SYSTEM')!r}")
    lines.append(f"SPACE_ID={os.environ.get('SPACE_ID')!r}")
    lines.append(f"HOME={os.environ.get('HOME')!r}")

    lines.append("")
    lines.append("--- ls -lan /data ---")
    try:
        out = subprocess.run(
            ["ls", "-lan", "/data"], capture_output=True, text=True, timeout=10
        )
        lines.append(out.stdout or "(empty)")
        if out.stderr:
            lines.append(f"stderr: {out.stderr}")
    except Exception as e:
        lines.append(f"ls failed: {e}")

    lines.append("--- stat /data ---")
    try:
        out = subprocess.run(
            ["stat", "/data"], capture_output=True, text=True, timeout=10
        )
        lines.append(out.stdout or "(empty)")
    except Exception as e:
        lines.append(f"stat failed: {e}")

    lines.append("--- mount | grep /data ---")
    try:
        out = subprocess.run(
            "mount | grep /data || true",
            shell=True,
            capture_output=True,
            text=True,
            timeout=10,
        )
        lines.append(out.stdout or "(no match)")
    except Exception as e:
        lines.append(f"mount failed: {e}")

    lines.append("")
    lines.append("--- write probes ---")

    def probe(label: str, fn):
        try:
            fn()
            lines.append(f"OK   {label}")
        except Exception as e:
            lines.append(f"FAIL {label}: {type(e).__name__}: {e}")

    probe(
        "touch /data/docker_probe_touch",
        lambda: open("/data/docker_probe_touch", "w").close(),
    )

    def _sqlite_create():
        conn = sqlite3.connect("/data/docker_probe.db", timeout=5.0)
        conn.execute("CREATE TABLE IF NOT EXISTS t(x INTEGER)")
        conn.execute("INSERT INTO t VALUES (1)")
        conn.commit()
        conn.close()

    probe("sqlite3 connect + CREATE + INSERT", _sqlite_create)

    def _sqlite_delete_journal():
        conn = sqlite3.connect("/data/docker_probe_delete.db", timeout=5.0)
        conn.execute("PRAGMA journal_mode = DELETE")
        conn.execute("CREATE TABLE IF NOT EXISTS t(x INTEGER)")
        conn.commit()
        conn.close()

    probe("sqlite3 journal_mode=DELETE", _sqlite_delete_journal)

    if fcntl is not None:

        def _flock():
            f = open("/data/docker_probe.lock", "w")
            fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
            f.close()

        probe("fcntl.flock LOCK_EX|LOCK_NB", _flock)
    else:
        lines.append("SKIP fcntl.flock: fcntl not importable")

    # Control: can we write to $HOME/app (the chown'd build-time dir)?
    # This should always work — it's the "normal" writable path for UID 1000.
    lines.append("")
    lines.append("--- control: write to $HOME/app ---")
    probe(
        "touch $HOME/app/control_probe",
        lambda: open(os.path.expanduser("~/app/control_probe"), "w").close(),
    )

    lines.append("=" * 60)
    return "\n".join(lines)


try:
    STARTUP_PROBE = _run_probe()
except Exception:
    STARTUP_PROBE = "startup probe crashed:\n" + traceback.format_exc()

print(STARTUP_PROBE, flush=True)

_HTML = """<!doctype html>
<html><head><meta charset="utf-8"><title>bucket-sqlite-probe-docker</title>
<style>
  body {{ font-family: ui-monospace, SFMono-Regular, Menlo, monospace;
          max-width: 900px; margin: 2em auto; padding: 0 1em;
          background: #0b0f17; color: #dfe6f0; }}
  h1 {{ color: #ff8f6e; }}
  pre {{ background: #121826; padding: 1em; border-radius: 6px;
         overflow-x: auto; font-size: 13px; line-height: 1.45; }}
  a {{ color: #7cc4ff; }}
  .note {{ color: #9aa7ba; font-size: 14px; }}
</style>
</head><body>
<h1>bucket-sqlite-probe-docker</h1>
<p class="note">
Matched pair with
<a href="https://huggingface.co/spaces/davanstrien/bucket-sqlite-probe-gradio">
bucket-sqlite-probe-gradio</a>.
This half runs as <code>UID 1000</code> per
<a href="https://huggingface.co/docs/hub/spaces-sdks-docker#permissions">
spaces-sdks-docker#permissions</a>. Output below captured at container startup.
</p>
<pre>{body}</pre>
</body></html>
"""


class _Handler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):  # noqa: N802
        import html

        body = _HTML.format(body=html.escape(STARTUP_PROBE))
        data = body.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, fmt, *args):  # noqa: N802
        # Quiet the access log; keep stderr clean for probe output visibility.
        return


if __name__ == "__main__":
    port = int(os.environ.get("PORT", "7860"))
    print(f"Serving probe output on 0.0.0.0:{port}", flush=True)
    with socketserver.TCPServer(("0.0.0.0", port), _Handler) as srv:
        srv.serve_forever()