|
| 1 | +"""CloudflareTunnel — ephemeral `--public-tunnel` helper. |
| 2 | +
|
| 3 | +These pin two contracts that broke (or could silently break) in messages |
| 4 | +the user actually reads when a tunnel fails to come up: |
| 5 | +
|
| 6 | +* missing binary -> a RuntimeError pointing at the install docs |
| 7 | +* startup timeout -> a RuntimeError that names the *actual* local port so the |
| 8 | + user can copy-paste the manual command (the port was once emitted as the |
| 9 | + literal text ``{self.local_port}`` because the line wasn't an f-string). |
| 10 | +""" |
| 11 | + |
| 12 | +import shutil |
| 13 | +import stat |
| 14 | +import sys |
| 15 | + |
| 16 | +import pytest |
| 17 | + |
| 18 | +from zhub.tunnel import CloudflareTunnel |
| 19 | + |
| 20 | + |
| 21 | +def test_is_available_reflects_path(monkeypatch): |
| 22 | + monkeypatch.setattr(shutil, "which", lambda _: "/usr/bin/cloudflared") |
| 23 | + assert CloudflareTunnel.is_available() is True |
| 24 | + monkeypatch.setattr(shutil, "which", lambda _: None) |
| 25 | + assert CloudflareTunnel.is_available() is False |
| 26 | + |
| 27 | + |
| 28 | +async def test_start_without_binary_points_at_install_docs(monkeypatch): |
| 29 | + monkeypatch.setattr(shutil, "which", lambda _: None) |
| 30 | + t = CloudflareTunnel(local_port=8787) |
| 31 | + assert t.binary is None |
| 32 | + with pytest.raises(RuntimeError) as ei: |
| 33 | + await t.start() |
| 34 | + assert "cloudflared not found" in str(ei.value) |
| 35 | + assert "downloads" in str(ei.value) |
| 36 | + |
| 37 | + |
| 38 | +@pytest.fixture |
| 39 | +def fake_cloudflared(tmp_path): |
| 40 | + """A stand-in binary that never prints a trycloudflare URL, so start() times out.""" |
| 41 | + script = tmp_path / "cloudflared" |
| 42 | + script.write_text("#!/bin/sh\nwhile true; do echo 'still booting'; sleep 0.05; done\n") |
| 43 | + script.chmod(script.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) |
| 44 | + return str(script) |
| 45 | + |
| 46 | + |
| 47 | +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell stub") |
| 48 | +async def test_start_timeout_reports_real_port(fake_cloudflared): |
| 49 | + port = 54321 |
| 50 | + t = CloudflareTunnel(local_port=port, binary=fake_cloudflared) |
| 51 | + with pytest.raises(RuntimeError) as ei: |
| 52 | + await t.start(timeout=0.3) |
| 53 | + msg = str(ei.value) |
| 54 | + # The actual port must be interpolated, not the literal placeholder. |
| 55 | + assert str(port) in msg |
| 56 | + assert "{self.local_port}" not in msg |
| 57 | + # And the process must be reaped, not left running. |
| 58 | + assert t.process is None |
| 59 | + |
| 60 | + |
| 61 | +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell stub") |
| 62 | +async def test_close_is_idempotent_and_safe_when_never_started(fake_cloudflared): |
| 63 | + t = CloudflareTunnel(local_port=1234, binary=fake_cloudflared) |
| 64 | + # never started -> no process -> close is a no-op |
| 65 | + await t.close() |
| 66 | + assert t.process is None |
0 commit comments