aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-cli/tests
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-10-15 17:42:30 +0200
committerChristophe Besson <cbesson@gmail.com>2026-10-15 17:45:41 +0200
commit41d03f0a35a95daa7d54e7f29ab924776bedf6e4 (patch)
tree2e631013746f913413b9811562b9c9a4734a0e41 /packages/meshbay-cli/tests
parent39cac2878de059992cb09a1ccf09b6bcece99fae (diff)
downloadmeshbay-41d03f0a35a95daa7d54e7f29ab924776bedf6e4.tar.gz
fix(cli): make the command-line client work on Windows
Its suite had only run on Linux. On Windows, against a real hub and node and from the installed meshbay-cli.exe: - secrets.enc was written in text mode (os.open without O_BINARY): every 0x0A of the ciphertext became CR LF and the saved session "does not open with its key" on the next run. - isatty() is true for NUL, so a script run with `< NUL` or by the Task Scheduler was asked questions instead of told to pass --yes; output.terminal() checks for a real console. - Ctrl+C waited for the selector loop's next timer, up to half a minute in `chat --follow`; the loop now points the signal wakeup at its self-pipe, as the proactor does. - Right-click copy and paste raised: asyncio cannot start a process on the selector loop; the clipboard tool runs in a thread. - Tab completion of remote paths ran on the default (proactor) loop, which never opens a node, and wrote in the locale encoding; the PowerShell script now reads it as UTF-8. - `get -R` into a folder past MAX_PATH died with a traceback; get, put, the resume file and the interactive mode use the extended form at each disk call, as the node does. - `groups` showed a SQLite hub's naive UTC time as local time. The opt-in end-to-end suite now runs on Windows (16 tests: Chrome found under Program Files and started with the real environment, VLC as the player, Ctrl+C through a console of its own, PowerShell 5.1 completion of an accented name). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-cli/tests')
-rw-r--r--packages/meshbay-cli/tests/console.py48
-rw-r--r--packages/meshbay-cli/tests/test_cli_actions.py63
-rw-r--r--packages/meshbay-cli/tests/test_clipboard.py23
-rw-r--r--packages/meshbay-cli/tests/test_ctrl_c.py34
-rw-r--r--packages/meshbay-cli/tests/test_e2e_local.py118
-rw-r--r--packages/meshbay-cli/tests/test_listing.py18
6 files changed, 262 insertions, 42 deletions
diff --git a/packages/meshbay-cli/tests/console.py b/packages/meshbay-cli/tests/console.py
new file mode 100644
index 0000000..d3ea6c9
--- /dev/null
+++ b/packages/meshbay-cli/tests/console.py
@@ -0,0 +1,48 @@
+"""
+A person's Ctrl+C, delivered to a client run started by a test.
+
+On Windows Ctrl+C is a console event, not a signal, so the run gets a hidden
+console of its own and a helper attached to that console raises the event. The
+run goes through a launcher that turns Ctrl+C handling back on first: a process
+inherits "ignore Ctrl+C" from its parent, and whatever started the test suite
+may have set it (an IDE, a CI runner, an agent's shell) where a person's console
+never does — measured, the event then reached nothing at all.
+"""
+
+import subprocess
+import sys
+
+_LAUNCHER = ("import ctypes, signal, subprocess, sys\n"
+ "ctypes.windll.kernel32.SetConsoleCtrlHandler(None, 0)\n"
+ "signal.signal(signal.SIGINT, signal.SIG_IGN)\n"
+ "sys.exit(subprocess.call(sys.argv[1:]))\n")
+
+# os._exit: with the console swapped, flushing stdout at exit fails (code 120).
+_SENDER = ("import ctypes, os, sys\n"
+ "k = ctypes.windll.kernel32\n"
+ "k.FreeConsole()\n"
+ "ok = k.AttachConsole(int(sys.argv[1])) and k.SetConsoleCtrlHandler(None, 1)\n"
+ "os._exit(0 if ok and k.GenerateConsoleCtrlEvent(0, 0) else 1)\n")
+
+
+def interruptible(argv: list[str], env: dict | None) -> subprocess.Popen:
+ """`argv` started so that `ctrl_c` reaches it, its output read as UTF-8."""
+ extra: dict = {}
+ if sys.platform == "win32":
+ si = subprocess.STARTUPINFO()
+ si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
+ si.wShowWindow = 0 # SW_HIDE
+ extra = {"creationflags": subprocess.CREATE_NEW_CONSOLE, "startupinfo": si}
+ argv = [sys.executable, "-c", _LAUNCHER, *argv]
+ return subprocess.Popen(argv, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+ text=True, encoding="utf-8", **extra)
+
+
+def ctrl_c(proc: subprocess.Popen) -> None:
+ if sys.platform != "win32":
+ import signal
+ proc.send_signal(signal.SIGINT)
+ return
+ subprocess.run([sys.executable, "-c", _SENDER, str(proc.pid)], check=True, timeout=30,
+ stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+ creationflags=subprocess.CREATE_NO_WINDOW)
diff --git a/packages/meshbay-cli/tests/test_cli_actions.py b/packages/meshbay-cli/tests/test_cli_actions.py
index 006c183..ad914fd 100644
--- a/packages/meshbay-cli/tests/test_cli_actions.py
+++ b/packages/meshbay-cli/tests/test_cli_actions.py
@@ -83,6 +83,17 @@ def test_no_passphrase_and_no_terminal_exits_2(hub, monkeypatch, capsys):
assert "MESHBAY_PASSWORD" in capsys.readouterr().err
+def test_the_null_device_is_no_terminal(hub, monkeypatch, capsys):
+ """`< NUL` on Windows: a character device, which isatty() calls a terminal.
+ A script run so was asked for a passphrase it could not type."""
+ from meshbay_cli.output import terminal
+ with open(os.devnull) as null:
+ assert not terminal(null)
+ monkeypatch.setattr("sys.stdin", null)
+ assert main("--hub", hub.url, "-u", USERNAME, "login") == 2
+ assert "MESHBAY_PASSWORD" in capsys.readouterr().err
+
+
async def _session():
return FakeSession()
@@ -177,6 +188,58 @@ def test_put_into_a_read_only_folder_sends_nothing(node, tmp_path, capsys):
assert "read-only" in capsys.readouterr().err
+def _max_path_in_force() -> bool:
+ if sys.platform != "win32":
+ return False
+ import winreg
+ try:
+ with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE,
+ r"SYSTEM\CurrentControlSet\Control\FileSystem") as key:
+ return winreg.QueryValueEx(key, "LongPathsEnabled")[0] == 0
+ except OSError:
+ return True
+
+
+@pytest.mark.skipif(not _max_path_in_force(), reason="Windows with MAX_PATH in force")
+def test_get_and_put_past_max_path(node, tmp_path, capsys):
+ """Windows refuses a plain path of 260 characters or more. The node uses the
+ long form at each disk call, and the client has to as well: `get -R` into a
+ deep folder died with a traceback creating its first folder."""
+ import shutil
+
+ from meshbay_common.paths import long_path
+ segment = "A folder name as long as the ones a real library has"
+ deep = tmp_path / "deep"
+ while len(str(deep)) < 240:
+ deep = deep / segment
+ try:
+ node.add("share/2024", "a.jpg", b"A" * 3000)
+ assert main("get", "-R", "Family/share/2024", "-o", f"{deep}{os.sep}") == 0, \
+ capsys.readouterr().err
+ got = deep / "2024" / "a.jpg"
+ assert len(str(got)) >= 260
+ with open(long_path(got), "rb") as f:
+ assert f.read() == b"A" * 3000
+
+ # A part left from an interrupted run, resumed past the limit too.
+ node.add("share/2024", "b.bin", b"B" * 10)
+ with open(long_path(deep / "2024" / "b.bin.meshbay-part"), "wb") as f:
+ f.write(b"B")
+ assert main("get", "-R", "Family/share/2024", "-o", f"{deep}{os.sep}", "-n") == 0
+ with open(long_path(deep / "2024" / "b.bin"), "rb") as f:
+ assert f.read() == b"B" * 10
+
+ # And a tree past the limit, sent.
+ src = deep / "Scans"
+ os.makedirs(long_path(src / segment))
+ with open(long_path(src / segment / "page.pdf"), "wb") as f:
+ f.write(b"%PDF")
+ assert main("put", "-R", str(src), "Family/share") == 0, capsys.readouterr().err
+ assert node.stored(f"share/Scans/{segment}/page.pdf") == b"%PDF"
+ finally:
+ shutil.rmtree("\\\\?\\" + str(tmp_path / "deep"), ignore_errors=True)
+
+
def test_trust_lists_and_forgets_pins(hub, capsys):
assert main("trust", "list") == 0 # nothing stored: nothing pinned
assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE,
diff --git a/packages/meshbay-cli/tests/test_clipboard.py b/packages/meshbay-cli/tests/test_clipboard.py
index 6296317..20e90b6 100644
--- a/packages/meshbay-cli/tests/test_clipboard.py
+++ b/packages/meshbay-cli/tests/test_clipboard.py
@@ -10,7 +10,7 @@ import time
import pytest
from meshbay_cli.tui import clipboard
-pytestmark = pytest.mark.skipif(sys.platform in ("win32", "darwin"),
+unix_tools = pytest.mark.skipif(sys.platform in ("win32", "darwin"),
reason="the Wayland and X11 tools")
# Found before a test leaves only the fake tools on PATH.
CAT, SLEEP = shutil.which("cat"), shutil.which("sleep")
@@ -31,12 +31,14 @@ def wayland(tmp_path, monkeypatch):
return held
+@unix_tools
async def test_what_is_copied_is_what_is_pasted(wayland):
assert await clipboard.write_text("héllo — 你好\nsecond") is True
assert wayland.read_text(encoding="utf-8") == "héllo — 你好\nsecond"
assert await clipboard.read_text() == "héllo — 你好\nsecond"
+@unix_tools
async def test_no_tool_is_none_not_an_error(tmp_path, monkeypatch):
monkeypatch.setenv("PATH", str(tmp_path))
monkeypatch.setenv("WAYLAND_DISPLAY", "wayland-0")
@@ -45,14 +47,33 @@ async def test_no_tool_is_none_not_an_error(tmp_path, monkeypatch):
assert await clipboard.write_text("x") is False
+@unix_tools
async def test_a_failing_tool_is_skipped(wayland):
(wayland.parent / "bin" / "wl-paste").write_text("#!/bin/sh\nexit 1\n")
assert await clipboard.read_text() is None
+@unix_tools
async def test_a_hanging_tool_is_given_up(wayland, monkeypatch):
(wayland.parent / "bin" / "wl-paste").write_text(f"#!/bin/sh\nexec {SLEEP} 30\n")
monkeypatch.setattr(clipboard, "TIMEOUT", 0.3)
started = time.monotonic()
assert await clipboard.read_text() is None
assert 0.3 <= time.monotonic() - started < 5
+
+
+def test_the_tools_run_on_a_selector_loop(monkeypatch):
+ """The client's loop on Windows (`__main__.loop_factory`), which cannot start
+ a process through asyncio: right-click copy and paste raised there."""
+ import asyncio
+
+ def tools(read):
+ return [[sys.executable, "-c", "print('pasted', end='')" if read
+ else "import sys; sys.stdin.buffer.read()"]]
+ monkeypatch.setattr(clipboard, "_tools", tools)
+ loop = asyncio.SelectorEventLoop()
+ try:
+ assert loop.run_until_complete(clipboard.read_text()) == "pasted"
+ assert loop.run_until_complete(clipboard.write_text("x")) is True
+ finally:
+ loop.close()
diff --git a/packages/meshbay-cli/tests/test_ctrl_c.py b/packages/meshbay-cli/tests/test_ctrl_c.py
new file mode 100644
index 0000000..53a798a
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_ctrl_c.py
@@ -0,0 +1,34 @@
+"""
+Ctrl+C stops the client at once, whatever its loop is waiting for.
+
+On Windows the client runs a selector loop (aiortc's ICE needs one), and a
+console event does not interrupt select(): Ctrl+C waited for the loop's next
+timer — half a minute in `chat --follow` between pings.
+"""
+
+import sys
+import time
+
+from console import ctrl_c, interruptible
+
+IDLE = ("import asyncio, sys\n"
+ "from meshbay_cli.__main__ import loop_factory\n"
+ "print('ready', flush=True)\n"
+ "try:\n"
+ " asyncio.run(asyncio.sleep(60), loop_factory=loop_factory())\n"
+ "except KeyboardInterrupt:\n"
+ " sys.exit(130)\n")
+
+
+def test_ctrl_c_wakes_an_idle_client_at_once():
+ proc = interruptible([sys.executable, "-c", IDLE], None)
+ try:
+ assert proc.stdout.readline().strip() == "ready", proc.stderr.read()
+ started = time.monotonic()
+ ctrl_c(proc)
+ proc.wait(timeout=30)
+ assert proc.returncode == 130, proc.stderr.read()
+ assert time.monotonic() - started < 5
+ finally:
+ if proc.poll() is None:
+ proc.kill()
diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py
index 6b7869b..b131658 100644
--- a/packages/meshbay-cli/tests/test_e2e_local.py
+++ b/packages/meshbay-cli/tests/test_e2e_local.py
@@ -42,6 +42,7 @@ from pathlib import Path
import httpx
import pytest
+from console import ctrl_c, interruptible
pytestmark = [
pytest.mark.skipif(os.environ.get("MESHBAY_CLI_E2E") != "1",
@@ -53,6 +54,28 @@ USERNAME = "alice-e2e-cli"
PASSPHRASE = "a long enough test passphrase, nobody's"
GROUP_NAME = "Some Saga/Box @Set"
BIN = Path(sys.executable).parent
+# The environment before conftest points HOME, USERPROFILE and LOCALAPPDATA at an
+# empty directory for each test: Chrome on Windows does not start in one.
+REAL_ENV = dict(os.environ)
+# Windows refuses a control character in a file name, so a node there cannot
+# hold the name with an escape character in it.
+ESCAPE_NAME = sys.platform != "win32"
+
+
+def _program(linux: str, *windows: str) -> str | None:
+ """A program by its Linux name, or where Windows installs it."""
+ found = shutil.which(linux)
+ if found or sys.platform != "win32":
+ return found
+ for base in (os.environ.get("ProgramFiles"), os.environ.get("ProgramFiles(x86)")):
+ candidate = os.path.join(base or "", *windows)
+ if base and os.path.isfile(candidate):
+ return candidate
+ return None
+
+
+CHROME = _program("google-chrome", "Google", "Chrome", "Application", "chrome.exe")
+VLC = _program("cvlc", "VideoLAN", "VLC", "vlc.exe")
def _free_port() -> int:
@@ -139,8 +162,13 @@ def _build(root: Path, procs: list) -> dict:
(share / "2024" / "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac")\
.write_bytes(secrets.token_bytes(2_500_000))
(share / "notes.txt").write_text("hello\n")
- (share / "evil\x1b[31mred.txt").write_text("escape\n")
- node_env = {**os.environ, "HOME": str(node_home),
+ if ESCAPE_NAME:
+ (share / "evil\x1b[31mred.txt").write_text("escape\n")
+ # USERPROFILE, LOCALAPPDATA and APPDATA: where the node's directories
+ # are on Windows; HOME alone would leave it in the developer's own.
+ node_env = {**os.environ, "HOME": str(node_home), "USERPROFILE": str(node_home),
+ "LOCALAPPDATA": str(node_home / "AppData" / "Local"),
+ "APPDATA": str(node_home / "AppData" / "Roaming"),
"XDG_CONFIG_HOME": str(node_home / ".config"),
"XDG_DATA_HOME": str(node_home / ".local/share"),
"XDG_STATE_HOME": str(node_home / ".local/state")}
@@ -398,7 +426,7 @@ def cli(world: dict, *args: str, stdin: str | None = None,
env.update(env_extra or {})
exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli")
return subprocess.run([exe, *args], env=env, input=stdin,
- capture_output=True, text=True, timeout=180)
+ capture_output=True, text=True, encoding="utf-8", timeout=180)
def _handle() -> str:
@@ -438,7 +466,8 @@ def test_the_cli_against_a_real_hub_and_node(world):
out = r.stdout
assert "plage de l'ouest.jpg" in out
assert "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac" in out
- assert "\x1b" not in out and "evil\\x1b[31mred.txt" in out
+ assert "\x1b" not in out
+ assert not ESCAPE_NAME or "evil\\x1b[31mred.txt" in out
assert "été à la mer/" in out
# By name alone, case-insensitively, and as JSON.
@@ -467,8 +496,9 @@ def test_the_cli_against_a_real_hub_and_node(world):
assert (got / "2024" / "été à la mer" / "plage de l'ouest.jpg").read_bytes() == \
(world["share"] / "2024" / "été à la mer" / "plage de l'ouest.jpg").read_bytes()
# The escape character could not be written as it is: renamed, and said so.
- assert "evil\\x1b[31mred.txt is saved as evil_[31mred.txt" in r.stderr
- assert (got / "evil_[31mred.txt").exists()
+ if ESCAPE_NAME:
+ assert "evil\\x1b[31mred.txt is saved as evil_[31mred.txt" in r.stderr
+ assert (got / "evil_[31mred.txt").exists()
r = cli(world, "get", f"{_handle()}/share/notes.txt", "-o", "-")
assert r.returncode == 0 and r.stdout == "hello\n"
@@ -486,7 +516,8 @@ def test_the_cli_against_a_real_hub_and_node(world):
# notes.txt is there already, whatever the case: nothing sent.
r = cli(world, "put", str(local / "Notes.TXT"), f"{_handle()}/share")
assert r.returncode == 8 and "already there" in r.stderr
- assert not (world["share"] / "Notes.TXT").exists()
+ # By listing, not exists(): on Windows that finds notes.txt under any case.
+ assert "Notes.TXT" not in os.listdir(world["share"])
r = cli(world, "put", "-", f"{_handle()}/share/piped.txt", stdin="through stdin")
assert r.returncode == 0, r.stderr
assert (world["share"] / "piped.txt").read_text() == "through stdin"
@@ -739,10 +770,10 @@ class _Chrome:
ignore_cleanup_errors=True)
port = _free_port()
self.proc = subprocess.Popen(
- ["google-chrome", "--headless=new", "--disable-gpu", "--no-sandbox",
+ [CHROME, "--headless=new", "--disable-gpu", "--no-sandbox",
f"--remote-debugging-port={port}", f"--user-data-dir={self.profile.name}",
"--lang=en-US", "about:blank"],
- stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+ stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=REAL_ENV)
deadline = time.monotonic() + 30
while True:
try:
@@ -830,8 +861,7 @@ def _linked_group_page(world: dict) -> "_Chrome":
raise
-@pytest.mark.skipif(not shutil.which("google-chrome"),
- reason="google-chrome is not available")
+@pytest.mark.skipif(not CHROME, reason="Chrome is not available")
def test_the_application_links_itself_from_its_group_page(world):
_linked_group_page(world).close()
@@ -997,8 +1027,7 @@ def _cli_signed_in(world: dict, home: str) -> None:
time.sleep(5)
-@pytest.mark.skipif(not shutil.which("google-chrome"),
- reason="google-chrome is not available")
+@pytest.mark.skipif(not CHROME, reason="Chrome is not available")
def test_chat_between_the_command_line_and_the_application(world):
"""Posted here, read on the group page in Chrome; answered there, read here —
the Python seal and the page's crypto.js, on a real node."""
@@ -1162,7 +1191,7 @@ async def test_put_and_get_in_the_interactive_chat_against_the_real_node(
assert json.loads(r.stdout)[0]["attachment"] == {
"filename": "from the room.bin", "size": len(sent), "type": "file"}
- if shutil.which("google-chrome"):
+ if CHROME:
await asyncio.to_thread(_the_page_shows_the_file, world, "from the room.bin")
@@ -1183,13 +1212,11 @@ def test_chat_follow_against_the_real_node(world):
"""`chat --follow` in its own process: messages posted from another run
arrive as the node pushes them, across more than one liveness ping, and
Ctrl+C ends it. Runs on the machine the chat tests above linked."""
- import signal
group = _handle()
env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"),
"PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"}
exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli")
- follower = subprocess.Popen([exe, "--json", "chat", "-1", group, "--follow"], env=env,
- stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+ follower = interruptible([exe, "--json", "chat", "-1", group, "--follow"], env)
try:
first = json.loads(follower.stdout.readline()) # the last one, at once
assert first["text"] or first.get("attachment") # /put's, above
@@ -1201,20 +1228,19 @@ def test_chat_follow_against_the_real_node(world):
assert got["text"] == text and got["sender"] == USERNAME, got
assert got["trust"] in ("pinned", "first", "linked")
finally:
- follower.send_signal(signal.SIGINT)
+ ctrl_c(follower)
_, err = follower.communicate(timeout=20)
assert follower.returncode == 130
assert "reconnecting" not in err, err
@pytest.mark.skipif(not (shutil.which("ffmpeg") and shutil.which("ffprobe")
- and shutil.which("cvlc")),
+ and VLC),
reason="ffmpeg, ffprobe and VLC are needed")
def test_stream_a_film_to_real_players(world, tmp_path):
"""A film put on the real node, then streamed: ffprobe reads it, ffmpeg seeks
near its end, a foreign Host is refused, Ctrl+C stops it; then VLC started
with --player plays it to the end and, closing, ends the stream."""
- import signal
_cli_signed_in(world, "cli-stream")
film = tmp_path / "Night Drive.mkv"
subprocess.run(["ffmpeg", "-v", "error", "-f", "lavfi", "-i",
@@ -1241,8 +1267,7 @@ def test_stream_a_film_to_real_players(world, tmp_path):
env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-stream"),
"PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"}
exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli")
- streamer = subprocess.Popen([exe, "--json", "stream", path], env=env,
- stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+ streamer = interruptible([exe, "--json", "stream", path], env)
try:
line = streamer.stdout.readline()
assert line, streamer.stderr.read()
@@ -1266,20 +1291,24 @@ def test_stream_a_film_to_real_players(world, tmp_path):
trust_env=False)
assert foreign.status_code == 403
finally:
- streamer.send_signal(signal.SIGINT)
+ ctrl_c(streamer)
_, err = streamer.communicate(timeout=20)
assert streamer.returncode == 130, err
with pytest.raises(httpx.HTTPError):
httpx.get(url, timeout=5, trust_env=False)
- player = tmp_path / "vlc-to-the-end"
log = tmp_path / "vlc.log"
- player.write_text("#!/bin/sh\nexec cvlc --play-and-exit --intf dummy --vout dummy "
- f"--aout dummy \"$1\" >{log} 2>&1\n")
- player.chmod(0o755)
+ flags = "--play-and-exit --intf dummy --vout dummy --aout dummy"
+ if sys.platform == "win32":
+ player = tmp_path / "vlc-to-the-end.cmd"
+ player.write_text(f'@"{VLC}" {flags} "%~1" >"{log}" 2>&1\r\n')
+ else:
+ player = tmp_path / "vlc-to-the-end"
+ player.write_text(f"#!/bin/sh\nexec {VLC} {flags} \"$1\" >{log} 2>&1\n")
+ player.chmod(0o755)
started = time.monotonic()
r = subprocess.run([exe, "stream", "--player", str(player), path], env=env,
- capture_output=True, text=True, timeout=120)
+ capture_output=True, text=True, encoding="utf-8", timeout=120)
assert r.returncode == 0, r.stderr + log.read_text()
assert "The player was closed" in r.stderr
assert time.monotonic() - started > 10, "VLC played the film, in real time"
@@ -1354,23 +1383,30 @@ def test_tab_completes_remote_paths_from_the_real_node(world):
"COMP_CWORD=2\n_meshbay_cli\nprintf '%s\\n' \"${COMPREPLY[@]}\"\n"],
env={**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"),
"PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring",
- "PATH": f"{BIN}:{os.environ['PATH']}"},
- capture_output=True, text=True, timeout=60)
+ "PATH": f"{BIN}{os.pathsep}{os.environ['PATH']}"},
+ capture_output=True, text=True, encoding="utf-8", timeout=60)
assert bash.stdout.splitlines() == [group + "/"], (bash.stdout, bash.stderr)
- pwsh = shutil.which("pwsh")
+ # PowerShell 7, or Windows PowerShell 5.1, which decodes a program's output
+ # with the console's code page unless told otherwise: an accented name is the
+ # test. The matches come back as code points, past PowerShell's own output.
+ pwsh = shutil.which("pwsh") or (sys.platform == "win32" and shutil.which("powershell"))
if pwsh:
ps = cli(world, "completion", "powershell").stdout
- line = f"meshbay-cli ls '{group}/sh"
- r = subprocess.run(
- [pwsh, "-NoProfile", "-Command",
- ps + f"\n$r = TabExpansion2 -inputScript \"{line}\" -cursorColumn {len(line)}"
- "\n$r.CompletionMatches | ForEach-Object { $_.CompletionText }"],
- env={**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"),
- "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring",
- "PATH": f"{BIN}:{os.environ['PATH']}"},
- capture_output=True, text=True, timeout=120)
- assert r.stdout.splitlines() == [f"'{group}/share/'"], (r.stdout, r.stderr)
+ for line, want in ((f"meshbay-cli ls '{group}/sh", f"'{group}/share/'"),
+ (f"meshbay-cli ls '{group}/share/2024/é",
+ f"'{group}/share/2024/été à la mer/'")):
+ r = subprocess.run(
+ [pwsh, "-NoProfile", "-Command",
+ ps + f"\n$r = TabExpansion2 -inputScript \"{line}\" -cursorColumn {len(line)}"
+ "\n$r.CompletionMatches | ForEach-Object {"
+ " ($_.CompletionText.ToCharArray() | ForEach-Object { [int]$_ }) -join ',' }"],
+ env={**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"),
+ "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring",
+ "PATH": f"{BIN}{os.pathsep}{os.environ['PATH']}"},
+ capture_output=True, text=True, timeout=120)
+ got = ["".join(map(chr, map(int, m.split(",")))) for m in r.stdout.split()]
+ assert got == [want], (got, r.stderr)
def _cli(world: dict, *args: str, new: str | None = None, **kw):
diff --git a/packages/meshbay-cli/tests/test_listing.py b/packages/meshbay-cli/tests/test_listing.py
index 407074e..f3e6a85 100644
--- a/packages/meshbay-cli/tests/test_listing.py
+++ b/packages/meshbay-cli/tests/test_listing.py
@@ -71,3 +71,21 @@ def test_control_characters_and_bidi_overrides_are_escaped():
def test_sizes():
assert human_size(6) == "6 B"
assert human_size(2_500_000) == "2.4 MB"
+
+
+def test_an_activity_time_without_a_zone_is_utc(monkeypatch):
+ """A hub on SQLite sends `last_activity_at` without its zone; it is UTC, as
+ PostgreSQL's says it is, and is shown in local time either way."""
+ import time
+
+ from meshbay_cli.commands.groups import local_time
+ if not hasattr(time, "tzset"):
+ pytest.skip("needs time.tzset to set the local zone")
+ monkeypatch.setenv("TZ", "UTC-02:00")
+ time.tzset()
+ try:
+ assert local_time("2026-10-15T13:59:00") == "2026-10-15 15:59"
+ assert local_time("2026-10-15T13:59:00+00:00") == "2026-10-15 15:59"
+ finally:
+ monkeypatch.undo()
+ time.tzset()