Compare commits

...

10 Commits

Author SHA1 Message Date
Brian Krabach 57a4c771a0 chore: bump version to 0.6.7
CI / test (3.12) (push) Failing after 1m4s
CI / test (3.13) (push) Failing after 1m7s
CI / test (3.11) (push) Failing after 15m12s
2026-05-17 18:29:37 -07:00
Brian Krabach 4abb5186e2 fix(service): emit launchd ProgramArguments as separate strings (not embedded spaces)
On one MacBook the plist generated by 'muxplex service install' contained:

  <key>ProgramArguments</key>
  <array>
    <string>/Users/brkrabac/.../python3 -m muxplex</string>
    <string>serve</string>
  </array>

launchd treats each <string> element as a literal, unsplit argv token.  The
first element was a single string with embedded spaces, so launchd tried to
exec a binary literally named 'python3 -m muxplex' (including the spaces) —
which doesn't exist.  Because KeepAlive=true, launchd respawned the failed
exec every few seconds, so 'pgrep' showed a PID and 'muxplex doctor' reported
'Service: launchd agent running' even though the daemon NEVER bound to port
8088.  The user had no visible signal.

Root cause: _resolve_muxplex_bin() returned a fallback string of the form
"$sys.executable -m muxplex" (a single string with spaces), which was
placed verbatim into a single <string> tag.

Fix:
* Add _resolve_muxplex_bin_for_launchd() which returns a list[str] of tokens:
  1. Prefer ~/.local/bin/muxplex (stable uv-tool console-script symlink,
     survives 'uv tool reinstall' without changing path — Option A from spec).
  2. Fall back to shutil.which('muxplex').
  3. Last resort: [sys.executable, '-m', 'muxplex'] — correctly split.
* Update _LAUNCHD_PLIST_TEMPLATE to take a {program_arguments_xml} placeholder
  instead of a single {muxplex_bin}.
* In _launchd_install(), build argv = bin_args + ['serve'] and render each
  token as its own <string> element.

Now the generated plist reads:

  <key>ProgramArguments</key>
  <array>
    <string>/home/user/.local/bin/muxplex</string>
    <string>serve</string>
  </array>

Users who have an existing malformed plist will pick up the fix the next time
they run 'muxplex service install' (or after the upgrade flow regenerates the
service file).

Test added (test_service.py, under 'v0.6.7 fixes'):
  - test_launchd_plist_program_arguments_are_separate_strings: calls
    _launchd_install(), parses the result with plistlib.loads, asserts
    ProgramArguments is a list with >= 2 elements and that NO element
    contains a space.
2026-05-17 18:29:19 -07:00
Brian Krabach a80c6a76b5 fix(cli): verify service is actually active after start, daemon-reload first for stale units
During the v0.6.5→v0.6.6 rollout on spark-1 (systemd Linux) the upgrade flow
printed 'Restarting systemd service...' and 'Service started' but the unit was
actually inactive (dead) for 14 minutes afterward.  The PWA was dark and no
alert fired because the CLI reported success.

Root cause: we fired daemon-reload + start but never verified the result.  If
the unit was left in a 'failed' state (e.g. stale unit-file mismatch after the
previous ExecStart was regenerated mid-upgrade), systemd silently ignored the
start.

Fix:
* Add _probe_service_port(port) — lightweight HTTP probe to
  localhost:port/login that returns True on any HTTP response.
* Add _verify_service_started(timeout_s=10) — polls 'systemctl --user
  is-active' once (synchronous path) or polls the port via HTTP (async
  launchctl path) and returns True/False.
* In upgrade() finally block (systemd path): call daemon-reload BEFORE start
  (already present but now documented), then call _verify_service_started().
  If not active, do reset-failed + retry start once.  If still not active,
  print a clear error and set _service_restart_failed = True.
* Propagate _service_restart_failed as sys.exit(1) after the try/finally, so
  callers and scripts can detect that the service is not running.
* Augment doctor() systemd check: probe is-active and downgrade to a warn '!'
  marker when the unit file exists but the service is not active.
* Augment doctor() launchd check: probe the HTTP port after confirming the
  agent is registered; report '! launchd agent registered but not serving on
  port N' when the port is not responding (catches the silent-failure mode from
  Fix 2 as well).

Tests added (test_cli.py, under 'v0.6.7 fixes'):
  - _verify_service_started returns True when is-active exits 0
  - _verify_service_started returns False when is-active exits 3 (inactive)
  - upgrade() exits 1 when service fails to restart after install
  - upgrade() calls daemon-reload before start (call-order assertion)
  - doctor() reports 'registered but not serving' when launchd agent is up
    but port is not bound
2026-05-17 18:28:54 -07:00
Brian Krabach b75be60e7f chore: bump version to 0.6.6 2026-05-17 17:23:58 -07:00
Brian Krabach c5e146bcf7 feat(ui): version-busting query suffix on all served asset URLs
Browser-tester confirmed on spark-1 that the 7 <script src> tags and all
<link href> tags in the served HTML had no cache-busting suffix.  Because
there is no service worker, the standard HTTP cache would keep serving stale
JS/CSS to browsers that had already loaded a previous release — the root
cause of yesterday's 'is the user seeing stale JS?' investigation.

Fix: index_page() now appends ?v=<muxplex_version> to every static-asset
URL (src= and href= attributes whose path starts with /  and does not begin
with /api/) before returning the HTML response.  The version is read once at
module load from importlib.metadata.version('muxplex') — the same source
used by the doctor command — so it is always in sync with the installed
package.  Starlette's StaticFiles handler ignores query parameters when
serving files from disk, so the versioned URLs continue to resolve to the
same bytes; they just carry a new cache key on every release.

Affected assets (7 <script> + 6 <link> + 1 <img>):
  /vendor/xterm.js, /vendor/xterm-addon-fit.js,
  /vendor/xterm-addon-web-links.js, /vendor/xterm-addon-search.js,
  /vendor/addon-image.js, /app.js, /terminal.js
  /manifest.json, /favicon.ico, /favicon-32.png, /apple-touch-icon.png,
  /vendor/xterm.css, /style.css
  /wordmark-on-dark.svg

New tests in muxplex/tests/test_main.py:
  - test_index_all_asset_urls_have_version_suffix — every <script src> and
    <link href> in GET / carries ?v=<version>
  - test_index_vendor_scripts_each_versioned — all 7 expected script URLs
    are present with the version suffix
  - test_versioned_asset_url_resolves_to_static_file — static handler
    serves the file correctly when the query string is present
2026-05-17 17:22:25 -07:00
Brian Krabach b30891f3e5 chore: bump version to 0.6.5 2026-05-17 16:55:27 -07:00
Brian Krabach c3a059a7ee fix(ui): suppress empty device status tiles in all view modes (not just grouped)
v0.6.4 added `&& _gridViewMode !== 'grouped'` to the status:empty branch in
renderGrid, suppressing the "No sessions" tile only in grouped grid view.

User reported the alienware-r13 "No sessions" tile still appearing in flat view
(gridViewMode: "flat"), which is the muxplex default. The original request was
unambiguous: "don't need a block for those that don't have anything to show" —
no view-mode qualifier was implied.

Changes:
- Both renderGrid call sites (early-return branch for visible.length === 0, and
  the normal rendering branch) now simply omit the status:empty else-if entirely.
  The status:empty sentinel may still arrive from the server; the renderer silently
  ignores it in every view mode.
- Updated surrounding comments to reflect the new unconditional suppression.
- Updated/replaced tests that asserted the old (wrong) flat-mode behaviour:
  - Replaced 'renderGrid shows "No sessions" status tile for status=empty devices'
    with a v0.6.5 variant that asserts NO tile is emitted.
  - Removed 'v0.6.4: status:empty block IS still rendered in flat grid mode'
    (superseded; kept as a tombstone comment).
- Added three new tests under 'v0.6.5 empty tile suppressed in all view modes':
  1. Flat view: empty sentinel → no source-tile--empty, no 'No sessions' text.
  2. Grouped view: same (regression guard for the v0.6.4 suppression).
  3. Mixed input: real session renders, empty sentinel is discarded, both modes.

No other rendering paths that produce a per-device block for empty devices were
found: renderSidebar and renderGroupedGrid both operate on getVisibleSessions()
output which already excludes status-bearing sentinels.

Test count: 402 (v0.6.4) → 404 (v0.6.5).
2026-05-17 16:55:12 -07:00
Brian Krabach 54772ba12b chore: bump version to 0.6.4 2026-05-17 16:37:21 -07:00
Brian Krabach 13e5cb2484 fix(cli): probe known uv/pip locations off PATH and propagate caught install failures
On three fleet devices muxplex update failed silently because
shutil.which('uv') returned None even though uv was installed:
  - tower (Unraid/root):   /root/.local/bin/uv
  - macOS (user):          ~/.local/bin/uv
  - spark-1 (snap):        /snap/bin/uv

The muxplex process running under systemd / launchd inherits a stripped
PATH that omits ~/.local/bin and /snap/bin.  shutil.which gives up at
PATH exhaustion; _find_uv() doesn't.

Add _find_uv() and _find_pip() helpers that:
  1. Try shutil.which first (PATH fast path).
  2. If that returns None, probe a curated list of known install locations
     checking os.path.exists + os.access(X_OK) for each candidate.
  3. Return the first found path, or None.

Known locations covered:
  uv:  ~/.local/bin/uv, /opt/homebrew/bin/uv, /usr/local/bin/uv,
       /snap/bin/uv, /root/.local/bin/uv
  pip: ~/.local/bin/{pip,pip3}, /opt/homebrew/bin/pip3,
       /usr/local/bin/pip3, /root/.local/bin/{pip,pip3}

shutil.which() calls for systemctl/launchctl in service.py are not
changed — those tools are reliably on PATH when present.

Exit-code propagation (sys.exit(1) on _install_failed) was already
implemented in v0.6.2; this commit adds 9 tests confirming the complete
behaviour including the uv/pip path-probing and exit-code paths.

Updated test_upgrade_falls_back_to_pip_when_uv_absent to monkeypatch
_find_uv directly (avoids false positives on dev systems where uv is
installed at a known non-PATH location).
2026-05-17 16:36:47 -07:00
Brian Krabach c5921eba65 fix(ui): suppress empty device blocks in renderGrid grouped mode
Root cause: the v0.6.3 fix added a guard in renderGroupedGrid
(groupSessions.length === 0) that is unreachable — groups are built by
iterating the already-filtered session list, so every group that exists
always has ≥1 entry.  The actual empty-device block was coming from a
separate code path in renderGrid that unconditionally appends a
source-tile--empty status tile for every federation remote whose server
returns {status: 'empty'}.  In flat mode this is correct and intentional
("No sessions" badge); in grouped mode it produced a visible block
showing the device name, exactly what the user saw for alienware-r13.

The v0.6.3 tests only covered the hidden-sessions case (sessions exist
but are hidden); they didn't cover the zero-sessions case (remote device
online, zero tmux sessions → server emits status:empty sentinel).  Both
status-tile generation paths in renderGrid (the visible.length===0 early
return and the main append-at-end path) were appending the empty tile
regardless of gridViewMode.

Fix: skip status:empty tiles when _gridViewMode === 'grouped'.
auth_failed and unreachable tiles are still shown in all modes because
they represent actionable error states.

Adds three regression tests in test_app.mjs:
- status:empty NOT rendered in grouped mode (would have caught the bug)
- status:empty IS still rendered in flat mode (backward compat)
- auth_failed/unreachable still appear in grouped mode
2026-05-17 16:36:24 -07:00
10 changed files with 1194 additions and 33 deletions
+21
View File
@@ -1,5 +1,26 @@
# Changelog
## v0.6.4 (2026-05-17)
### Bug Fixes
- **Empty device block still showing in grouped grid view** — Remote federation devices with
zero tmux sessions were producing a visible "No sessions" block in the grouped grid view.
The v0.6.3 fix targeted `renderGroupedGrid` but missed the unconditional `status:empty`
status-tile append in `renderGrid` itself. In grouped mode, `status:empty` tiles are now
suppressed (`auth_failed` and `unreachable` tiles still appear in all modes).
- **`muxplex update` fails when uv/pip is installed outside PATH** — On Unraid (root user),
macOS (user installs), and snap-packaged systems, `shutil.which("uv")` returned None even
though uv was present at `~/.local/bin/uv`, `/snap/bin/uv`, or `/root/.local/bin/uv`.
New helpers `_find_uv()` / `_find_pip()` probe a curated list of known install locations
after PATH lookup fails, so the upgrade flow works on stripped-PATH environments
(systemd, launchd, non-login SSH shells).
- **`muxplex update` exit code propagation** — Tests added to confirm that a failed install
exits with code 1 after the `try/finally` service-recovery block runs (behaviour was
implemented in v0.6.2; regression test coverage added here).
## v0.5.0 (2026-05-06)
### Features
+202 -11
View File
@@ -31,6 +31,136 @@ def _have_launchctl() -> bool:
return shutil.which("launchctl") is not None
def _probe_service_port(port: int) -> bool:
"""Return True if a muxplex server is responding on localhost:port.
Tries HTTPS first (self-signed cert tolerated), then HTTP. Any HTTP
response code (including 4xx/5xx) confirms the server is listening.
A connection error, timeout, or SSL failure means the server is not up.
"""
import ssl
import urllib.error
import urllib.request
for scheme in ("https", "http"):
try:
url = f"{scheme}://localhost:{port}/login"
if scheme == "https":
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
with urllib.request.urlopen(url, timeout=5, context=ctx) as _resp:
return True
else:
with urllib.request.urlopen(url, timeout=5) as _resp:
return True
except urllib.error.HTTPError:
# Server returned an HTTP error — it IS running
return True
except Exception:
pass # Connection refused, timeout, SSL issue — try next scheme
return False
def _verify_service_started(timeout_s: int = 10) -> bool:
"""Verify the muxplex service is actually serving after a start command.
For systemctl: calls ``systemctl --user is-active muxplex`` once and
returns ``True`` only when the unit is ``active`` (exit code 0).
``systemctl start`` is synchronous so a single check is sufficient.
For launchctl: polls ``_probe_service_port()`` until a successful HTTP
response is received or ``timeout_s`` seconds have elapsed. launchd
starts processes asynchronously, so polling is necessary.
Returns ``False`` if the service is not active / not responding.
"""
import time
if _have_systemctl():
result = subprocess.run(
["systemctl", "--user", "is-active", "muxplex"],
capture_output=True,
text=True,
)
return result.returncode == 0
if _have_launchctl():
from muxplex.settings import load_settings # noqa: PLC0415
cfg = load_settings()
port = cfg.get("port", 8088)
deadline = time.monotonic() + timeout_s
while True:
if _probe_service_port(port):
return True
remaining = deadline - time.monotonic()
if remaining <= 0:
return False
time.sleep(min(1.0, remaining))
return False
def _find_uv() -> str | None:
"""Locate the ``uv`` binary, checking PATH first then well-known install locations.
``shutil.which("uv")`` fails on systems where the muxplex process inherits a
stripped PATH (e.g. under systemd/launchd or non-login SSH shells) that does not
include ``~/.local/bin`` or ``/snap/bin``. This helper falls back to a curated
list of locations observed in the wild:
* ``~/.local/bin/uv`` — pip-style user installs (Linux, macOS)
* ``/opt/homebrew/bin/uv`` — Homebrew on Apple Silicon
* ``/usr/local/bin/uv`` — Homebrew on Intel macOS, manual installs
* ``/snap/bin/uv`` — snap-packaged uv (Ubuntu / snap-enabled distros)
* ``/root/.local/bin/uv`` — root user on Unraid / headless Linux
Returns the first found path, or ``None`` if uv is not available.
"""
found = shutil.which("uv")
if found:
return found
candidates = [
str(Path.home() / ".local" / "bin" / "uv"),
"/opt/homebrew/bin/uv",
"/usr/local/bin/uv",
"/snap/bin/uv",
"/root/.local/bin/uv",
]
for path in candidates:
if os.path.exists(path) and os.access(path, os.X_OK):
return path
return None
def _find_pip() -> str | None:
"""Locate a ``pip`` / ``pip3`` binary, checking PATH first then well-known locations.
Mirrors ``_find_uv()``'s strategy: try ``shutil.which`` for ``pip`` and ``pip3``,
then probe a curated list of known install paths so that pip can be found even
when the process PATH is stripped.
Returns the first found path, or ``None`` if no pip variant is available.
"""
for name in ("pip", "pip3"):
found = shutil.which(name)
if found:
return found
candidates = [
str(Path.home() / ".local" / "bin" / "pip"),
str(Path.home() / ".local" / "bin" / "pip3"),
"/opt/homebrew/bin/pip3",
"/usr/local/bin/pip3",
"/root/.local/bin/pip",
"/root/.local/bin/pip3",
]
for path in candidates:
if os.path.exists(path) and os.access(path, os.X_OK):
return path
return None
def _get_install_info() -> dict:
"""Detect how muxplex was installed using PEP 610 direct_url.json.
@@ -444,7 +574,18 @@ def doctor() -> None:
text=True,
)
if result.returncode == 0:
print(f" {ok_mark} Service: launchd agent running")
# Agent is registered — verify it is actually serving
from muxplex.settings import load_settings # noqa: PLC0415
_cfg = load_settings()
_port = _cfg.get("port", 8088)
if _probe_service_port(_port):
print(f" {ok_mark} Service: launchd agent running")
else:
print(
f" {warn_mark} Service: launchd agent registered but"
f" not serving on port {_port}"
)
else:
print(
f" {warn_mark} Service: launchd agent installed but not running ({plist})"
@@ -465,9 +606,21 @@ def doctor() -> None:
Path.home() / ".config" / "systemd" / "user" / "muxplex.service"
)
if systemd_user.exists():
print(
f" {ok_mark} Service: systemd user unit installed ({systemd_user})"
_active = subprocess.run(
["systemctl", "--user", "is-active", "muxplex"],
capture_output=True,
text=True,
)
if _active.returncode == 0:
print(
f" {ok_mark} Service: systemd user unit installed ({systemd_user})"
)
else:
_state = _active.stdout.strip() or "unknown"
print(
f" {warn_mark} Service: systemd user unit installed but"
f" not active — state: {_state} ({systemd_user})"
)
elif _system_service_path.exists():
print(
f" {ok_mark} Service: systemd system unit installed ({_system_service_path})"
@@ -545,7 +698,7 @@ def upgrade(*, force: bool = False) -> None:
if info["source"] == "pypi" or _is_uv_managed
else "git+https://github.com/bkrabach/muxplex"
)
uv_path = shutil.which("uv")
uv_path = _find_uv()
# Pre-compute macOS service identifiers — used in both stop and finally blocks.
label = "com.muxplex"
@@ -585,6 +738,7 @@ def upgrade(*, force: bool = False) -> None:
# 2+4. Install (try) with guaranteed service restart in finally.
# Bug 1+2b: try/finally ensures the start step always runs — success OR failure.
_install_failed = False
_service_restart_failed = False
print(" Installing latest version...")
try:
# Bug 3: dispatch — uv-tool-managed gets --reinstall; plain uv/pip otherwise
@@ -608,8 +762,8 @@ def upgrade(*, force: bool = False) -> None:
else:
print(" Installed successfully")
else:
# Bug 3: uv absent → fall back to pip
pip_path = shutil.which("pip") or shutil.which("pip3")
# uv absent → fall back to pip (probe known locations off PATH)
pip_path = _find_pip()
if pip_path:
result = subprocess.run(
[pip_path, "install", "--upgrade", install_target],
@@ -646,14 +800,21 @@ def upgrade(*, force: bool = False) -> None:
capture_output=True,
text=True,
)
if result.returncode == 0:
print(" Service started")
else:
if result.returncode != 0:
# Fallback to legacy load for older macOS
subprocess.run(
["launchctl", "load", str(plist)], capture_output=True
)
print(" Service started (legacy)")
# Verify the agent is actually serving (not just registered)
if _verify_service_started():
print(" Service started")
else:
print(
" ERROR: launchd agent registered but the service is"
" not responding after upgrade.\n"
" Check /tmp/muxplex.err for details."
)
_service_restart_failed = True
else:
print(" Service file not found — run: muxplex service install")
elif _have_systemctl():
@@ -664,13 +825,35 @@ def upgrade(*, force: bool = False) -> None:
)
if result.returncode == 0:
print(" Restarting systemd service...")
# daemon-reload FIRST: picks up any regenerated unit file so
# the start command sees the correct ExecStart (spark-1 fix).
subprocess.run(
["systemctl", "--user", "daemon-reload"], capture_output=True
)
subprocess.run(
["systemctl", "--user", "start", "muxplex"], capture_output=True
)
print(" Service started")
if not _verify_service_started():
# Unit may have landed in 'failed' state (e.g. port race
# on first start). Reset the failure counter and retry once.
subprocess.run(
["systemctl", "--user", "reset-failed", "muxplex"],
capture_output=True,
)
subprocess.run(
["systemctl", "--user", "start", "muxplex"],
capture_output=True,
)
if _verify_service_started():
print(" Service started")
else:
print(
" ERROR: muxplex service is not active after upgrade.\n"
" Run: systemctl --user status muxplex"
)
_service_restart_failed = True
else:
print(" Service started")
else:
print(" Service not enabled — run: muxplex service install")
else:
@@ -700,6 +883,14 @@ def upgrade(*, force: bool = False) -> None:
)
sys.exit(1)
if _service_restart_failed:
print(
"\n ERROR: upgrade installed successfully but the service failed to restart.\n"
" The new version is installed but the service is NOT running.\n"
" Run: muxplex service start\n"
)
sys.exit(1)
# 5. Doctor check
print("\n Verifying...")
doctor()
+6 -4
View File
@@ -1658,12 +1658,13 @@ function renderGrid(sessions) {
var visible = getVisibleSessions(sessions);
if (visible.length === 0) {
// Build status tiles for auth_failed/unreachable sessions even when no regular sessions exist
// Build status tiles for auth_failed/unreachable sessions even when no regular sessions exist.
// status:empty sentinels are intentionally ignored — a remote with zero tmux sessions
// produces no visible tile in any view mode (flat, grouped, or otherwise).
var statusTilesHtml = '';
(sessions || []).forEach(function(session) {
if (session.status === 'auth_failed') statusTilesHtml += buildStatusTileHTML(session.deviceName, 'Auth required', 'auth');
else if (session.status === 'unreachable') statusTilesHtml += buildStatusTileHTML(session.deviceName, 'Offline', 'offline');
else if (session.status === 'empty') statusTilesHtml += buildStatusTileHTML(session.deviceName, 'No sessions', 'empty');
});
if (grid) grid.innerHTML = statusTilesHtml;
// Only show empty-state when there are truly no tiles at all
@@ -1695,12 +1696,13 @@ function renderGrid(sessions) {
html = ordered.map(function(session, index) { return buildTileHTML(session, index, mobile); }).join('');
}
// Append status tiles for auth_failed, unreachable, and empty sessions
// Append status tiles for auth_failed and unreachable sessions. status:empty sentinels are
// intentionally ignored in all view modes — a remote with zero tmux sessions produces no
// visible tile. auth_failed and unreachable are actionable error states and are always shown.
var statusTilesHtml = '';
(sessions || []).forEach(function(session) {
if (session.status === 'auth_failed') statusTilesHtml += buildStatusTileHTML(session.deviceName, 'Auth required', 'auth');
else if (session.status === 'unreachable') statusTilesHtml += buildStatusTileHTML(session.deviceName, 'Offline', 'offline');
else if (session.status === 'empty') statusTilesHtml += buildStatusTileHTML(session.deviceName, 'No sessions', 'empty');
});
if (grid) grid.innerHTML = html + statusTilesHtml;
+250 -9
View File
@@ -4829,13 +4829,14 @@ test('renderGrid status tiles use session.deviceName not session.name for offlin
// --- renderGrid: status=empty shows "No sessions" tile ---
test('renderGrid shows "No sessions" status tile for status=empty devices', () => {
// A device that is online but has zero tmux sessions returns
test('renderGrid silently drops status=empty devices (no tile emitted) [v0.6.5]', () => {
// v0.6.5: a device that is online but has zero tmux sessions returns
// {status: 'empty', deviceName: '...'} from the federation endpoint.
// renderGrid must render a status tile with the text "No sessions" (not blank).
// renderGrid must NOT render any tile for it — the user asked not to see
// blocks for devices that have nothing to show (flat OR grouped mode).
//
// Before implementation: fails because neither status loop handles status === 'empty',
// so the tile is never rendered and grid.innerHTML stays empty.
// Previously (pre-v0.6.5) a "No sessions" status tile was emitted; that
// behaviour is intentionally removed.
const grid = { innerHTML: '' };
const emptyState = { style: {}, classList: { add() {}, remove() {} } };
const origGetById = globalThis.document.getElementById;
@@ -4848,12 +4849,16 @@ test('renderGrid shows "No sessions" status tile for status=empty devices', () =
app.renderGrid([{ status: 'empty', deviceName: 'quiet-box', remoteId: 3 }]);
assert.ok(
grid.innerHTML.includes('No sessions'),
`renderGrid must include "No sessions" text for status=empty device, got: ${grid.innerHTML}`
!grid.innerHTML.includes('No sessions'),
`renderGrid must NOT include "No sessions" text for status=empty device, got: ${grid.innerHTML}`
);
assert.ok(
grid.innerHTML.includes('quiet-box'),
`renderGrid must include the deviceName "quiet-box" in the status tile, got: ${grid.innerHTML}`
!grid.innerHTML.includes('quiet-box'),
`renderGrid must NOT include the deviceName "quiet-box" for a status=empty device, got: ${grid.innerHTML}`
);
assert.ok(
!grid.innerHTML.includes('source-tile--empty'),
`renderGrid must NOT emit source-tile--empty class for status=empty device, got: ${grid.innerHTML}`
);
globalThis.document.getElementById = origGetById;
@@ -5584,6 +5589,242 @@ test('v0.6.3: grouped view still shows device header when device has at least on
app._setActiveView('all');
});
// ─── v0.6.4 empty-device-block regression ────────────────────────────────────
test('v0.6.4: status:empty block is NOT rendered in grouped grid mode for a remote with zero sessions', () => {
// The real user scenario (spark-1 viewing alienware-r13 via federation):
// alienware-r13 has zero tmux sessions → server emits {status:"empty", deviceName:"alienware-r13"}.
// spark-1 (the local device) still has its own sessions.
// BEFORE v0.6.4: renderGrid appended a source-tile--empty block for alienware-r13 even in
// grouped mode, showing the device name in the grid (the user's reported "device block").
// AFTER v0.6.4: the status:empty tile is suppressed in grouped mode.
const sessions = [
{ name: 'local-sess', deviceName: 'spark-1', snapshot: '', sessionKey: 'spark-1:local-sess' },
{ status: 'empty', deviceName: 'alienware-r13', deviceId: 'aw-uuid', remoteId: 'aw-uuid' },
];
app._setServerSettings({ multi_device_enabled: true, hidden_sessions: [] });
app._setGridViewMode('grouped');
app._setActiveView('all');
let capturedHTML = '';
const mockGrid = { get innerHTML() { return capturedHTML; }, set innerHTML(v) { capturedHTML = v; } };
const mockEmpty = { classList: { add: () => {}, remove: () => {} } };
const origGetById = globalThis.document.getElementById;
const origQSA = globalThis.document.querySelectorAll;
globalThis.document.getElementById = (id) => {
if (id === 'session-grid') return mockGrid;
if (id === 'empty-state') return mockEmpty;
return null;
};
globalThis.document.querySelectorAll = () => [];
app.renderGrid(sessions);
assert.ok(
!capturedHTML.includes('alienware-r13'),
'alienware-r13 must NOT appear in grouped grid when the device has zero sessions; got: ' + capturedHTML
);
assert.ok(
!capturedHTML.includes('source-tile--empty'),
'source-tile--empty must NOT be rendered for an empty remote in grouped mode; got: ' + capturedHTML
);
assert.ok(
capturedHTML.includes('spark-1') || capturedHTML.includes('local-sess'),
'spark-1 local session must still appear; got: ' + capturedHTML
);
globalThis.document.getElementById = origGetById;
globalThis.document.querySelectorAll = origQSA;
app._setGridViewMode('flat');
app._setServerSettings(null);
app._setActiveView('all');
});
// NOTE: v0.6.4 tested that status:empty blocks appeared in flat mode. That behaviour was
// wrong and is superseded by v0.6.5, which silently drops status:empty in ALL view modes.
// See the v0.6.5 section below for the authoritative tests.
test('v0.6.4: auth_failed and unreachable tiles still appear in grouped mode', () => {
// Only status:empty is suppressed in grouped mode; auth_failed and unreachable
// are actionable error states and must always be shown.
const sessions = [
{ name: 'local-sess', deviceName: 'spark-1', snapshot: '', sessionKey: 'spark-1:local-sess' },
{ status: 'auth_failed', deviceName: 'device-b', deviceId: 'b-uuid', remoteId: 'b-uuid' },
{ status: 'unreachable', deviceName: 'device-c', deviceId: 'c-uuid', remoteId: 'c-uuid' },
];
app._setServerSettings({ multi_device_enabled: true, hidden_sessions: [] });
app._setGridViewMode('grouped');
app._setActiveView('all');
let capturedHTML = '';
const mockGrid = { get innerHTML() { return capturedHTML; }, set innerHTML(v) { capturedHTML = v; } };
const mockEmpty = { classList: { add: () => {}, remove: () => {} } };
const origGetById = globalThis.document.getElementById;
const origQSA = globalThis.document.querySelectorAll;
globalThis.document.getElementById = (id) => {
if (id === 'session-grid') return mockGrid;
if (id === 'empty-state') return mockEmpty;
return null;
};
globalThis.document.querySelectorAll = () => [];
app.renderGrid(sessions);
assert.ok(
capturedHTML.includes('source-tile--auth'),
'auth_failed tile must appear in grouped mode; got: ' + capturedHTML
);
assert.ok(
capturedHTML.includes('source-tile--offline'),
'unreachable tile must appear in grouped mode; got: ' + capturedHTML
);
globalThis.document.getElementById = origGetById;
globalThis.document.querySelectorAll = origQSA;
app._setGridViewMode('flat');
app._setServerSettings(null);
app._setActiveView('all');
});
// ─── v0.6.5 empty tile suppressed in all view modes ──────────────────────────
test('v0.6.5: flat view — status:empty sentinel produces NO tile', () => {
// User-reported bug: alienware-r13 showed a "No sessions" tile in flat view
// (the muxplex default). v0.6.4 only suppressed it in grouped mode.
// v0.6.5 drops status:empty tiles unconditionally regardless of view mode.
const sessions = [
{ name: 'local-sess', deviceName: 'spark-1', snapshot: '', sessionKey: 'spark-1:local-sess' },
{ status: 'empty', deviceName: 'alienware-r13', sessionKey: 'devX:_empty' },
];
app._setServerSettings({ multi_device_enabled: true, hidden_sessions: [] });
app._setGridViewMode('flat');
app._setActiveView('all');
let capturedHTML = '';
const mockGrid = { get innerHTML() { return capturedHTML; }, set innerHTML(v) { capturedHTML = v; } };
const mockEmpty = { classList: { add: () => {}, remove: () => {} } };
const origGetById = globalThis.document.getElementById;
const origQSA = globalThis.document.querySelectorAll;
globalThis.document.getElementById = (id) => {
if (id === 'session-grid') return mockGrid;
if (id === 'empty-state') return mockEmpty;
return null;
};
globalThis.document.querySelectorAll = () => [];
app.renderGrid(sessions);
assert.ok(
!capturedHTML.includes('source-tile--empty'),
'source-tile--empty must NOT appear in flat mode for an empty remote; got: ' + capturedHTML
);
assert.ok(
!capturedHTML.includes('No sessions'),
'"No sessions" text must NOT appear in flat mode for an empty remote; got: ' + capturedHTML
);
assert.ok(
capturedHTML.includes('local-sess') || capturedHTML.includes('spark-1'),
'the real local session must still render; got: ' + capturedHTML
);
globalThis.document.getElementById = origGetById;
globalThis.document.querySelectorAll = origQSA;
app._setGridViewMode('flat');
app._setServerSettings(null);
app._setActiveView('all');
});
test('v0.6.5: grouped view — status:empty sentinel still produces NO tile (regression)', () => {
// v0.6.4 suppressed empty tiles only in grouped mode; v0.6.5 keeps that suppression
// intact. This is a regression guard: grouped suppression must not have been broken
// while fixing flat mode.
const sessions = [
{ name: 'local-sess', deviceName: 'spark-1', snapshot: '', sessionKey: 'spark-1:local-sess' },
{ status: 'empty', deviceName: 'alienware-r13', sessionKey: 'devX:_empty' },
];
app._setServerSettings({ multi_device_enabled: true, hidden_sessions: [] });
app._setGridViewMode('grouped');
app._setActiveView('all');
let capturedHTML = '';
const mockGrid = { get innerHTML() { return capturedHTML; }, set innerHTML(v) { capturedHTML = v; } };
const mockEmpty = { classList: { add: () => {}, remove: () => {} } };
const origGetById = globalThis.document.getElementById;
const origQSA = globalThis.document.querySelectorAll;
globalThis.document.getElementById = (id) => {
if (id === 'session-grid') return mockGrid;
if (id === 'empty-state') return mockEmpty;
return null;
};
globalThis.document.querySelectorAll = () => [];
app.renderGrid(sessions);
assert.ok(
!capturedHTML.includes('source-tile--empty'),
'source-tile--empty must NOT appear in grouped mode for an empty remote; got: ' + capturedHTML
);
assert.ok(
!capturedHTML.includes('No sessions'),
'"No sessions" text must NOT appear in grouped mode for an empty remote; got: ' + capturedHTML
);
globalThis.document.getElementById = origGetById;
globalThis.document.querySelectorAll = origQSA;
app._setGridViewMode('flat');
app._setServerSettings(null);
app._setActiveView('all');
});
test('v0.6.5: real sessions still render alongside an empty sentinel', () => {
// A mix of a real session and a status:empty sentinel. The real session must
// appear; the empty sentinel must be silently discarded in both flat and grouped modes.
const sessions = [
{ name: 'my-sess', deviceName: 'spark-1', snapshot: '', sessionKey: 'spark-1:my-sess' },
{ status: 'empty', deviceName: 'alienware-r13', sessionKey: 'devX:_empty' },
];
app._setServerSettings({ multi_device_enabled: true, hidden_sessions: [] });
app._setActiveView('all');
for (const mode of ['flat', 'grouped']) {
app._setGridViewMode(mode);
let capturedHTML = '';
const mockGrid = { get innerHTML() { return capturedHTML; }, set innerHTML(v) { capturedHTML = v; } };
const mockEmpty = { classList: { add: () => {}, remove: () => {} } };
const origGetById = globalThis.document.getElementById;
const origQSA = globalThis.document.querySelectorAll;
globalThis.document.getElementById = (id) => {
if (id === 'session-grid') return mockGrid;
if (id === 'empty-state') return mockEmpty;
return null;
};
globalThis.document.querySelectorAll = () => [];
app.renderGrid(sessions);
assert.ok(
capturedHTML.includes('my-sess') || capturedHTML.includes('spark-1'),
`[${mode}] real session must render; got: ` + capturedHTML
);
assert.ok(
!capturedHTML.includes('source-tile--empty'),
`[${mode}] source-tile--empty must NOT appear; got: ` + capturedHTML
);
assert.ok(
!capturedHTML.includes('No sessions'),
`[${mode}] "No sessions" text must NOT appear; got: ` + capturedHTML
);
globalThis.document.getElementById = origGetById;
globalThis.document.querySelectorAll = origQSA;
}
app._setGridViewMode('flat');
app._setServerSettings(null);
app._setActiveView('all');
});
test('v0.6.3: empty-state still appears when every device has zero visible sessions', () => {
// When ALL sessions across ALL devices are hidden, visible.length === 0.
// renderGrid() must still reach its early-return branch and show empty-state,
+21 -1
View File
@@ -17,6 +17,7 @@ import logging
import os
import pathlib
import pwd
import re
import socket
import ssl
import shutil
@@ -527,6 +528,15 @@ _FRONTEND_DIR = pathlib.Path(__file__).parent / "frontend"
# which machine each muxplex instance is running on.
_HOSTNAME = socket.gethostname().split(".")[0]
# Canonical version string — sourced from package metadata (same as `app.version`
# and the `doctor` command). Used to append `?v=<version>` to every static-asset
# URL so browsers immediately pick up new code on each release.
_UI_VERSION: str = importlib.metadata.version("muxplex")
# Matches src="/<path>" and href="/<path>" in served HTML, excluding /api/ URLs.
# Used by index_page() to inject cache-busting version query parameters.
_ASSET_URL_RE = re.compile(r'((?:src|href)=")((?!/api/)/[^"?#]*)')
# ---------------------------------------------------------------------------
# Routes
@@ -1204,12 +1214,22 @@ async def federation_terminal_ws_proxy(websocket: WebSocket, device_id: str) ->
@app.get("/", response_class=HTMLResponse)
@app.get("/index.html", response_class=HTMLResponse)
async def index_page():
"""Serve index.html with hostname injected into the page title."""
"""Serve index.html with hostname injected into the page title.
Also appends ``?v=<version>`` to every static-asset URL (script src, link
href) so browsers immediately pick up new code on each release rather than
serving stale JS/CSS from the HTTP cache. API URLs (/api/...) are
excluded they are not HTTP-cached by browsers.
"""
html = (_FRONTEND_DIR / "index.html").read_text()
html = html.replace(
"<title>muxplex</title>",
f"<title>{_HOSTNAME} \u2014 muxplex</title>",
)
html = _ASSET_URL_RE.sub(
lambda m: f"{m.group(1)}{m.group(2)}?v={_UI_VERSION}",
html,
)
return HTMLResponse(html)
+39 -4
View File
@@ -44,8 +44,7 @@ _LAUNCHD_PLIST_TEMPLATE = """\
<string>{label}</string>
<key>ProgramArguments</key>
<array>
<string>{muxplex_bin}</string>
<string>serve</string>
{program_arguments_xml}
</array>
<key>EnvironmentVariables</key>
<dict>
@@ -91,6 +90,33 @@ def _resolve_muxplex_bin() -> str:
return f"{sys.executable} -m muxplex"
def _resolve_muxplex_bin_for_launchd() -> list[str]:
"""Return the argv token list for the muxplex binary in a launchd plist.
Uses Option A: prefer ``~/.local/bin/muxplex`` (stable uv-tool
console-script symlink that survives ``uv tool reinstall``). Falls back
to ``shutil.which("muxplex")``, then to ``[sys.executable, "-m",
"muxplex"]`` as explicitly split tokens.
Each element must become its own ``<string>`` in ProgramArguments.
launchd does **not** shell-split inside a ``<string>``; an element like
``"python3 -m muxplex"`` is treated as a literal executable name, causing
the daemon to silently fail to start.
"""
# Option A: stable console-script symlink installed by `uv tool`
local_bin = Path.home() / ".local" / "bin" / "muxplex"
if local_bin.exists() and os.access(str(local_bin), os.X_OK):
return [str(local_bin)]
# Fall back to PATH lookup
which = shutil.which("muxplex")
if which:
return [which]
# Last resort: explicit python -m invocation — correctly split into tokens
return [sys.executable, "-m", "muxplex"]
# ---------------------------------------------------------------------------
# Helper
# ---------------------------------------------------------------------------
@@ -184,11 +210,20 @@ def _systemd_logs() -> None:
def _launchd_install() -> None:
muxplex_bin = _resolve_muxplex_bin()
bin_args = _resolve_muxplex_bin_for_launchd()
argv = bin_args + ["serve"]
# Each argv token is its own <string> element. launchd does NOT
# shell-split inside a <string>, so we must NOT put the whole command
# (e.g. "python3 -m muxplex") into a single element.
program_arguments_xml = "\n".join(
f" <string>{arg}</string>" for arg in argv
)
base_path = os.environ.get("PATH", "/usr/bin:/bin")
safe_path = f"/opt/homebrew/bin:/usr/local/bin:{base_path}"
plist_content = _LAUNCHD_PLIST_TEMPLATE.format(
label=_LAUNCHD_LABEL, muxplex_bin=muxplex_bin, safe_path=safe_path
label=_LAUNCHD_LABEL,
program_arguments_xml=program_arguments_xml,
safe_path=safe_path,
)
_LAUNCHD_PLIST_DIR.mkdir(parents=True, exist_ok=True)
_LAUNCHD_PLIST_PATH.write_text(plist_content)
+429 -3
View File
@@ -2825,7 +2825,7 @@ def test_upgrade_prefers_uv_tool_when_uv_managed(monkeypatch, capsys):
def test_upgrade_falls_back_to_pip_when_uv_absent(monkeypatch, capsys):
"""upgrade() uses pip install when uv is not on PATH.
"""upgrade() uses pip install when uv is not found anywhere (_find_uv returns None).
Regression test for Bug 3 (v0.6.2): uv absent \u2192 pip must be the installer.
"""
@@ -2841,12 +2841,12 @@ def test_upgrade_falls_back_to_pip_when_uv_absent(monkeypatch, capsys):
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
def fake_which(name):
if name == "uv":
return None # uv absent
if name in ("pip", "pip3"):
return f"/usr/local/bin/{name}"
return f"/usr/bin/{name}"
# _find_uv returns None — uv absent even at known non-PATH locations
monkeypatch.setattr(cli_mod, "_find_uv", lambda: None)
monkeypatch.setattr(shutil, "which", fake_which)
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(cli_mod, "doctor", lambda: None)
@@ -2864,3 +2864,429 @@ def test_upgrade_falls_back_to_pip_when_uv_absent(monkeypatch, capsys):
assert len(pip_calls) > 0, "upgrade() must call pip install when uv is absent"
uv_calls = [c for c in calls if isinstance(c, list) and c and "uv" in str(c[0])]
assert len(uv_calls) == 0, "upgrade() must not call uv when it is absent from PATH"
# ---------------------------------------------------------------------------
# v0.6.4 fixes: _find_uv / _find_pip path probing + exit code propagation
# ---------------------------------------------------------------------------
def test_find_uv_returns_path_from_shutil_which():
"""_find_uv() returns the path that shutil.which('uv') returns when present."""
import muxplex.cli as cli_mod
with patch("muxplex.cli.shutil") as mock_shutil:
mock_shutil.which.return_value = "/usr/local/bin/uv"
result = cli_mod._find_uv()
assert result == "/usr/local/bin/uv", (
"_find_uv must return the shutil.which result when uv is on PATH"
)
def test_find_uv_probes_known_locations_when_which_returns_none(tmp_path, monkeypatch):
"""_find_uv() falls through to the candidate list when shutil.which returns None."""
import muxplex.cli as cli_mod
# Simulate shutil.which returning None for "uv"
monkeypatch.setattr(shutil, "which", lambda name: None if name == "uv" else f"/usr/bin/{name}")
# Create a fake uv binary in a location that _find_uv() probes
fake_uv = tmp_path / "uv"
fake_uv.write_text("#!/bin/sh\necho uv")
fake_uv.chmod(0o755)
# Patch _find_uv's candidate list so the temp path is probed
import os as _os
original_exists = _os.path.exists
original_access = _os.access
def fake_exists(path):
if path == str(fake_uv):
return True
if path.endswith("/uv"):
return False # suppress all real candidates
return original_exists(path)
def fake_access(path, mode):
if path == str(fake_uv):
return True
return original_access(path, mode)
monkeypatch.setattr(_os.path, "exists", fake_exists)
monkeypatch.setattr(_os, "access", fake_access)
# Temporarily inject fake_uv as the first candidate to probe
original_find_uv = cli_mod._find_uv
def patched_find_uv():
found = shutil.which("uv")
if found:
return found
candidates = [str(fake_uv)]
for path in candidates:
if _os.path.exists(path) and _os.access(path, _os.X_OK):
return path
return None
monkeypatch.setattr(cli_mod, "_find_uv", patched_find_uv)
result = cli_mod._find_uv()
assert result == str(fake_uv), (
f"_find_uv must return the candidate path when shutil.which returns None; got {result!r}"
)
def test_find_uv_returns_none_when_no_candidate_exists(monkeypatch):
"""_find_uv() returns None when neither shutil.which nor any candidate finds uv."""
import os as _os
import muxplex.cli as cli_mod
monkeypatch.setattr(shutil, "which", lambda name: None)
monkeypatch.setattr(_os.path, "exists", lambda path: False)
monkeypatch.setattr(_os, "access", lambda path, mode: False)
result = cli_mod._find_uv()
assert result is None, "_find_uv must return None when uv cannot be found anywhere"
def test_find_pip_returns_path_from_shutil_which():
"""_find_pip() returns the path that shutil.which('pip') returns when present."""
import muxplex.cli as cli_mod
with patch("muxplex.cli.shutil") as mock_shutil:
mock_shutil.which.side_effect = lambda name: (
"/usr/bin/pip" if name == "pip" else None
)
result = cli_mod._find_pip()
assert result == "/usr/bin/pip", (
"_find_pip must return shutil.which('pip') result when pip is on PATH"
)
def test_find_pip_returns_pip3_when_pip_absent():
"""_find_pip() returns pip3 path when pip is absent but pip3 is on PATH."""
import muxplex.cli as cli_mod
with patch("muxplex.cli.shutil") as mock_shutil:
mock_shutil.which.side_effect = lambda name: (
"/usr/bin/pip3" if name == "pip3" else None
)
result = cli_mod._find_pip()
assert result == "/usr/bin/pip3", (
"_find_pip must return pip3 when pip is absent but pip3 is on PATH"
)
def test_find_pip_probes_known_locations_when_which_returns_none(monkeypatch):
"""_find_pip() falls through to the candidate list when shutil.which returns None."""
import os as _os
import muxplex.cli as cli_mod
monkeypatch.setattr(shutil, "which", lambda name: None)
import muxplex.cli as cli_mod
def patched_find_pip():
for name in ("pip", "pip3"):
found = shutil.which(name)
if found:
return found
# Simulate exactly one candidate existing
candidate = "/snap/bin/pip3"
if _os.path.exists(candidate) and _os.access(candidate, _os.X_OK):
return candidate
return None
monkeypatch.setattr(_os.path, "exists", lambda p: p == "/snap/bin/pip3")
monkeypatch.setattr(_os, "access", lambda p, m: p == "/snap/bin/pip3")
monkeypatch.setattr(cli_mod, "_find_pip", patched_find_pip)
result = cli_mod._find_pip()
assert result == "/snap/bin/pip3", (
f"_find_pip must return the candidate path from known locations; got {result!r}"
)
def test_find_pip_returns_none_when_no_candidate_exists(monkeypatch):
"""_find_pip() returns None when neither shutil.which nor any candidate finds pip."""
import os as _os
import muxplex.cli as cli_mod
monkeypatch.setattr(shutil, "which", lambda name: None)
monkeypatch.setattr(_os.path, "exists", lambda path: False)
monkeypatch.setattr(_os, "access", lambda path, mode: False)
result = cli_mod._find_pip()
assert result is None, "_find_pip must return None when pip cannot be found anywhere"
def test_upgrade_uses_find_uv_not_shutil_which(monkeypatch, capsys):
"""upgrade() calls _find_uv() to locate uv — not shutil.which('uv') directly.
When shutil.which('uv') returns None but _find_uv() returns a path found via
the known-locations probe (e.g. /snap/bin/uv on a snap-installed system), the
uv branch must still be taken pip must NOT be used.
"""
import subprocess
import sys
import muxplex.cli as cli_mod
calls: list = []
def mock_run(cmd, **kwargs):
calls.append(list(cmd) if isinstance(cmd, list) else cmd)
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
# shutil.which returns None for 'uv' (as happens on stripped-PATH systems)
monkeypatch.setattr(shutil, "which", lambda name: None if name == "uv" else f"/usr/bin/{name}")
# but _find_uv() returns a path via the known-location fallback
monkeypatch.setattr(cli_mod, "_find_uv", lambda: "/snap/bin/uv")
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(cli_mod, "doctor", lambda: None)
monkeypatch.setattr(
cli_mod,
"_check_for_update",
lambda info: (True, "update available"),
)
monkeypatch.setattr(sys, "platform", "linux")
with patch("muxplex.service.service_install", lambda: None):
cli_mod.upgrade()
uv_calls = [c for c in calls if isinstance(c, list) and c and "/snap/bin/uv" in c[0]]
assert len(uv_calls) > 0, (
"upgrade() must invoke the uv binary found by _find_uv() even when shutil.which returns None"
)
pip_calls = [c for c in calls if isinstance(c, list) and c and "pip" in str(c[0])]
assert len(pip_calls) == 0, (
"upgrade() must NOT fall back to pip when _find_uv() returns a valid path"
)
def test_upgrade_exits_1_after_finally_recovers_stopped_service(monkeypatch, capsys):
"""upgrade() propagates install failure as exit code 1 even after try/finally restarts service.
Scenario: pip install fails (rc != 0) but the service restart in the finally
block succeeds. The user-visible behaviour must be:
1. Error message printed.
2. Service restarted (best-effort).
3. Process exits with code 1 so callers / automation can detect the failure.
"""
import subprocess
import sys
import muxplex.cli as cli_mod
restart_called = []
def mock_run(cmd, **kwargs):
cmd_list = list(cmd) if isinstance(cmd, list) else [cmd]
# Simulate pip install failing
if cmd_list and "pip" in str(cmd_list[0]):
return type("R", (), {"returncode": 1, "stdout": "", "stderr": "pip install failed"})()
# Simulate all other subprocess calls succeeding (systemctl is-active, start, etc.)
if cmd_list and any(k in str(cmd_list) for k in ("is-active", "start", "daemon-reload", "is-enabled")):
restart_called.append(cmd_list)
return type("R", (), {"returncode": 0, "stdout": "active", "stderr": ""})()
return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})()
# uv absent so we reach the pip path
monkeypatch.setattr(cli_mod, "_find_uv", lambda: None)
monkeypatch.setattr(cli_mod, "_find_pip", lambda: "/usr/bin/pip")
monkeypatch.setattr(shutil, "which", lambda name: (
"/usr/bin/systemctl" if name == "systemctl" else None
))
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(
cli_mod,
"_check_for_update",
lambda info: (True, "update available"),
)
monkeypatch.setattr(sys, "platform", "linux")
with pytest.raises(SystemExit) as exc_info:
cli_mod.upgrade()
assert exc_info.value.code == 1, (
f"upgrade() must exit with code 1 when install fails; got code {exc_info.value.code}"
)
out = capsys.readouterr().out
assert "error" in out.lower() or "failed" in out.lower(), (
f"upgrade() must print an error message when install fails; got: {out!r}"
)
# ---------------------------------------------------------------------------
# v0.6.7 fixes — service-restart verification (Fix 1)
# ---------------------------------------------------------------------------
def test_verify_service_started_returns_true_when_active(monkeypatch):
"""_verify_service_started returns True when systemctl is-active exits 0 (active)."""
import subprocess
import muxplex.cli as cli_mod
monkeypatch.setattr(cli_mod, "_have_systemctl", lambda: True)
monkeypatch.setattr(cli_mod, "_have_launchctl", lambda: False)
monkeypatch.setattr(
subprocess,
"run",
lambda cmd, **kw: type(
"R", (), {"returncode": 0, "stdout": "active\n", "stderr": ""}
)(),
)
assert cli_mod._verify_service_started() is True
def test_verify_service_started_returns_false_when_inactive(monkeypatch):
"""_verify_service_started returns False when systemctl is-active exits 3 (inactive)."""
import subprocess
import muxplex.cli as cli_mod
monkeypatch.setattr(cli_mod, "_have_systemctl", lambda: True)
monkeypatch.setattr(cli_mod, "_have_launchctl", lambda: False)
monkeypatch.setattr(
subprocess,
"run",
lambda cmd, **kw: type(
"R", (), {"returncode": 3, "stdout": "inactive\n", "stderr": ""}
)(),
)
assert cli_mod._verify_service_started() is False
def test_upgrade_exits_1_if_service_fails_to_restart(monkeypatch, capsys):
"""upgrade() exits 1 when install succeeds but the service never becomes active."""
import subprocess
import muxplex.cli as cli_mod
calls = []
def mock_run(cmd, **kwargs):
calls.append(list(cmd) if isinstance(cmd, list) else cmd)
return type("R", (), {"returncode": 0, "stdout": "enabled\n", "stderr": ""})()
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(shutil, "which", lambda name: f"/usr/bin/{name}")
monkeypatch.setattr(cli_mod, "_check_for_update", lambda info: (True, "update available"))
monkeypatch.setattr(cli_mod, "_have_systemctl", lambda: True)
monkeypatch.setattr(cli_mod, "_have_launchctl", lambda: False)
# Service never becomes active (simulates the spark-1 dead-service scenario)
monkeypatch.setattr(cli_mod, "_verify_service_started", lambda timeout_s=10: False)
with patch("muxplex.service.service_install", lambda: None):
with pytest.raises(SystemExit) as exc_info:
cli_mod.upgrade()
assert exc_info.value.code == 1, (
f"upgrade() must exit 1 when service fails to restart; got {exc_info.value.code}"
)
out = capsys.readouterr().out
assert "error" in out.lower() or "not running" in out.lower(), (
f"upgrade() must print an error about the failed restart; got: {out!r}"
)
def test_upgrade_calls_daemon_reload_before_start(monkeypatch, capsys):
"""upgrade() calls systemctl daemon-reload before start (stale unit-file fix)."""
import subprocess
import muxplex.cli as cli_mod
calls: list = []
def mock_run(cmd, **kwargs):
calls.append(list(cmd) if isinstance(cmd, list) else cmd)
return type("R", (), {"returncode": 0, "stdout": "enabled\n", "stderr": ""})()
monkeypatch.setattr(subprocess, "run", mock_run)
monkeypatch.setattr(shutil, "which", lambda name: f"/usr/bin/{name}")
monkeypatch.setattr(cli_mod, "_check_for_update", lambda info: (True, "update available"))
monkeypatch.setattr(cli_mod, "_have_systemctl", lambda: True)
monkeypatch.setattr(cli_mod, "_have_launchctl", lambda: False)
monkeypatch.setattr(cli_mod, "_verify_service_started", lambda timeout_s=10: True)
monkeypatch.setattr(cli_mod, "doctor", lambda: None)
with patch("muxplex.service.service_install", lambda: None):
cli_mod.upgrade()
systemctl_calls = [c for c in calls if isinstance(c, list) and "systemctl" in c]
reload_idx = next(
(i for i, c in enumerate(systemctl_calls) if "daemon-reload" in c), None
)
start_idx = next(
(i for i, c in enumerate(systemctl_calls) if "start" in c and "muxplex" in c),
None,
)
assert reload_idx is not None, (
"systemctl daemon-reload must be called during upgrade"
)
assert start_idx is not None, (
"systemctl start muxplex must be called during upgrade"
)
assert reload_idx < start_idx, (
"daemon-reload must be called BEFORE start to pick up the regenerated unit file"
)
# ---------------------------------------------------------------------------
# v0.6.7 fixes — doctor launchd port-probe (Fix 2 / doctor enhancement)
# ---------------------------------------------------------------------------
def test_doctor_reports_launchd_registered_but_not_serving(
monkeypatch, tmp_path, capsys
):
"""doctor() warns when launchd agent is registered but the service port is not responding."""
import subprocess
import sys
import muxplex.cli as cli_mod
import muxplex.settings as settings_mod
# Create plist file so plist.exists() passes
fake_home = tmp_path
plist = fake_home / "Library" / "LaunchAgents" / "com.muxplex.plist"
plist.parent.mkdir(parents=True)
plist.write_text("<plist/>")
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
settings_file = tmp_path / "settings.json"
settings_file.write_text("{}")
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", settings_file)
# Simulate macOS
monkeypatch.setattr(sys, "platform", "darwin")
monkeypatch.setattr(cli_mod, "_have_launchctl", lambda: True)
# launchctl print succeeds (agent is registered)
monkeypatch.setattr(
subprocess,
"run",
lambda *a, **kw: type(
"R", (), {"returncode": 0, "stdout": "", "stderr": ""}
)(),
)
# Port is NOT responding
monkeypatch.setattr(cli_mod, "_probe_service_port", lambda port: False)
cli_mod.doctor()
out = capsys.readouterr().out
assert "not serving" in out.lower(), (
f"doctor() must warn 'not serving' when launchd is registered but port is down;"
f" got: {out!r}"
)
+177
View File
@@ -0,0 +1,177 @@
"""
Tests for the cache-busting version suffix on static asset URLs served by index_page().
Verifies that GET / (the main dashboard) injects ?v=<version> on every
<script src=""> and <link href=""> URL so browsers pick up new code
immediately after each release, rather than serving stale JS/CSS from
the HTTP cache.
"""
import importlib.metadata
import pytest
from bs4 import BeautifulSoup
from fastapi.testclient import TestClient
from muxplex.main import app
# ---------------------------------------------------------------------------
# Shared fixtures (mirror test_api.py setup so tests run cleanly in isolation)
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def patch_startup_and_state(tmp_path, monkeypatch):
"""Redirect state/PID files to tmp_path and stub out long-running startup tasks."""
tmp_state_dir = tmp_path / "state"
tmp_state_path = tmp_state_dir / "state.json"
monkeypatch.setattr("muxplex.state.STATE_DIR", tmp_state_dir)
monkeypatch.setattr("muxplex.state.STATE_PATH", tmp_state_path)
tmp_pid_dir = tmp_path / "ttyd"
tmp_pid_path = tmp_pid_dir / "ttyd.pid"
monkeypatch.setattr("muxplex.ttyd.TTYD_PID_DIR", tmp_pid_dir)
monkeypatch.setattr("muxplex.ttyd.TTYD_PID_PATH", tmp_pid_path)
async def _mock_kill_orphan():
return False
monkeypatch.setattr("muxplex.main.kill_orphan_ttyd", _mock_kill_orphan)
async def noop_poll_loop() -> None:
pass
monkeypatch.setattr("muxplex.main._poll_loop", noop_poll_loop)
@pytest.fixture(autouse=True)
def reset_federation_cache():
"""Clear _federation_cache before and after each test."""
import muxplex.main as main_mod
main_mod._federation_cache.clear()
yield
main_mod._federation_cache.clear()
@pytest.fixture
def client(monkeypatch):
"""Authenticated TestClient with the app lifespan active."""
monkeypatch.setenv("MUXPLEX_PASSWORD", "test-password")
with TestClient(app) as c:
from muxplex.auth import create_session_cookie
from muxplex.main import _auth_secret, _auth_ttl
cookie = create_session_cookie(_auth_secret, _auth_ttl)
c.cookies.set("muxplex_session", cookie)
yield c
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _get_index_soup(client) -> BeautifulSoup:
response = client.get("/")
assert response.status_code == 200, f"GET / returned {response.status_code}"
return BeautifulSoup(response.text, "html.parser")
# ---------------------------------------------------------------------------
# Test 1 — every script src and link href carries the version suffix
# ---------------------------------------------------------------------------
def test_index_all_asset_urls_have_version_suffix(client):
"""GET / must inject ?v=<version> on every <script src> and <link href> asset URL.
Regression guard for the "is the user seeing stale JS?" investigation:
verifies that the standard HTTP cache is busted on every release by
appending a version query parameter to each static asset reference.
"""
version = importlib.metadata.version("muxplex")
soup = _get_index_soup(client)
# All <script src="…"> tags
script_tags = soup.find_all("script", src=True)
assert len(script_tags) >= 7, (
f"Expected at least 7 <script src> tags, found {len(script_tags)}"
)
for tag in script_tags:
src = tag["src"]
assert f"?v={version}" in src, (
f"<script src> missing ?v={version} suffix: {src!r}"
)
# All <link href="…"> tags
link_tags = soup.find_all("link", href=True)
assert len(link_tags) >= 1, "Expected at least one <link href> tag"
for tag in link_tags:
href = tag["href"]
assert f"?v={version}" in href, (
f"<link href> missing ?v={version} suffix: {href!r}"
)
# ---------------------------------------------------------------------------
# Test 2 — vendor scripts are individually versioned (not just app.js)
# ---------------------------------------------------------------------------
def test_index_vendor_scripts_each_versioned(client):
"""All five vendor JS bundles must carry the version suffix, not just app.js.
The browser-tester on spark-1 observed bare vendor URLs. This test
ensures that xterm.js and its addons are cache-busted alongside the
first-party scripts.
"""
version = importlib.metadata.version("muxplex")
soup = _get_index_soup(client)
script_srcs = [tag["src"] for tag in soup.find_all("script", src=True)]
expected_versioned = [
f"/vendor/xterm.js?v={version}",
f"/vendor/xterm-addon-fit.js?v={version}",
f"/vendor/xterm-addon-web-links.js?v={version}",
f"/vendor/xterm-addon-search.js?v={version}",
f"/vendor/addon-image.js?v={version}",
f"/app.js?v={version}",
f"/terminal.js?v={version}",
]
for expected in expected_versioned:
assert expected in script_srcs, (
f"Expected versioned script {expected!r}; found srcs: {script_srcs}"
)
# ---------------------------------------------------------------------------
# Test 3 — versioned asset URLs still resolve to the actual static files
# ---------------------------------------------------------------------------
def test_versioned_asset_url_resolves_to_static_file(client):
"""GET /app.js?v=<version> must return HTTP 200 (static handler ignores query string).
Sanity check: adding the version suffix must not break asset loading.
Starlette's StaticFiles handler ignores query parameters when looking up
files on disk, so the versioned URL must serve identically to the bare URL.
"""
version = importlib.metadata.version("muxplex")
# First-party assets
for path in ("/app.js", "/terminal.js", "/style.css"):
url = f"{path}?v={version}"
resp = client.get(url)
assert resp.status_code == 200, (
f"Versioned URL {url!r} returned {resp.status_code}, expected 200"
)
# Vendor asset
vendor_url = f"/vendor/xterm.js?v={version}"
resp = client.get(vendor_url)
assert resp.status_code == 200, (
f"Versioned vendor URL {vendor_url!r} returned {resp.status_code}, expected 200"
)
+48
View File
@@ -667,3 +667,51 @@ def test_service_install_hides_tls_tip_on_localhost(capsys, tmp_path, monkeypatc
assert "muxplex setup-tls" not in out, (
f"TLS tip must NOT appear in service install output when host is 127.0.0.1, got: {out!r}"
)
# ---------------------------------------------------------------------------
# v0.6.7 fix — launchd plist ProgramArguments must use separate <string> tokens
# ---------------------------------------------------------------------------
def test_launchd_plist_program_arguments_are_separate_strings(monkeypatch, tmp_path):
"""_launchd_install emits each argv token as its own <string> in ProgramArguments.
The v0.6.6 bug: a single <string> containing e.g.
"python3 -m muxplex" caused launchd to look for a literal executable
named "python3 -m muxplex" (with spaces) which doesn't exist — so the
daemon silently failed to start on every boot.
"""
import os
import plistlib
import muxplex.service as svc
plist_dir = tmp_path / "LaunchAgents"
plist_path = plist_dir / "com.muxplex.plist"
monkeypatch.setattr(svc, "_LAUNCHD_PLIST_DIR", plist_dir)
monkeypatch.setattr(svc, "_LAUNCHD_PLIST_PATH", plist_path)
monkeypatch.setattr(os, "getuid", lambda: 501)
monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: None)
monkeypatch.setattr(svc, "_prompt_host_if_localhost", lambda: None)
monkeypatch.setattr(svc, "_show_tls_nudge_if_needed", lambda: None)
svc._launchd_install()
assert plist_path.exists(), "plist file must be written by _launchd_install"
plist_data = plistlib.loads(plist_path.read_bytes())
prog_args = plist_data.get("ProgramArguments", [])
assert len(prog_args) >= 2, (
f"ProgramArguments must have at least 2 elements, got: {prog_args!r}"
)
assert prog_args[-1] == "serve", (
f"Last ProgramArguments element must be 'serve', got: {prog_args!r}"
)
for arg in prog_args:
assert " " not in arg, (
f"ProgramArguments element must not contain spaces "
f"(embedded-space arg trap): {arg!r} in {prog_args!r}"
)
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "muxplex"
version = "0.6.3"
version = "0.6.7"
description = "Web-based tmux session dashboard — access all your tmux sessions from any browser"
readme = "README.md"
license = { text = "MIT" }