opencode-config/tests/test_tunnel_tool.py

"""Tests for .opencode/scripts/tunnel.sh — the cloudflare tunnel toggle script.

The tunnel tool (.opencode/tools/tunnel.ts) is a thin wrapper that spawns
``bash .opencode/scripts/tunnel.sh``. The toggle logic, PID-file management
and error handling all live in the bash script, so these tests exercise the
script directly (the .ts wrapper is covered by tests/test_tunnel_tool.ts).

Isolation strategy: ``tunnel.sh`` hardcodes ``PID_FILE=/tmp/tunnel.pid`` and
``LOG_FILE=/tmp/tunnel.log``. To avoid clobbering a real tunnel on the host
(or interference between tests), each test copies the script to a temp dir
and rewrites the PID/LOG paths to point inside that temp dir. ``cloudflared``
is shadowed by a fake binary on PATH that sleeps long enough for the script
to verify the process is alive (``kill -0``), so the "started" branch fires.

Covered cases (issue #64):
- Start without ``CLOUDFLARE_TUNNEL_TOKEN`` → ``❌ ... is not set``, exit 1.
- Start with token (no domain) → ``started (PID: N)``, PID file created.
- Start with token + ``TUNNEL_DOMAIN`` → ``started (PID: N, domain: <d>)``.
- Stop when process alive (PID file present, process live) → ``stopped``,
  PID file removed.
- Stale PID file (PID file present, process dead) → cleaned up, then start.
- Toggle: second start while process alive → ``stopped`` (toggle semantics).
"""

import contextlib
import os
import stat
import subprocess
import sys
import textwrap
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
SCRIPT_SRC = REPO_ROOT / ".opencode" / "scripts" / "tunnel.sh"


def _make_fake_cloudflared(tmp_path: Path) -> Path:
    """Write a fake `cloudflared` that sleeps long enough for kill -0 to pass.

    The real cloudflared runs the tunnel; for tests we only need a long-lived
    child so the script's `kill -0 $PID` check succeeds and the "started"
    branch fires. The fake ignores its args and sleeps 30s (cleaned up when
    the test process exits or the script kills it via the toggle/stop path).
    """
    fake = tmp_path / "bin" / "cloudflared"
    fake.parent.mkdir(parents=True, exist_ok=True)
    fake.write_text(
        textwrap.dedent("""\
        #!/usr/bin/env bash
        # Fake cloudflared: sleep so the parent script sees us alive (kill -0).
        exec sleep 30
        """)
    )
    fake.chmod(fake.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
    return fake


def _make_isolated_script(tmp_path: Path) -> Path:
    """Copy tunnel.sh to tmp and rewrite PID/LOG paths to live inside tmp.

    tunnel.sh hardcodes /tmp/tunnel.pid and /tmp/tunnel.log. Running the
    tests as-is would clobber a real tunnel on the host. We rewrite both
    paths to point at <tmp>/tunnel.pid and <tmp>/tunnel.log so each test is
    isolated and cleanup is automatic (tmp_path is pytest-managed).
    """
    if not SCRIPT_SRC.exists():
        pytest.skip("tunnel.sh not present")
    dst = tmp_path / "tunnel.sh"
    src = SCRIPT_SRC.read_text()
    src = src.replace('PID_FILE="/tmp/tunnel.pid"', f'PID_FILE="{tmp_path}/tunnel.pid"')
    src = src.replace('LOG_FILE="/tmp/tunnel.log"', f'LOG_FILE="{tmp_path}/tunnel.log"')
    dst.write_text(src)
    dst.chmod(dst.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
    return dst


def _run(script: Path, env: dict, timeout: float = 10.0) -> subprocess.CompletedProcess:
    """Run the (isolated) tunnel.sh with the given env and return the result."""
    return subprocess.run(
        ["bash", str(script)],
        capture_output=True,
        text=True,
        check=False,
        timeout=timeout,
        env=env,
    )


def test_start_without_token(tmp_path: Path):
    """Start without CLOUDFLARE_TUNNEL_TOKEN → error message, exit 1.

    The script checks the token first (set -euo pipefail + [[ -z ... ]]) and
    exits before touching the PID file or spawning cloudflared.
    """
    script = _make_isolated_script(tmp_path)
    env = {**os.environ}
    env.pop("CLOUDFLARE_TUNNEL_TOKEN", None)
    env.pop("TUNNEL_DOMAIN", None)
    # Ensure PATH has no real cloudflared that could shadow the check — the
    # token check fires before cloudflared is invoked, but be defensive.
    proc = _run(script, env)
    assert proc.returncode == 1, f"expected exit 1, got {proc.returncode}; stderr={proc.stderr!r}"
    assert "CLOUDFLARE_TUNNEL_TOKEN is not set" in proc.stdout, (
        f"expected token-missing message, got stdout={proc.stdout!r}"
    )
    # PID file must NOT be created on the early-exit path.
    assert not (tmp_path / "tunnel.pid").exists(), "PID file created despite missing token"


def test_start_with_token(tmp_path: Path):
    """Start with token (no domain) → 'started (PID: N)', PID file created.

    The fake cloudflared sleeps 30s so kill -0 succeeds and the script reports
    'started (PID: <N>)'. The PID file must contain the child PID.
    """
    script = _make_isolated_script(tmp_path)
    fake = _make_fake_cloudflared(tmp_path)
    base_env = {**os.environ}
    base_env.pop("TUNNEL_DOMAIN", None)
    env = {
        **base_env,
        "CLOUDFLARE_TUNNEL_TOKEN": "fake-token",
        "PATH": f"{fake.parent}:{base_env['PATH']}",
    }
    proc = _run(script, env)
    assert proc.returncode == 0, f"expected exit 0, got {proc.returncode}; stderr={proc.stderr!r}"
    assert "started (PID:" in proc.stdout, (
        f"expected 'started (PID:...)', got stdout={proc.stdout!r}"
    )
    pid_file = tmp_path / "tunnel.pid"
    assert pid_file.exists(), "PID file not created on start"
    pid = pid_file.read_text().strip()
    assert pid.isdigit(), f"PID file content is not a number: {pid!r}"
    # Clean up: kill the sleeping fake cloudflared so it doesn't linger.
    with contextlib.suppress(ProcessLookupError, PermissionError):
        os.kill(int(pid), 9)


def test_start_with_token_and_domain(tmp_path: Path):
    """Start with token + TUNNEL_DOMAIN → 'started (PID: N, domain: <d>)'.

    When TUNNEL_DOMAIN is set, the script appends ', domain: <d>' to the
    started line (display only — the tunnel itself is the same).
    """
    script = _make_isolated_script(tmp_path)
    fake = _make_fake_cloudflared(tmp_path)
    base_env = {**os.environ}
    env = {
        **base_env,
        "CLOUDFLARE_TUNNEL_TOKEN": "fake-token",
        "TUNNEL_DOMAIN": "example.com",
        "PATH": f"{fake.parent}:{base_env['PATH']}",
    }
    proc = _run(script, env)
    assert proc.returncode == 0, f"expected exit 0, got {proc.returncode}; stderr={proc.stderr!r}"
    assert "started (PID:" in proc.stdout, (
        f"expected 'started (PID:...)', got stdout={proc.stdout!r}"
    )
    assert "domain: example.com" in proc.stdout, (
        f"expected domain in output, got stdout={proc.stdout!r}"
    )
    pid = (tmp_path / "tunnel.pid").read_text().strip()
    with contextlib.suppress(ProcessLookupError, PermissionError):
        os.kill(int(pid), 9)


def test_stop_when_process_alive(tmp_path: Path):
    """Stop when PID file exists + process alive → 'stopped', PID file removed.

    Toggle semantics: first start creates the PID file, second invocation sees
    the live process and stops it. We simulate a 'live' process by writing the
    PID of the current test process (which is alive) into the PID file, then
    run the script — it should kill... wait, it would kill the test process.
    Instead we start a long-lived child (sleep) ourselves, write its PID, then
    run the script and verify it stops the child and removes the PID file.
    """
    script = _make_isolated_script(tmp_path)
    # Start a long-lived child whose PID we control.
    child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
    pid_file = tmp_path / "tunnel.pid"
    pid_file.write_text(str(child.pid))
    env = {**os.environ, "CLOUDFLARE_TUNNEL_TOKEN": "fake-token"}
    env.pop("TUNNEL_DOMAIN", None)
    proc = _run(script, env)
    assert proc.returncode == 0, f"expected exit 0, got {proc.returncode}; stderr={proc.stderr!r}"
    assert proc.stdout.strip() == "stopped", f"expected 'stopped', got stdout={proc.stdout!r}"
    assert not pid_file.exists(), "PID file not removed on stop"
    # The child must have been terminated by the script's `kill $PID`.
    child.wait(timeout=5)
    assert child.poll() is not None, "child still alive after stop"


def test_stale_pid_file_cleaned_then_start(tmp_path: Path):
    """Stale PID file (process dead) → cleaned up, then start.

    If the PID file points at a dead process, kill -0 fails and the script
    removes the stale file (rm -f) before falling through to the start path.
    We write a PID that is guaranteed dead (a recently-exited subprocess).
    """
    script = _make_isolated_script(tmp_path)
    fake = _make_fake_cloudflared(tmp_path)
    # Use a PID that is guaranteed dead: a very high PID that no real process
    # holds on this system. kill -0 fails (ESRCH) → the script treats the PID
    # file as stale, removes it, and falls through to the start path. Using a
    # recently-reaped PID risks PID reuse by the OS between reap and script
    # run; a high sentinel PID avoids that race entirely.
    dead_pid = 999999
    pid_file = tmp_path / "tunnel.pid"
    pid_file.write_text(str(dead_pid))
    base_env = {**os.environ}
    base_env.pop("TUNNEL_DOMAIN", None)
    env = {
        **base_env,
        "CLOUDFLARE_TUNNEL_TOKEN": "fake-token",
        "PATH": f"{fake.parent}:{base_env['PATH']}",
    }
    proc = _run(script, env)
    assert proc.returncode == 0, f"expected exit 0, got {proc.returncode}; stderr={proc.stderr!r}"
    assert "started (PID:" in proc.stdout, (
        f"expected start after stale cleanup, got stdout={proc.stdout!r}"
    )
    # PID file now holds the NEW cloudflared PID, not the stale one.
    new_pid = pid_file.read_text().strip()
    assert new_pid.isdigit() and int(new_pid) != dead_pid, (
        f"PID file still holds stale PID {new_pid} (expected new cloudflared PID)"
    )
    with contextlib.suppress(ProcessLookupError, PermissionError):
        os.kill(int(new_pid), 9)


def test_toggle_second_start_stops(tmp_path: Path):
    """Toggle: a second start while a process is alive → 'stopped'.

    The script is a toggle: if the PID file exists and the process is alive,
    the next invocation stops it (instead of starting a second tunnel). We
    start a tunnel first, then run the script again and expect 'stopped'.
    """
    script = _make_isolated_script(tmp_path)
    fake = _make_fake_cloudflared(tmp_path)
    base_env = {**os.environ}
    base_env.pop("TUNNEL_DOMAIN", None)
    env = {
        **base_env,
        "CLOUDFLARE_TUNNEL_TOKEN": "fake-token",
        "PATH": f"{fake.parent}:{base_env['PATH']}",
    }
    # First call: start.
    proc1 = _run(script, env)
    assert "started (PID:" in proc1.stdout, f"first call should start, got {proc1.stdout!r}"
    pid = (tmp_path / "tunnel.pid").read_text().strip()
    assert pid.isdigit(), f"PID file should hold a number after start, got {pid!r}"
    # Second call: toggle → stop.
    proc2 = _run(script, env)
    assert proc2.returncode == 0, (
        f"expected exit 0, got {proc2.returncode}; stderr={proc2.stderr!r}"
    )
    assert proc2.stdout.strip() == "stopped", (
        f"second call should stop (toggle), got stdout={proc2.stdout!r}"
    )
    assert not (tmp_path / "tunnel.pid").exists(), "PID file not removed on toggle-stop"