wehub-resource-sync
3e779be6f3
CI / lint (push) Failing after 13m4s
CI / test (3.11, ubuntu-latest) (push) Failing after 2m4s
CI / test (3.13, ubuntu-latest) (push) Successful in 13m30s
CI / test (3.14, ubuntu-latest) (push) Successful in 17m21s
CI / test (3.12, ubuntu-latest) (push) Successful in 17m55s
CI / discover-apps-ps (push) Successful in 1m56s
CI / test (3.9, ubuntu-latest) (push) Successful in 13m17s
CI / test (3.10, ubuntu-latest) (push) Successful in 26m21s
CI / audit (push) Successful in 13m38s
Deploy site / deploy (push) Has been cancelled
CI / test (3.14, ubuntu-24.04-arm) (push) Has been cancelled
401 lines
13 KiB
Python
401 lines
13 KiB
Python
# SPDX-License-Identifier: MIT
|
|
"""Tests for ``winpodx host-open`` CLI subcommands.
|
|
|
|
Invokes the dispatcher through ``winpodx.cli.main.main`` so the
|
|
parser + handler wiring is exercised end-to-end. XDG isolation comes
|
|
from ``conftest._isolate_xdg_and_home``; we write fake ``.desktop``
|
|
files under ``$XDG_DATA_HOME/applications`` and assert on the staged
|
|
``apps.json`` + the persisted ``winpodx.toml``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from winpodx.cli.main import cli as _cli_entry
|
|
from winpodx.core.config import Config
|
|
|
|
pytest.importorskip("PIL") # convert_to_ico needs Pillow
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_xdg_data_dirs(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
"""Pin XDG_DATA_DIRS to an empty dir so the host's /usr/share doesn't leak."""
|
|
empty = tmp_path / "_empty_xdg_dirs"
|
|
empty.mkdir()
|
|
monkeypatch.setenv("XDG_DATA_DIRS", str(empty))
|
|
monkeypatch.delenv("XDG_CURRENT_DESKTOP", raising=False)
|
|
|
|
|
|
def cli_main(argv: list[str]) -> int:
|
|
"""Invoke the top-level CLI and translate ``SystemExit`` back to ``int``.
|
|
|
|
``_dispatch`` wraps every host-open invocation in ``sys.exit(...)`` so
|
|
the dispatcher matches the convention used elsewhere (``migrate``,
|
|
``check``). Tests want a plain return code, so we trap the
|
|
``SystemExit`` here and surface its ``code`` value.
|
|
"""
|
|
try:
|
|
_cli_entry(argv)
|
|
except SystemExit as exc:
|
|
return int(exc.code or 0)
|
|
return 0
|
|
|
|
|
|
def _xdg_apps() -> Path:
|
|
base = Path(os.environ["XDG_DATA_HOME"]) / "applications"
|
|
base.mkdir(parents=True, exist_ok=True)
|
|
return base
|
|
|
|
|
|
def _write_app(name: str, body: str) -> None:
|
|
(_xdg_apps() / name).write_text(body, encoding="utf-8")
|
|
|
|
|
|
_KATE = (
|
|
"[Desktop Entry]\n"
|
|
"Type=Application\n"
|
|
"Name=Kate\n"
|
|
"Exec=/usr/bin/kate %F\n"
|
|
"Icon=kate\n"
|
|
"MimeType=text/plain;text/xml;\n"
|
|
)
|
|
_GIMP = (
|
|
"[Desktop Entry]\n"
|
|
"Type=Application\n"
|
|
"Name=GIMP\n"
|
|
"Exec=/usr/bin/gimp %F\n"
|
|
"Icon=gimp\n"
|
|
"MimeType=image/png;image/jpeg;\n"
|
|
)
|
|
|
|
|
|
# --- status -------------------------------------------------------------------
|
|
|
|
|
|
def test_status_default_enabled(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "status"])
|
|
out = capsys.readouterr().out
|
|
assert "enabled" in out
|
|
assert "0 slug(s)" in out # allowlist
|
|
assert "(none — run `winpodx host-open refresh`)" in out
|
|
|
|
|
|
def test_status_json(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "status", "--json"])
|
|
out = capsys.readouterr().out
|
|
payload = json.loads(out)
|
|
assert payload["enabled"] is True
|
|
assert payload["allowlist"] == []
|
|
assert payload["cache"]["exists"] is False
|
|
|
|
|
|
# --- enable / disable ---------------------------------------------------------
|
|
|
|
|
|
def test_enable_when_already_default_on(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "enable"])
|
|
assert "already enabled" in capsys.readouterr().out
|
|
cfg = Config.load()
|
|
assert cfg.reverse_open.enabled is True
|
|
|
|
|
|
def test_disable_then_re_enable(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "disable"])
|
|
cfg = Config.load()
|
|
assert cfg.reverse_open.enabled is False
|
|
assert "disabled" in capsys.readouterr().out
|
|
|
|
cli_main(["host-open", "enable"])
|
|
cfg = Config.load()
|
|
assert cfg.reverse_open.enabled is True
|
|
|
|
|
|
# --- add / remove -------------------------------------------------------------
|
|
|
|
|
|
def test_add_to_allowlist(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "add", "kate"])
|
|
out = capsys.readouterr().out
|
|
assert "added kate" in out
|
|
cfg = Config.load()
|
|
assert "kate" in cfg.reverse_open.allowlist
|
|
|
|
|
|
def test_add_to_denylist_with_flag(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "add", "evil-app", "--deny"])
|
|
capsys.readouterr()
|
|
cfg = Config.load()
|
|
assert "evil-app" in cfg.reverse_open.denylist
|
|
assert "evil-app" not in cfg.reverse_open.allowlist
|
|
|
|
|
|
def test_add_invalid_slug_rejected(capsys: pytest.CaptureFixture[str]) -> None:
|
|
rc = cli_main(["host-open", "add", "Bad Slug!"])
|
|
err = capsys.readouterr().err
|
|
assert rc != 0 if rc is not None else True
|
|
assert "not a valid" in err
|
|
|
|
|
|
def test_add_idempotent(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "add", "kate"])
|
|
capsys.readouterr()
|
|
cli_main(["host-open", "add", "kate"])
|
|
assert "already present" in capsys.readouterr().out
|
|
|
|
|
|
def test_add_to_allowlist_removes_from_denylist(
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
cli_main(["host-open", "add", "kate", "--deny"])
|
|
capsys.readouterr()
|
|
cli_main(["host-open", "add", "kate"])
|
|
cfg = Config.load()
|
|
assert "kate" in cfg.reverse_open.allowlist
|
|
assert "kate" not in cfg.reverse_open.denylist
|
|
|
|
|
|
def test_remove_subcommand(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "add", "kate"])
|
|
capsys.readouterr()
|
|
cli_main(["host-open", "remove", "kate"])
|
|
cfg = Config.load()
|
|
assert "kate" not in cfg.reverse_open.allowlist
|
|
|
|
|
|
def test_remove_missing_slug_is_noop(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "remove", "not-there"])
|
|
assert "not present" in capsys.readouterr().out
|
|
|
|
|
|
# --- refresh ------------------------------------------------------------------
|
|
|
|
|
|
def test_refresh_writes_manifest_and_icons(
|
|
capsys: pytest.CaptureFixture[str], tmp_path: Path
|
|
) -> None:
|
|
_write_app("kate.desktop", _KATE)
|
|
_write_app("gimp.desktop", _GIMP)
|
|
|
|
cli_main(["host-open", "refresh"])
|
|
out = capsys.readouterr().out
|
|
assert "Discovered 2 apps" in out
|
|
|
|
manifest_path = Path(os.environ["XDG_DATA_HOME"]) / "winpodx" / "reverse-open" / "apps.json"
|
|
assert manifest_path.is_file()
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
assert manifest["version"] == 1
|
|
slugs = sorted(a["slug"] for a in manifest["apps"])
|
|
assert slugs == ["gimp", "kate"]
|
|
|
|
icons_dir = manifest_path.parent / "icons"
|
|
assert (icons_dir / "kate.ico").is_file()
|
|
assert (icons_dir / "gimp.ico").is_file()
|
|
|
|
|
|
def test_refresh_json_summary(capsys: pytest.CaptureFixture[str]) -> None:
|
|
_write_app("kate.desktop", _KATE)
|
|
cli_main(["host-open", "refresh", "--json", "--skip-icons"])
|
|
out = capsys.readouterr().out
|
|
payload = json.loads(out)
|
|
assert payload["discovered"] == 1
|
|
assert payload["kept"] == 1
|
|
assert payload["skipped"] == []
|
|
# --skip-icons makes the icon counters zero (we never enter the
|
|
# conversion loop).
|
|
assert payload["icons_real"] == 0
|
|
assert payload["icons_placeholder"] == 0
|
|
|
|
|
|
def test_refresh_respects_denylist(capsys: pytest.CaptureFixture[str]) -> None:
|
|
_write_app("kate.desktop", _KATE)
|
|
_write_app("gimp.desktop", _GIMP)
|
|
cli_main(["host-open", "add", "gimp", "--deny"])
|
|
capsys.readouterr()
|
|
|
|
cli_main(["host-open", "refresh", "--json", "--skip-icons"])
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["discovered"] == 2
|
|
assert payload["kept"] == 1
|
|
assert any(s["slug"] == "gimp" and s["reason"] == "denylist" for s in payload["skipped"])
|
|
|
|
|
|
def test_refresh_respects_allowlist(capsys: pytest.CaptureFixture[str]) -> None:
|
|
_write_app("kate.desktop", _KATE)
|
|
_write_app("gimp.desktop", _GIMP)
|
|
cli_main(["host-open", "add", "kate"])
|
|
capsys.readouterr()
|
|
|
|
cli_main(["host-open", "refresh", "--json", "--skip-icons"])
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["kept"] == 1
|
|
assert any(
|
|
s["slug"] == "gimp" and s["reason"] == "not-in-allowlist" for s in payload["skipped"]
|
|
)
|
|
|
|
|
|
# --- list ---------------------------------------------------------------------
|
|
|
|
|
|
def test_list_scan(capsys: pytest.CaptureFixture[str]) -> None:
|
|
_write_app("kate.desktop", _KATE)
|
|
cli_main(["host-open", "list"])
|
|
out = capsys.readouterr().out
|
|
assert "kate" in out
|
|
|
|
|
|
def test_list_cached_missing(capsys: pytest.CaptureFixture[str]) -> None:
|
|
rc = cli_main(["host-open", "list", "--cached"])
|
|
err = capsys.readouterr().err
|
|
assert "no cached manifest" in err
|
|
assert rc != 0
|
|
|
|
|
|
def test_list_cached_reads_manifest(capsys: pytest.CaptureFixture[str]) -> None:
|
|
_write_app("kate.desktop", _KATE)
|
|
cli_main(["host-open", "refresh", "--skip-icons"])
|
|
capsys.readouterr()
|
|
cli_main(["host-open", "list", "--cached"])
|
|
out = capsys.readouterr().out
|
|
assert "Cached manifest" in out
|
|
assert "kate" in out
|
|
|
|
|
|
def test_list_json_scan(capsys: pytest.CaptureFixture[str]) -> None:
|
|
_write_app("kate.desktop", _KATE)
|
|
cli_main(["host-open", "list", "--json"])
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["source"] == "scan"
|
|
assert payload["apps"][0]["slug"] == "kate"
|
|
|
|
|
|
# --- subcommand dispatch -----------------------------------------------------
|
|
|
|
|
|
def test_no_subcommand_errors_helpfully(capsys: pytest.CaptureFixture[str]) -> None:
|
|
rc = cli_main(["host-open"])
|
|
err = capsys.readouterr().err
|
|
assert "missing subcommand" in err
|
|
assert rc != 0
|
|
|
|
|
|
# --- daemon lifecycle subcommands -------------------------------------------
|
|
|
|
|
|
def test_daemon_status_when_not_running(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "daemon-status"])
|
|
out = capsys.readouterr().out
|
|
assert "not running" in out
|
|
|
|
|
|
def test_daemon_status_json(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "daemon-status", "--json"])
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["running"] is False
|
|
assert payload["pid"] is None
|
|
assert payload["pid_file"].endswith("reverse-open.pid")
|
|
|
|
|
|
def test_stop_listener_when_not_running(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cli_main(["host-open", "stop-listener"])
|
|
out = capsys.readouterr().out
|
|
assert "not running" in out
|
|
|
|
|
|
def test_start_then_stop_then_status(capsys: pytest.CaptureFixture[str]) -> None:
|
|
_write_app("kate.desktop", _KATE)
|
|
cli_main(["host-open", "refresh", "--skip-icons"])
|
|
capsys.readouterr()
|
|
|
|
cli_main(["host-open", "start-listener"])
|
|
out = capsys.readouterr().out
|
|
assert "pid" in out
|
|
|
|
cli_main(["host-open", "daemon-status"])
|
|
assert "running (pid " in capsys.readouterr().out
|
|
|
|
cli_main(["host-open", "stop-listener"])
|
|
assert "stopped" in capsys.readouterr().out
|
|
|
|
cli_main(["host-open", "daemon-status"])
|
|
assert "not running" in capsys.readouterr().out
|
|
|
|
|
|
def test_refresh_signals_running_daemon(capsys: pytest.CaptureFixture[str]) -> None:
|
|
_write_app("kate.desktop", _KATE)
|
|
cli_main(["host-open", "refresh", "--skip-icons"])
|
|
capsys.readouterr()
|
|
cli_main(["host-open", "start-listener"])
|
|
capsys.readouterr()
|
|
|
|
try:
|
|
cli_main(["host-open", "refresh", "--skip-icons"])
|
|
out = capsys.readouterr().out
|
|
assert "SIGHUP" in out
|
|
finally:
|
|
cli_main(["host-open", "stop-listener"])
|
|
capsys.readouterr()
|
|
|
|
|
|
class TestEnsureListenerRunning:
|
|
"""`ensure_listener_running` self-heal (0.8.0) — listener revived after a
|
|
pod stop / tray Quit stop_listener() while the pod keeps running."""
|
|
|
|
def test_disabled_when_feature_off(self):
|
|
from winpodx.cli.host_open import ensure_listener_running
|
|
|
|
cfg = Config()
|
|
cfg.reverse_open.enabled = False
|
|
assert ensure_listener_running(cfg) == "disabled"
|
|
|
|
def test_running_when_already_up(self, monkeypatch):
|
|
import winpodx.cli.host_open as ho
|
|
|
|
monkeypatch.setattr(ho, "is_listener_running", lambda *a, **k: 4321)
|
|
started = {"n": 0}
|
|
monkeypatch.setattr(ho, "start_listener", lambda *a, **k: started.__setitem__("n", 1))
|
|
cfg = Config()
|
|
cfg.reverse_open.enabled = True
|
|
assert ho.ensure_listener_running(cfg) == "running"
|
|
assert started["n"] == 0 # must NOT spawn a second listener
|
|
|
|
def test_started_when_down(self, monkeypatch, tmp_path):
|
|
import winpodx.cli.host_open as ho
|
|
|
|
monkeypatch.setattr(ho, "is_listener_running", lambda *a, **k: None)
|
|
monkeypatch.setattr(
|
|
ho, "_listener_config", lambda _cfg: type("LC", (), {"incoming_dir": tmp_path / "in"})()
|
|
)
|
|
monkeypatch.setattr(ho, "_apps_json", lambda: tmp_path / "apps.json")
|
|
monkeypatch.setattr(ho, "_seen_uuids_path", lambda: tmp_path / "seen")
|
|
spawned = []
|
|
monkeypatch.setattr(ho, "start_listener", lambda *a, **k: spawned.append(a) or 9999)
|
|
cfg = Config()
|
|
cfg.reverse_open.enabled = True
|
|
assert ho.ensure_listener_running(cfg) == "started"
|
|
assert len(spawned) == 1
|
|
assert (tmp_path / "in").is_dir() # incoming dir created
|
|
|
|
def test_failed_when_start_raises(self, monkeypatch, tmp_path):
|
|
import winpodx.cli.host_open as ho
|
|
from winpodx.reverse_open.lifecycle import ListenerStartFailed
|
|
|
|
monkeypatch.setattr(ho, "is_listener_running", lambda *a, **k: None)
|
|
monkeypatch.setattr(
|
|
ho, "_listener_config", lambda _cfg: type("LC", (), {"incoming_dir": tmp_path / "in"})()
|
|
)
|
|
monkeypatch.setattr(ho, "_apps_json", lambda: tmp_path / "apps.json")
|
|
monkeypatch.setattr(ho, "_seen_uuids_path", lambda: tmp_path / "seen")
|
|
|
|
def _boom(*a, **k):
|
|
raise ListenerStartFailed("boom")
|
|
|
|
monkeypatch.setattr(ho, "start_listener", _boom)
|
|
cfg = Config()
|
|
cfg.reverse_open.enabled = True
|
|
assert ho.ensure_listener_running(cfg) == "failed"
|