soubarnak
ea7be5f111
The fork takes its own name everywhere a user can see it: binaries (voculinux, voculinux-gui), desktop entry, autostart, tray, D-Bus namespace (com.voculinux.Voculinux, no alias for the old name), config/data/cache dirs, single-instance lock, IBus engine identity, tray icons, resource prefixes, VOCULINUX_* environment names, and the update-checker user agent. The internal Python module stays vocalinux and the distribution metadata keeps the upstream URLs as provenance. On first run a previous Vocalinux install's config.json, custom dictionary, and portal restore token are copied once into the new locations (new config always wins, no merge, no write-back); models re-resolve through the pinned-checksum download path. An offline naming-sweep test proves no old-name strings remain on user-visible surfaces outside the allowlist. Refs #18
979 lines
38 KiB
Python
979 lines
38 KiB
Python
"""Unit tests for local VocaGateway embed helpers (no real containers)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from vocalinux.gateway_embed.pairing import (
|
|
MAX_QR_SVG_BYTES,
|
|
decode_pairing_payload,
|
|
)
|
|
from vocalinux.gateway_embed.preset import (
|
|
GATEWAY_TRANSCRIPTIONS_ENDPOINT,
|
|
remote_api_preset_from_pairing,
|
|
)
|
|
from vocalinux.gateway_embed.runtime import ContainerRuntime, detect_container_runtime
|
|
from vocalinux.gateway_embed.sandbox import detect_sandbox
|
|
from vocalinux.gateway_embed.status import GatewayStatus
|
|
from vocalinux.gateway_embed.urls import is_loopback_url, reject_loopback_url
|
|
from vocalinux.ui.config_manager import DEFAULT_CONFIG
|
|
|
|
|
|
class TestLoopbackRejection(unittest.TestCase):
|
|
def test_rejects_localhost_variants(self):
|
|
for url in (
|
|
"http://127.0.0.1:8765",
|
|
"http://localhost:8765",
|
|
"http://[::1]:8765",
|
|
"https://127.0.0.1/",
|
|
"",
|
|
None,
|
|
):
|
|
self.assertTrue(is_loopback_url(url), url)
|
|
self.assertIsNone(reject_loopback_url(url))
|
|
|
|
def test_rejects_short_form_loopback_and_localhost_star(self):
|
|
# Contract / gateway policy: 127.1 and localhost.* must never be Pairable.
|
|
for url in (
|
|
"http://127.1:8765",
|
|
"http://127.0.1:8765",
|
|
"http://localhost.localdomain:8765",
|
|
"http://localhost.foo:8765",
|
|
):
|
|
self.assertTrue(is_loopback_url(url), url)
|
|
self.assertIsNone(reject_loopback_url(url), url)
|
|
|
|
def test_rejects_link_local_for_qr(self):
|
|
for url in (
|
|
"http://169.254.10.20:8765",
|
|
"http://[fe80::1]:8765",
|
|
):
|
|
self.assertIsNone(reject_loopback_url(url), url)
|
|
|
|
def test_rejects_default_container_bridge(self):
|
|
for url in (
|
|
"http://172.17.0.1:8765",
|
|
"http://172.17.0.2:8765",
|
|
"http://10.88.0.1:8765",
|
|
):
|
|
self.assertIsNone(reject_loopback_url(url), url)
|
|
|
|
def test_accepts_lan(self):
|
|
url = "http://192.168.1.20:8765"
|
|
self.assertFalse(is_loopback_url(url))
|
|
self.assertEqual(reject_loopback_url(url), url)
|
|
|
|
|
|
class TestRuntimeDetector(unittest.TestCase):
|
|
def test_prefers_podman(self):
|
|
def lookup(name):
|
|
return {
|
|
"podman": "/usr/bin/podman",
|
|
"docker": "/usr/bin/docker",
|
|
}.get(name)
|
|
|
|
def probe(argv, **_kwargs):
|
|
return argv[0] in {"/usr/bin/podman", "/usr/bin/docker"} and argv[1] in {
|
|
"info",
|
|
"version",
|
|
"compose",
|
|
}
|
|
|
|
info = detect_container_runtime(path_lookup=lookup, probe=probe)
|
|
self.assertEqual(info.kind, ContainerRuntime.PODMAN)
|
|
self.assertEqual(info.binary, "/usr/bin/podman")
|
|
|
|
def test_falls_back_to_docker(self):
|
|
def lookup(name):
|
|
return {"docker": "/usr/bin/docker"}.get(name)
|
|
|
|
def probe(argv, **_kwargs):
|
|
return argv[0] == "/usr/bin/docker"
|
|
|
|
info = detect_container_runtime(path_lookup=lookup, probe=probe)
|
|
self.assertEqual(info.kind, ContainerRuntime.DOCKER)
|
|
|
|
def test_none_with_hint(self):
|
|
info = detect_container_runtime(path_lookup=lambda _n: None, probe=lambda *_a, **_k: False)
|
|
self.assertEqual(info.kind, ContainerRuntime.NONE)
|
|
self.assertIn("podman", info.hint.lower())
|
|
|
|
def test_engine_without_compose_is_unavailable(self):
|
|
def lookup(name):
|
|
return {"podman": "/usr/bin/podman"}.get(name)
|
|
|
|
def probe(argv, **_kwargs):
|
|
# Engine probes succeed; the compose plugin probe (argv[1]=="compose") fails.
|
|
return (
|
|
bool(argv)
|
|
and argv[0] == "/usr/bin/podman"
|
|
and len(argv) > 1
|
|
and argv[1] in {"info", "version"}
|
|
)
|
|
|
|
info = detect_container_runtime(path_lookup=lookup, probe=probe)
|
|
self.assertEqual(info.kind, ContainerRuntime.PODMAN)
|
|
self.assertEqual(info.compose_args, ())
|
|
self.assertIn("compose", info.hint.lower())
|
|
|
|
from vocalinux.gateway_embed.runner import GatewayRunner
|
|
|
|
runner = GatewayRunner(
|
|
runtime=info,
|
|
sandbox=detect_sandbox({}),
|
|
run=MagicMock(),
|
|
)
|
|
self.assertFalse(runner.available)
|
|
self.assertEqual(runner.unavailable_hint, info.hint)
|
|
self.assertNotIn("No container runtime", runner.unavailable_hint)
|
|
|
|
|
|
class TestStatusMachine(unittest.TestCase):
|
|
def test_ready_beats_pairable(self):
|
|
self.assertEqual(
|
|
GatewayStatus.from_health(live=True, ready=True, pairable=True, running=True),
|
|
GatewayStatus.READY,
|
|
)
|
|
|
|
def test_pairable_before_ready(self):
|
|
self.assertEqual(
|
|
GatewayStatus.from_health(live=True, ready=False, pairable=True, running=True),
|
|
GatewayStatus.PAIRABLE,
|
|
)
|
|
|
|
def test_live_only(self):
|
|
self.assertEqual(
|
|
GatewayStatus.from_health(live=True, ready=False, pairable=False, running=True),
|
|
GatewayStatus.LIVE,
|
|
)
|
|
|
|
def test_starting(self):
|
|
self.assertEqual(
|
|
GatewayStatus.from_health(
|
|
live=False, ready=False, pairable=False, running=True, starting=True
|
|
),
|
|
GatewayStatus.STARTING,
|
|
)
|
|
|
|
def test_stopped(self):
|
|
self.assertEqual(
|
|
GatewayStatus.from_health(live=False, ready=False, pairable=False, running=False),
|
|
GatewayStatus.STOPPED,
|
|
)
|
|
|
|
def test_error(self):
|
|
self.assertEqual(
|
|
GatewayStatus.from_health(
|
|
live=False, ready=False, pairable=False, error=True, running=False
|
|
),
|
|
GatewayStatus.ERROR,
|
|
)
|
|
|
|
|
|
class TestRemoteApiPreset(unittest.TestCase):
|
|
def test_maps_openai_path(self):
|
|
preset = remote_api_preset_from_pairing(
|
|
url="http://192.168.1.20:8765",
|
|
token="a" * 32,
|
|
)
|
|
self.assertEqual(preset["engine"], "remote_api")
|
|
self.assertEqual(preset["remote_api_url"], "http://192.168.1.20:8765")
|
|
self.assertEqual(preset["remote_api_key"], "a" * 32)
|
|
self.assertEqual(preset["remote_api_endpoint"], GATEWAY_TRANSCRIPTIONS_ENDPOINT)
|
|
|
|
def test_rejects_empty(self):
|
|
with self.assertRaises(ValueError):
|
|
remote_api_preset_from_pairing(url="", token="x" * 32)
|
|
|
|
|
|
class TestPairingDecode(unittest.TestCase):
|
|
def test_payload_object(self):
|
|
info = decode_pairing_payload(
|
|
{
|
|
"payload": {
|
|
"v": 1,
|
|
"url": "http://10.0.0.5:8765",
|
|
"token": "tokentokentokentokentokentoken12",
|
|
}
|
|
}
|
|
)
|
|
self.assertEqual(info.display_url, "http://10.0.0.5:8765")
|
|
self.assertTrue(info.pairable)
|
|
|
|
def test_payload_string_and_loopback_hidden(self):
|
|
raw = json.dumps({"v": 1, "url": "http://127.0.0.1:8765", "token": "t" * 32})
|
|
info = decode_pairing_payload({"payload": raw})
|
|
self.assertIsNone(info.display_url)
|
|
self.assertFalse(info.pairable)
|
|
|
|
def test_docker_bridge_not_pairable(self):
|
|
info = decode_pairing_payload(
|
|
{
|
|
"payload": {
|
|
"v": 1,
|
|
"url": "http://172.17.0.2:8765",
|
|
"token": "tokentokentokentokentokentoken12",
|
|
}
|
|
}
|
|
)
|
|
self.assertIsNone(info.display_url)
|
|
self.assertFalse(info.pairable)
|
|
|
|
def test_link_local_not_pairable(self):
|
|
info = decode_pairing_payload(
|
|
{
|
|
"payload": {
|
|
"v": 1,
|
|
"url": "http://169.254.1.2:8765",
|
|
"token": "tokentokentokentokentokentoken12",
|
|
}
|
|
}
|
|
)
|
|
self.assertIsNone(info.display_url)
|
|
self.assertFalse(info.pairable)
|
|
|
|
def test_short_form_loopback_not_pairable(self):
|
|
info = decode_pairing_payload(
|
|
{
|
|
"payload": {
|
|
"v": 1,
|
|
"url": "http://127.1:8765",
|
|
"token": "tokentokentokentokentokentoken12",
|
|
}
|
|
}
|
|
)
|
|
self.assertIsNone(info.display_url)
|
|
self.assertFalse(info.pairable)
|
|
|
|
def test_qr_download_capped(self):
|
|
from vocalinux.gateway_embed.pairing import _read_capped
|
|
|
|
class FakeResp:
|
|
def __init__(self, payload: bytes):
|
|
self._buf = io.BytesIO(payload)
|
|
|
|
def read(self, n=-1):
|
|
return self._buf.read(n if n is not None else -1)
|
|
|
|
huge = b"x" * (MAX_QR_SVG_BYTES + 10)
|
|
with self.assertRaises(ValueError):
|
|
_read_capped(FakeResp(huge), MAX_QR_SVG_BYTES)
|
|
|
|
|
|
class TestSandbox(unittest.TestCase):
|
|
def test_flatpak_fails_closed(self):
|
|
state = detect_sandbox({"FLATPAK_ID": "com.vocahq.Voculinux"})
|
|
self.assertTrue(state.blocked)
|
|
self.assertEqual(state.kind, "flatpak")
|
|
self.assertIn("Flatpak", state.hint)
|
|
|
|
def test_host_ok(self):
|
|
state = detect_sandbox({})
|
|
self.assertFalse(state.blocked)
|
|
|
|
|
|
class TestDefaultEngineUnchanged(unittest.TestCase):
|
|
def test_default_still_whisper_cpp(self):
|
|
self.assertEqual(DEFAULT_CONFIG["speech_recognition"]["engine"], "whisper_cpp")
|
|
self.assertIn("gateway_embed", DEFAULT_CONFIG)
|
|
self.assertFalse(DEFAULT_CONFIG["gateway_embed"]["lan_publish"])
|
|
|
|
|
|
class TestRunnerNoVolumeWipe(unittest.TestCase):
|
|
def test_stop_refuses_wipe_flag(self):
|
|
from vocalinux.gateway_embed.runner import GatewayRunner
|
|
from vocalinux.gateway_embed.runtime import RuntimeInfo
|
|
|
|
runner = GatewayRunner(
|
|
runtime=RuntimeInfo(
|
|
kind=ContainerRuntime.PODMAN,
|
|
binary="/usr/bin/podman",
|
|
compose_args=("/usr/bin/podman", "compose"),
|
|
),
|
|
sandbox=detect_sandbox({}),
|
|
run=MagicMock(),
|
|
)
|
|
with self.assertRaises(RuntimeError):
|
|
runner.stop(wipe_volumes=True)
|
|
|
|
|
|
class TestComposeCpuProfile(unittest.TestCase):
|
|
def test_start_uses_profile_cpu(self):
|
|
from vocalinux.gateway_embed.runner import GatewayRunner
|
|
from vocalinux.gateway_embed.runtime import RuntimeInfo
|
|
|
|
calls = []
|
|
|
|
def fake_run(argv, **kwargs):
|
|
calls.append(list(argv))
|
|
completed = MagicMock()
|
|
completed.returncode = 0
|
|
completed.stdout = b""
|
|
completed.stderr = b""
|
|
return completed
|
|
|
|
runner = GatewayRunner(
|
|
runtime=RuntimeInfo(
|
|
kind=ContainerRuntime.PODMAN,
|
|
binary="/usr/bin/podman",
|
|
compose_args=("/usr/bin/podman", "compose"),
|
|
),
|
|
sandbox=detect_sandbox({}),
|
|
run=fake_run,
|
|
)
|
|
with patch.object(runner, "prepare", return_value=("/tmp/gw", "t" * 32, "/tmp/.env")):
|
|
result = runner.start(lan_publish=False)
|
|
self.assertTrue(result.ok)
|
|
compose_calls = [c for c in calls if c[:2] == ["/usr/bin/podman", "compose"]]
|
|
self.assertTrue(compose_calls)
|
|
argv = compose_calls[0]
|
|
self.assertIn("--profile", argv)
|
|
self.assertEqual(argv[argv.index("--profile") + 1], "cpu")
|
|
self.assertIn("up", argv)
|
|
self.assertIn("-d", argv)
|
|
self.assertIn("gateway", argv)
|
|
|
|
|
|
class TestImagePin(unittest.TestCase):
|
|
def test_rejects_latest_override(self):
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.runner import write_env_file
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = f"{tmp}/.env"
|
|
with patch.dict("os.environ", {"VOCAGATEWAY_IMAGE": "vocagateway:latest"}):
|
|
write_env_file(token="a" * 32, lan_publish=False, path=env_path)
|
|
body = Path(env_path).read_text(encoding="utf-8")
|
|
self.assertIn("VOCAGATEWAY_IMAGE=vocagateway:v0.1.0", body)
|
|
self.assertNotIn(":latest", body)
|
|
|
|
def test_rejects_newline_image_injection(self):
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.runner import write_env_file
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = f"{tmp}/.env"
|
|
evil = "vocagateway:v0.1.0\nEVIL=1"
|
|
with patch.dict("os.environ", {"VOCAGATEWAY_IMAGE": evil}):
|
|
write_env_file(token="a" * 32, lan_publish=False, path=env_path)
|
|
body = Path(env_path).read_text(encoding="utf-8")
|
|
self.assertIn("VOCAGATEWAY_IMAGE=vocagateway:v0.1.0\n", body)
|
|
self.assertNotIn("EVIL=", body)
|
|
|
|
def test_omits_loopback_public_url(self):
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.runner import write_env_file
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = f"{tmp}/.env"
|
|
write_env_file(
|
|
token="a" * 32,
|
|
lan_publish=True,
|
|
public_url="http://127.0.0.1:8765",
|
|
path=env_path,
|
|
)
|
|
body = Path(env_path).read_text(encoding="utf-8")
|
|
self.assertNotIn("VOCAGATEWAY_PUBLIC_URL=", body)
|
|
|
|
def test_read_lan_publish_from_env_matches_write(self):
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.runner import (
|
|
read_lan_publish_from_env,
|
|
write_env_file,
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = f"{tmp}/.env"
|
|
write_env_file(token="a" * 32, lan_publish=True, path=env_path)
|
|
self.assertTrue(read_lan_publish_from_env(env_path))
|
|
write_env_file(token="a" * 32, lan_publish=False, path=env_path)
|
|
self.assertFalse(read_lan_publish_from_env(env_path))
|
|
self.assertFalse(read_lan_publish_from_env(f"{tmp}/missing.env"))
|
|
|
|
|
|
class TestTokenFileCap(unittest.TestCase):
|
|
def test_oversized_token_file_regenerated(self):
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.paths_embed import ensure_token_file
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = f"{tmp}/token"
|
|
with open(path, "wb") as handle:
|
|
handle.write(b"a" * 9000)
|
|
token = ensure_token_file(path)
|
|
self.assertEqual(len(token), 64)
|
|
self.assertLess(Path(path).stat().st_size, 200)
|
|
|
|
|
|
class TestLanPublishGate(unittest.TestCase):
|
|
def _manager(self):
|
|
from vocalinux.gateway_embed.manager import GatewayEmbedManager
|
|
from vocalinux.gateway_embed.runner import GatewayRunner
|
|
from vocalinux.gateway_embed.runtime import RuntimeInfo
|
|
|
|
runner = GatewayRunner(
|
|
runtime=RuntimeInfo(
|
|
kind=ContainerRuntime.PODMAN,
|
|
binary="/usr/bin/podman",
|
|
compose_args=("/usr/bin/podman", "compose"),
|
|
),
|
|
sandbox=detect_sandbox({}),
|
|
run=MagicMock(),
|
|
)
|
|
return GatewayEmbedManager(runner=runner), runner
|
|
|
|
def test_lan_toggle_without_republish_not_pairable(self):
|
|
"""Desired LAN alone must not advertise Pairable on a loopback bind."""
|
|
from vocalinux.gateway_embed.pairing import PairingInfo
|
|
|
|
manager, runner = self._manager()
|
|
runner.managed_by_us = True
|
|
manager.lan_publish = True
|
|
manager._compose_lan_publish = False
|
|
manager._token = "t" * 32
|
|
self.assertFalse(manager._effective_lan_for_pairing())
|
|
|
|
captured = {}
|
|
|
|
def fake_fetch(base_url, token, *, public_url=None, fetch_qr=True, timeout=3.0):
|
|
captured["public_url"] = public_url
|
|
return PairingInfo(
|
|
version=1,
|
|
url="http://127.0.0.1:8765",
|
|
token=token,
|
|
display_url=None,
|
|
raw_payload={"v": 1, "url": "http://127.0.0.1:8765", "token": token},
|
|
)
|
|
|
|
with patch("vocalinux.gateway_embed.manager.probe_health") as health:
|
|
health.return_value = MagicMock(live=True, ready=False, error="")
|
|
with patch("vocalinux.gateway_embed.manager.fetch_pairing", side_effect=fake_fetch):
|
|
status = manager.refresh_status()
|
|
self.assertIsNone(captured.get("public_url"))
|
|
self.assertEqual(status, GatewayStatus.LIVE)
|
|
self.assertIsNotNone(manager.pairing)
|
|
self.assertFalse(manager.pairing.pairable)
|
|
|
|
def test_apply_lan_publish_republishes_when_managed(self):
|
|
from vocalinux.gateway_embed.pairing import PairingInfo
|
|
from vocalinux.gateway_embed.runner import RunnerResult
|
|
|
|
manager, runner = self._manager()
|
|
runner.managed_by_us = True
|
|
manager.lan_publish = False
|
|
manager._compose_lan_publish = False
|
|
manager._pairing = PairingInfo(
|
|
version=1,
|
|
url="http://127.0.0.1:8765",
|
|
token="t" * 32,
|
|
display_url=None,
|
|
raw_payload={},
|
|
)
|
|
calls = {"n": 0}
|
|
|
|
def fake_republish(*, lan_publish, public_url=None):
|
|
calls["n"] += 1
|
|
self.assertTrue(lan_publish)
|
|
return RunnerResult(ok=True, message="republished")
|
|
|
|
with patch("threading.Thread") as fake_thread:
|
|
fake_thread.return_value = MagicMock()
|
|
with patch.object(runner, "republish", side_effect=fake_republish):
|
|
with patch.object(manager, "refresh_status", return_value=GatewayStatus.LIVE):
|
|
manager.apply_lan_publish(True)
|
|
self.assertIsNone(manager.pairing)
|
|
self.assertTrue(manager._republish_started)
|
|
# Run worker synchronously (Thread.start was mocked).
|
|
manager._republish_worker()
|
|
self.assertEqual(calls["n"], 1)
|
|
self.assertTrue(manager._compose_lan_publish)
|
|
self.assertTrue(manager.lan_publish)
|
|
self.assertFalse(manager._republish_started)
|
|
|
|
def test_start_worker_snapshots_lan_if_switch_flips_mid_start(self):
|
|
from vocalinux.gateway_embed.runner import RunnerResult
|
|
|
|
manager, runner = self._manager()
|
|
manager.lan_publish = True
|
|
manager._compose_lan_publish = None
|
|
|
|
def fake_start(*, lan_publish, public_url=None):
|
|
self.assertTrue(lan_publish)
|
|
manager.lan_publish = False
|
|
runner.managed_by_us = True
|
|
return RunnerResult(ok=True, message="started")
|
|
|
|
with patch("threading.Thread") as fake_thread:
|
|
fake_thread.return_value = MagicMock()
|
|
with patch(
|
|
"vocalinux.gateway_embed.paths_embed.ensure_token_file",
|
|
return_value="t" * 32,
|
|
):
|
|
with patch.object(runner, "ensure_runtime"):
|
|
with patch.object(runner, "start", side_effect=fake_start):
|
|
with patch.object(manager, "_start_polling"):
|
|
with patch.object(manager, "refresh_status"):
|
|
manager._start_worker()
|
|
|
|
self.assertIs(manager._compose_lan_publish, True)
|
|
self.assertFalse(manager.lan_publish)
|
|
self.assertTrue(manager._republish_started)
|
|
|
|
def test_republish_worker_snapshots_lan_if_switch_flips_mid_republish(self):
|
|
from vocalinux.gateway_embed.runner import RunnerResult
|
|
|
|
manager, runner = self._manager()
|
|
runner.managed_by_us = True
|
|
manager.lan_publish = True
|
|
manager._compose_lan_publish = False
|
|
manager._republish_started = True
|
|
|
|
def fake_republish(*, lan_publish, public_url=None):
|
|
self.assertTrue(lan_publish)
|
|
manager.lan_publish = False
|
|
return RunnerResult(ok=True, message="republished")
|
|
|
|
with patch("threading.Thread") as fake_thread:
|
|
fake_thread.return_value = MagicMock()
|
|
with patch.object(runner, "republish", side_effect=fake_republish):
|
|
with patch.object(manager, "refresh_status"):
|
|
manager._republish_worker()
|
|
|
|
self.assertIs(manager._compose_lan_publish, True)
|
|
self.assertFalse(manager.lan_publish)
|
|
self.assertTrue(manager._republish_started)
|
|
|
|
def test_live_detail_warns_when_lan_off_but_compose_still_open(self):
|
|
from vocalinux.gateway_embed.pairing import PairingInfo
|
|
|
|
manager, runner = self._manager()
|
|
runner.managed_by_us = True
|
|
manager.lan_publish = False
|
|
manager._compose_lan_publish = True
|
|
manager._token = "t" * 32
|
|
|
|
def fake_fetch(base_url, token, *, public_url=None, fetch_qr=True, timeout=3.0):
|
|
return PairingInfo(
|
|
version=1,
|
|
url="http://127.0.0.1:8765",
|
|
token=token,
|
|
display_url=None,
|
|
raw_payload={"v": 1, "url": "http://127.0.0.1:8765", "token": token},
|
|
)
|
|
|
|
with patch("vocalinux.gateway_embed.manager.probe_health") as health:
|
|
health.return_value = MagicMock(live=True, ready=False, error="")
|
|
with patch("vocalinux.gateway_embed.manager.fetch_pairing", side_effect=fake_fetch):
|
|
status = manager.refresh_status()
|
|
self.assertEqual(status, GatewayStatus.LIVE)
|
|
self.assertIn("still open on the LAN", manager.status_detail)
|
|
self.assertFalse(manager._effective_lan_for_pairing())
|
|
|
|
def test_omits_bridge_public_url_from_env(self):
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.runner import write_env_file
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = f"{tmp}/.env"
|
|
write_env_file(
|
|
token="a" * 32,
|
|
lan_publish=True,
|
|
public_url="http://172.17.0.2:8765",
|
|
path=env_path,
|
|
)
|
|
body = Path(env_path).read_text(encoding="utf-8")
|
|
self.assertNotIn("VOCAGATEWAY_PUBLIC_URL=", body)
|
|
|
|
|
|
class TestOrphanComposeDetect(unittest.TestCase):
|
|
"""Leftover compose after Quit must not look Stopped."""
|
|
|
|
def _manager(self):
|
|
from vocalinux.gateway_embed.manager import GatewayEmbedManager
|
|
from vocalinux.gateway_embed.runner import GatewayRunner
|
|
from vocalinux.gateway_embed.runtime import RuntimeInfo
|
|
|
|
runner = GatewayRunner(
|
|
runtime=RuntimeInfo(
|
|
kind=ContainerRuntime.PODMAN,
|
|
binary="/usr/bin/podman",
|
|
compose_args=("/usr/bin/podman", "compose"),
|
|
),
|
|
sandbox=detect_sandbox({}),
|
|
run=MagicMock(),
|
|
)
|
|
runner.managed_by_us = False
|
|
return GatewayEmbedManager(runner=runner), runner
|
|
|
|
def test_refresh_status_does_not_stay_stopped_for_orphan(self):
|
|
manager, runner = self._manager()
|
|
self.assertFalse(manager.managed_by_us)
|
|
self.assertEqual(manager.status, GatewayStatus.STOPPED)
|
|
|
|
with patch.object(runner, "is_compose_running", return_value=True):
|
|
with patch("vocalinux.gateway_embed.manager.probe_health") as health:
|
|
health.return_value = MagicMock(live=True, ready=False, error="")
|
|
status = manager.refresh_status()
|
|
|
|
self.assertEqual(status, GatewayStatus.LIVE)
|
|
self.assertNotEqual(status, GatewayStatus.STOPPED)
|
|
self.assertFalse(manager.managed_by_us)
|
|
stoppable = {
|
|
GatewayStatus.STARTING,
|
|
GatewayStatus.LIVE,
|
|
GatewayStatus.PAIRABLE,
|
|
GatewayStatus.READY,
|
|
GatewayStatus.ERROR,
|
|
}
|
|
self.assertIn(status, stoppable)
|
|
|
|
def test_refresh_status_starting_when_compose_up_but_not_live(self):
|
|
manager, runner = self._manager()
|
|
with patch.object(runner, "is_compose_running", return_value=True):
|
|
with patch("vocalinux.gateway_embed.manager.probe_health") as health:
|
|
health.return_value = MagicMock(live=False, ready=False, error="")
|
|
status = manager.refresh_status()
|
|
self.assertEqual(status, GatewayStatus.STARTING)
|
|
self.assertNotEqual(status, GatewayStatus.STOPPED)
|
|
|
|
def test_runtime_ready_probe_adopts_orphan_and_starts_polling(self):
|
|
manager, runner = self._manager()
|
|
self.assertTrue(manager.runtime_ready)
|
|
self.assertFalse(manager._runtime_detect_started)
|
|
|
|
with patch.object(runner, "is_compose_running", return_value=True):
|
|
with patch("vocalinux.gateway_embed.manager.probe_health") as health:
|
|
health.return_value = MagicMock(live=True, ready=False, error="")
|
|
with patch.object(manager, "_start_polling") as poll:
|
|
manager.begin_runtime_detection()
|
|
|
|
self.assertEqual(manager.status, GatewayStatus.LIVE)
|
|
poll.assert_called_once()
|
|
|
|
def test_orphan_probe_loads_existing_token(self):
|
|
import tempfile
|
|
|
|
manager, runner = self._manager()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
token_path = f"{tmp}/token"
|
|
existing = "ab" * 32
|
|
with open(token_path, "w", encoding="utf-8") as handle:
|
|
handle.write(existing + "\n")
|
|
with patch(
|
|
"vocalinux.gateway_embed.paths_embed.token_file_path",
|
|
return_value=token_path,
|
|
):
|
|
with patch.object(runner, "is_compose_running", return_value=True):
|
|
with patch("vocalinux.gateway_embed.manager.probe_health") as health:
|
|
health.return_value = MagicMock(live=False, ready=False, error="")
|
|
with patch.object(manager, "_start_polling"):
|
|
manager.begin_runtime_detection()
|
|
self.assertEqual(manager._token, existing)
|
|
|
|
def test_stop_worker_stops_orphan_without_managed_by_us(self):
|
|
from vocalinux.gateway_embed.runner import RunnerResult
|
|
|
|
manager, runner = self._manager()
|
|
self.assertFalse(manager.managed_by_us)
|
|
|
|
with patch.object(
|
|
runner, "stop", return_value=RunnerResult(ok=True, message="stopped")
|
|
) as stop:
|
|
manager._stop_worker()
|
|
|
|
stop.assert_called_once_with(wipe_volumes=False)
|
|
self.assertEqual(manager.status, GatewayStatus.STOPPED)
|
|
self.assertFalse(manager.managed_by_us)
|
|
|
|
def test_stop_async_starts_worker_for_orphan(self):
|
|
manager, runner = self._manager()
|
|
self.assertFalse(manager.managed_by_us)
|
|
with patch("threading.Thread") as fake_thread:
|
|
started = MagicMock()
|
|
fake_thread.return_value = started
|
|
manager.stop_async()
|
|
fake_thread.assert_called_once()
|
|
kwargs = fake_thread.call_args.kwargs
|
|
self.assertEqual(kwargs.get("name"), "voculinux-gateway-stop")
|
|
self.assertEqual(kwargs.get("target").__func__, manager._stop_worker.__func__)
|
|
started.start.assert_called_once()
|
|
|
|
def test_orphan_adopt_restores_lan_on_from_env(self):
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.runner import write_env_file
|
|
|
|
manager, runner = self._manager()
|
|
self.assertFalse(runner.managed_by_us)
|
|
manager.lan_publish = False
|
|
manager._compose_lan_publish = None
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = f"{tmp}/.env"
|
|
write_env_file(token="a" * 32, lan_publish=True, path=env_path)
|
|
with patch(
|
|
"vocalinux.gateway_embed.runner.env_file_path",
|
|
return_value=env_path,
|
|
):
|
|
with patch.object(runner, "is_compose_running", return_value=True):
|
|
with patch("vocalinux.gateway_embed.manager.probe_health") as health:
|
|
health.return_value = MagicMock(live=True, ready=False, error="")
|
|
with patch.object(manager, "_start_polling"):
|
|
manager.begin_runtime_detection()
|
|
|
|
self.assertIs(manager._compose_lan_publish, True)
|
|
self.assertTrue(manager.lan_publish)
|
|
self.assertTrue(runner.managed_by_us)
|
|
self.assertTrue(manager.managed_by_us)
|
|
|
|
def test_orphan_adopt_restores_lan_off_from_loopback_env(self):
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.runner import write_env_file
|
|
|
|
manager, runner = self._manager()
|
|
manager.lan_publish = True
|
|
manager._compose_lan_publish = None
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = f"{tmp}/.env"
|
|
write_env_file(token="a" * 32, lan_publish=False, path=env_path)
|
|
with patch(
|
|
"vocalinux.gateway_embed.runner.env_file_path",
|
|
return_value=env_path,
|
|
):
|
|
with patch.object(runner, "is_compose_running", return_value=True):
|
|
with patch("vocalinux.gateway_embed.manager.probe_health") as health:
|
|
health.return_value = MagicMock(live=True, ready=False, error="")
|
|
with patch.object(manager, "_start_polling"):
|
|
manager.begin_runtime_detection()
|
|
|
|
self.assertIs(manager._compose_lan_publish, False)
|
|
self.assertFalse(manager.lan_publish)
|
|
self.assertTrue(runner.managed_by_us)
|
|
|
|
def test_apply_lan_publish_republishes_after_orphan_lan_adopt(self):
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.runner import RunnerResult, write_env_file
|
|
|
|
manager, runner = self._manager()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = f"{tmp}/.env"
|
|
write_env_file(token="a" * 32, lan_publish=True, path=env_path)
|
|
with patch(
|
|
"vocalinux.gateway_embed.runner.env_file_path",
|
|
return_value=env_path,
|
|
):
|
|
with patch.object(runner, "is_compose_running", return_value=True):
|
|
with patch("vocalinux.gateway_embed.manager.probe_health") as health:
|
|
health.return_value = MagicMock(live=True, ready=False, error="")
|
|
with patch.object(manager, "_start_polling"):
|
|
manager.begin_runtime_detection()
|
|
|
|
self.assertTrue(runner.managed_by_us)
|
|
self.assertIs(manager._compose_lan_publish, True)
|
|
|
|
calls = {"n": 0}
|
|
|
|
def fake_republish(*, lan_publish, public_url=None):
|
|
calls["n"] += 1
|
|
self.assertFalse(lan_publish)
|
|
return RunnerResult(ok=True, message="republished")
|
|
|
|
with patch("threading.Thread") as fake_thread:
|
|
fake_thread.return_value = MagicMock()
|
|
with patch.object(runner, "republish", side_effect=fake_republish):
|
|
with patch.object(manager, "refresh_status", return_value=GatewayStatus.LIVE):
|
|
manager.apply_lan_publish(False)
|
|
self.assertTrue(manager._republish_started)
|
|
manager._republish_worker()
|
|
|
|
self.assertEqual(calls["n"], 1)
|
|
self.assertIs(manager._compose_lan_publish, False)
|
|
self.assertFalse(manager.lan_publish)
|
|
|
|
|
|
class TestFailedRepublishEnvRestore(unittest.TestCase):
|
|
"""Failed force-recreate must not leave .env lying about the live bind."""
|
|
|
|
def _podman_runner(self, run):
|
|
from vocalinux.gateway_embed.runner import GatewayRunner
|
|
from vocalinux.gateway_embed.runtime import RuntimeInfo
|
|
|
|
return GatewayRunner(
|
|
runtime=RuntimeInfo(
|
|
kind=ContainerRuntime.PODMAN,
|
|
binary="/usr/bin/podman",
|
|
compose_args=("/usr/bin/podman", "compose"),
|
|
),
|
|
sandbox=detect_sandbox({}),
|
|
run=run,
|
|
)
|
|
|
|
def test_failed_republish_restores_previous_env_publish_host(self):
|
|
import os
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.runner import (
|
|
read_lan_publish_from_env,
|
|
write_env_file,
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = f"{tmp}/.env"
|
|
checkout = f"{tmp}/checkout"
|
|
os.makedirs(checkout)
|
|
Path(checkout, "compose.yaml").write_text("services: {}\n", encoding="utf-8")
|
|
write_env_file(token="a" * 32, lan_publish=True, path=env_path)
|
|
self.assertTrue(read_lan_publish_from_env(env_path))
|
|
|
|
def fake_run(argv, **_kwargs):
|
|
completed = MagicMock()
|
|
completed.returncode = 1
|
|
completed.stdout = b""
|
|
completed.stderr = b"force-recreate failed"
|
|
return completed
|
|
|
|
runner = self._podman_runner(fake_run)
|
|
runner.managed_by_us = True
|
|
with patch("vocalinux.gateway_embed.runner.env_file_path", return_value=env_path):
|
|
with patch(
|
|
"vocalinux.gateway_embed.runner.ensure_gateway_checkout",
|
|
return_value=checkout,
|
|
):
|
|
with patch(
|
|
"vocalinux.gateway_embed.runner.ensure_token_file",
|
|
return_value="a" * 32,
|
|
):
|
|
result = runner.republish(lan_publish=False)
|
|
|
|
self.assertFalse(result.ok)
|
|
self.assertTrue(read_lan_publish_from_env(env_path))
|
|
body = Path(env_path).read_text(encoding="utf-8")
|
|
self.assertIn("VOCAGATEWAY_PUBLISH_HOST=0.0.0.0", body)
|
|
self.assertNotIn("VOCAGATEWAY_PUBLISH_HOST=127.0.0.1", body)
|
|
|
|
def test_failed_republish_reverts_manager_lan_publish_to_compose(self):
|
|
from vocalinux.gateway_embed.manager import GatewayEmbedManager
|
|
from vocalinux.gateway_embed.runner import RunnerResult
|
|
|
|
runner = self._podman_runner(MagicMock())
|
|
runner.managed_by_us = True
|
|
manager = GatewayEmbedManager(runner=runner)
|
|
manager.lan_publish = False
|
|
manager._compose_lan_publish = True
|
|
manager._republish_started = True
|
|
|
|
def fake_republish(*, lan_publish, public_url=None):
|
|
self.assertFalse(lan_publish)
|
|
return RunnerResult(ok=False, message="force-recreate failed")
|
|
|
|
with patch("threading.Thread") as fake_thread:
|
|
fake_thread.return_value = MagicMock()
|
|
with patch.object(runner, "republish", side_effect=fake_republish):
|
|
manager._republish_worker()
|
|
|
|
self.assertIs(manager._compose_lan_publish, True)
|
|
self.assertTrue(manager.lan_publish)
|
|
self.assertEqual(manager.status, GatewayStatus.ERROR)
|
|
self.assertIn("previous bind is still in use", manager.status_detail)
|
|
self.assertIn("Stop and Run", manager.status_detail)
|
|
self.assertFalse(manager._republish_started)
|
|
|
|
def test_orphan_adopt_after_failed_republish_still_sees_lan_from_env(self):
|
|
import os
|
|
import tempfile
|
|
|
|
from vocalinux.gateway_embed.manager import GatewayEmbedManager
|
|
from vocalinux.gateway_embed.runner import (
|
|
read_lan_publish_from_env,
|
|
write_env_file,
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = f"{tmp}/.env"
|
|
checkout = f"{tmp}/checkout"
|
|
os.makedirs(checkout)
|
|
Path(checkout, "compose.yaml").write_text("services: {}\n", encoding="utf-8")
|
|
write_env_file(token="a" * 32, lan_publish=True, path=env_path)
|
|
|
|
def fake_run(argv, **_kwargs):
|
|
completed = MagicMock()
|
|
completed.returncode = 1
|
|
completed.stdout = b""
|
|
completed.stderr = b"force-recreate failed"
|
|
return completed
|
|
|
|
runner = self._podman_runner(fake_run)
|
|
runner.managed_by_us = True
|
|
with patch("vocalinux.gateway_embed.runner.env_file_path", return_value=env_path):
|
|
with patch(
|
|
"vocalinux.gateway_embed.runner.ensure_gateway_checkout",
|
|
return_value=checkout,
|
|
):
|
|
with patch(
|
|
"vocalinux.gateway_embed.runner.ensure_token_file",
|
|
return_value="a" * 32,
|
|
):
|
|
result = runner.republish(lan_publish=False)
|
|
self.assertFalse(result.ok)
|
|
self.assertTrue(read_lan_publish_from_env(env_path))
|
|
|
|
next_runner = self._podman_runner(MagicMock())
|
|
manager = GatewayEmbedManager(runner=next_runner)
|
|
manager.lan_publish = False
|
|
manager._compose_lan_publish = None
|
|
with patch.object(next_runner, "is_compose_running", return_value=True):
|
|
with patch("vocalinux.gateway_embed.manager.probe_health") as health:
|
|
health.return_value = MagicMock(live=True, ready=False, error="")
|
|
with patch.object(manager, "_start_polling"):
|
|
manager.begin_runtime_detection()
|
|
|
|
self.assertTrue(manager.lan_publish)
|
|
self.assertIs(manager._compose_lan_publish, True)
|
|
self.assertTrue(next_runner.managed_by_us)
|
|
|
|
|
|
class TestManagerListenerCleanup(unittest.TestCase):
|
|
def test_remove_listener_stops_callbacks(self):
|
|
"""Settings destroy must be able to detach without further emits."""
|
|
from vocalinux.gateway_embed.manager import GatewayEmbedManager
|
|
from vocalinux.gateway_embed.runner import GatewayRunner
|
|
from vocalinux.gateway_embed.runtime import RuntimeInfo
|
|
|
|
runner = GatewayRunner(
|
|
runtime=RuntimeInfo(
|
|
kind=ContainerRuntime.PODMAN,
|
|
binary="/usr/bin/podman",
|
|
compose_args=("/usr/bin/podman", "compose"),
|
|
),
|
|
sandbox=detect_sandbox({}),
|
|
run=MagicMock(),
|
|
)
|
|
manager = GatewayEmbedManager(runner=runner)
|
|
seen: list[str] = []
|
|
|
|
def listener(status, detail):
|
|
seen.append(detail)
|
|
|
|
manager.add_listener(listener)
|
|
manager._emit(manager.status, "one")
|
|
manager.remove_listener(listener)
|
|
manager._emit(manager.status, "two")
|
|
self.assertEqual(seen, ["one"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|