mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-07-13 12:38:02 +00:00
fix(platform): Improve WSL SSH remote compatibility (#3316)
* fix(platform): add WSL compatibility functions and path translation fix(cookbook): enhance model scan script to support additional HuggingFace cache paths fix(hardware): improve cache key generation for remote SSH context test(tests): add tests for WSL detection and path translation functionality * fix(cookbook): prefer prebuilt wheels for llama-cpp-python and normalize package aliases * fix: enable StrictHostKeyChecking in nvidia probe refactor: consolidate ssh & powershell command execution to utility functions in core module refactor: consolidate nvidia path candidates in to single variables in core module tests: add tests for new utility functions * fix: correct wrong variable name
This commit is contained in:
@@ -1,6 +1,8 @@
|
||||
"""Regression tests for cross-platform helper behavior."""
|
||||
|
||||
import importlib.util
|
||||
import io
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@@ -59,3 +61,243 @@ def test_find_bash_skips_windows_wsl_stub(monkeypatch):
|
||||
monkeypatch.setattr(platform_compat.os.path, "exists", lambda path: path == expected)
|
||||
|
||||
assert platform_compat.find_bash() == expected
|
||||
|
||||
|
||||
def test_is_wsl_true_when_proc_version_mentions_microsoft(monkeypatch):
|
||||
monkeypatch.setattr(sys, "platform", "linux", raising=False)
|
||||
|
||||
def fake_open(path, mode="r", *args, **kwargs):
|
||||
assert path == "/proc/version"
|
||||
assert mode == "r"
|
||||
return io.StringIO("Linux version 6.6.0 microsoft standard")
|
||||
|
||||
monkeypatch.setattr("builtins.open", fake_open)
|
||||
|
||||
assert platform_compat.is_wsl() is True
|
||||
|
||||
|
||||
def test_is_wsl_false_when_proc_version_is_not_microsoft(monkeypatch):
|
||||
monkeypatch.setattr(sys, "platform", "linux", raising=False)
|
||||
monkeypatch.setattr("builtins.open", lambda *_a, **_k: io.StringIO("Linux version 6.6.0 generic"))
|
||||
|
||||
assert platform_compat.is_wsl() is False
|
||||
|
||||
|
||||
def test_is_wsl_false_on_non_posix_without_proc_probe(monkeypatch):
|
||||
monkeypatch.setattr(sys, "platform", "win32", raising=False)
|
||||
monkeypatch.setattr(platform_compat.os, "name", "nt", raising=False)
|
||||
|
||||
def fail_open(*_args, **_kwargs):
|
||||
raise AssertionError("open should not be called when platform is not Linux/POSIX")
|
||||
|
||||
monkeypatch.setattr("builtins.open", fail_open)
|
||||
|
||||
assert platform_compat.is_wsl() is False
|
||||
|
||||
|
||||
def test_translate_path_converts_windows_drive_path_on_wsl(monkeypatch):
|
||||
monkeypatch.setattr(platform_compat, "is_wsl", lambda: True)
|
||||
|
||||
out = platform_compat.translate_path(r"C:\Users\alice\models\qwen.gguf")
|
||||
|
||||
assert out == "/mnt/c/Users/alice/models/qwen.gguf"
|
||||
|
||||
|
||||
def test_translate_path_resolves_paths_when_not_wsl(monkeypatch):
|
||||
monkeypatch.setattr(platform_compat, "is_wsl", lambda: False)
|
||||
|
||||
assert platform_compat.translate_path(".") == str(Path(".").resolve())
|
||||
|
||||
|
||||
def test_translate_path_returns_input_when_resolve_fails(monkeypatch):
|
||||
monkeypatch.setattr(platform_compat, "is_wsl", lambda: False)
|
||||
|
||||
class _BrokenPath:
|
||||
def __init__(self, _value):
|
||||
pass
|
||||
|
||||
def resolve(self):
|
||||
raise RuntimeError("boom")
|
||||
|
||||
monkeypatch.setattr(platform_compat, "Path", _BrokenPath)
|
||||
|
||||
assert platform_compat.translate_path("weird::path") == "weird::path"
|
||||
|
||||
|
||||
def test_get_wsl_windows_user_profile_prefers_powershell(monkeypatch):
|
||||
monkeypatch.setattr(platform_compat, "is_wsl", lambda: True)
|
||||
|
||||
class _Result:
|
||||
returncode = 0
|
||||
stdout = "C:\\Users\\alice\\n"
|
||||
|
||||
monkeypatch.setattr(platform_compat.subprocess, "run", lambda *_a, **_k: _Result())
|
||||
monkeypatch.setattr(platform_compat, "translate_path", lambda _v: "/mnt/c/Users/alice")
|
||||
|
||||
assert platform_compat.get_wsl_windows_user_profile() == "/mnt/c/Users/alice"
|
||||
|
||||
|
||||
def test_get_wsl_windows_user_profile_falls_back_to_users_dir(monkeypatch):
|
||||
monkeypatch.setattr(platform_compat, "is_wsl", lambda: True)
|
||||
|
||||
def raise_run(*_a, **_k):
|
||||
raise OSError("powershell unavailable")
|
||||
|
||||
monkeypatch.setattr(platform_compat.subprocess, "run", raise_run)
|
||||
monkeypatch.setattr(
|
||||
platform_compat.os,
|
||||
"listdir",
|
||||
lambda _path: ["All Users", "Default", "Public", "alice"],
|
||||
)
|
||||
|
||||
def fake_isdir(path):
|
||||
return path in {"/mnt/c/Users", "/mnt/c/Users/alice"}
|
||||
|
||||
monkeypatch.setattr(platform_compat.os.path, "isdir", fake_isdir)
|
||||
|
||||
assert platform_compat.get_wsl_windows_user_profile() == "/mnt/c/Users/alice"
|
||||
|
||||
|
||||
def test_get_wsl_windows_user_profile_returns_none_when_nothing_found(monkeypatch):
|
||||
monkeypatch.setattr(platform_compat, "is_wsl", lambda: True)
|
||||
monkeypatch.setattr(
|
||||
platform_compat.subprocess,
|
||||
"run",
|
||||
lambda *_a, **_k: (_ for _ in ()).throw(OSError("powershell unavailable")),
|
||||
)
|
||||
monkeypatch.setattr(platform_compat.os.path, "isdir", lambda _path: False)
|
||||
|
||||
assert platform_compat.get_wsl_windows_user_profile() is None
|
||||
|
||||
|
||||
def test_nvidia_path_override_is_correct_string(monkeypatch):
|
||||
monkeypatch.setattr(platform_compat, "_SSH_PATH_MEMBERS", ["path1", "path2"])
|
||||
assert platform_compat._ssh_path_override() == "export PATH=\"$PATH:path1:path2\"; "
|
||||
|
||||
|
||||
def test_windows_powershell_argv_defaults_include_no_profile_and_noninteractive():
|
||||
argv = platform_compat._windows_powershell_argv("Write-Output Hello")
|
||||
assert argv == [
|
||||
"powershell.exe",
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
"-Command",
|
||||
"Write-Output Hello",
|
||||
]
|
||||
|
||||
|
||||
def test_windows_powershell_argv_respects_disabled_flags():
|
||||
argv = platform_compat._windows_powershell_argv(
|
||||
"Write-Output Hello",
|
||||
no_profile=False,
|
||||
non_interactive=False,
|
||||
)
|
||||
assert argv == ["powershell.exe", "-Command", "Write-Output Hello"]
|
||||
|
||||
|
||||
def test_run_wsl_windows_powershell_raises_outside_wsl(monkeypatch):
|
||||
monkeypatch.setattr(platform_compat, "is_wsl", lambda: False)
|
||||
try:
|
||||
platform_compat.run_wsl_windows_powershell("Write-Output Hello", timeout=2)
|
||||
raise AssertionError("Expected RuntimeError")
|
||||
except RuntimeError as exc:
|
||||
assert "only supported in WSL" in str(exc)
|
||||
|
||||
|
||||
def test_run_wsl_windows_powershell_calls_subprocess_with_expected_argv(monkeypatch):
|
||||
monkeypatch.setattr(platform_compat, "is_wsl", lambda: True)
|
||||
captured = {}
|
||||
|
||||
class _Result:
|
||||
returncode = 0
|
||||
stdout = "ok\n"
|
||||
stderr = ""
|
||||
|
||||
def _fake_run(args, **kwargs):
|
||||
captured["args"] = list(args)
|
||||
captured["kwargs"] = kwargs
|
||||
return _Result()
|
||||
|
||||
monkeypatch.setattr(platform_compat.subprocess, "run", _fake_run)
|
||||
|
||||
result = platform_compat.run_wsl_windows_powershell("Write-Output Hello", timeout=9)
|
||||
|
||||
assert result.returncode == 0
|
||||
assert captured["args"] == [
|
||||
"powershell.exe",
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
"-Command",
|
||||
"Write-Output Hello",
|
||||
]
|
||||
assert captured["kwargs"]["capture_output"] is True
|
||||
assert captured["kwargs"]["text"] is True
|
||||
assert captured["kwargs"]["timeout"] == 9
|
||||
|
||||
|
||||
def test_ssh_exec_argv_builds_default_command():
|
||||
argv = platform_compat._ssh_exec_argv("alice@gpu-box", None, remote_cmd="echo ok")
|
||||
assert argv == ["ssh", "alice@gpu-box", "echo ok"]
|
||||
|
||||
|
||||
def test_ssh_exec_argv_includes_port_and_options():
|
||||
argv = platform_compat._ssh_exec_argv(
|
||||
"alice@gpu-box",
|
||||
"2222",
|
||||
remote_cmd="tmux ls",
|
||||
connect_timeout=6,
|
||||
strict_host_key_checking=False,
|
||||
)
|
||||
assert argv == [
|
||||
"ssh",
|
||||
"-o",
|
||||
"ConnectTimeout=6",
|
||||
"-o",
|
||||
"StrictHostKeyChecking=no",
|
||||
"-p",
|
||||
"2222",
|
||||
"alice@gpu-box",
|
||||
"tmux ls",
|
||||
]
|
||||
|
||||
|
||||
def test_run_ssh_command_uses_built_argv(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
class _Result:
|
||||
returncode = 0
|
||||
stdout = "ok"
|
||||
stderr = ""
|
||||
|
||||
def _fake_run(args, **kwargs):
|
||||
captured["args"] = list(args)
|
||||
captured["kwargs"] = kwargs
|
||||
return _Result()
|
||||
|
||||
monkeypatch.setattr(platform_compat.subprocess, "run", _fake_run)
|
||||
|
||||
result = platform_compat.run_ssh_command(
|
||||
"alice@gpu-box",
|
||||
"2200",
|
||||
"tmux ls",
|
||||
timeout=7,
|
||||
connect_timeout=3,
|
||||
strict_host_key_checking=True,
|
||||
text=False,
|
||||
)
|
||||
|
||||
assert result.returncode == 0
|
||||
assert captured["args"] == [
|
||||
"ssh",
|
||||
"-o",
|
||||
"ConnectTimeout=3",
|
||||
"-o",
|
||||
"StrictHostKeyChecking=yes",
|
||||
"-p",
|
||||
"2200",
|
||||
"alice@gpu-box",
|
||||
"tmux ls",
|
||||
]
|
||||
assert captured["kwargs"]["timeout"] == 7
|
||||
assert captured["kwargs"]["capture_output"] is True
|
||||
assert captured["kwargs"]["text"] is False
|
||||
|
||||
Reference in New Issue
Block a user