"""cli/remote.py（sm-remote CLI）の仕様ベーステスト．
仕様出典:
- docs/02_ENV/ENV_04_開発コマンド.md「研究室サーバでの実行」「環境変数」
- docs/02_ENV/ENV_02_環境構築手順.md「研究室サーバ編」
- 実装サマリー 仕様 1〜6（push / run / status・logs・attach・kill / pull / 設定の解決 /
サーバ上での実行検出）
テスト方針:
- ssh・tmux・実サーバには依存しない．subprocess.Popen を偽物に差し替え，
「ssh にどんな引数・標準入力を渡したか」と「ssh の終了コードをどう扱うか」を検証する
- ホスト名はテストごとに固定し（dev-pc），CWD はリポジトリ外にする
（このテストを研究室サーバ上で回してもサーバ上検出が誤作動しないように）
"""
from __future__ import annotations
import gzip
import io
import os
import re
import shlex
import subprocess
import tarfile
import types
from pathlib import Path
import pytest
import sm_tias_lightsim.cli.remote as remote_cli
from sm_tias_lightsim.cli.remote import (
DEFAULT_HOST,
DEFAULT_REMOTE_DIR,
DEFAULT_UV,
build_job_env,
default_ssh_executable,
detect_running_on_server,
main,
resolve_settings,
)
# ---------------------------------------------------------------------------
# 偽の ssh（subprocess.Popen）
# ---------------------------------------------------------------------------
class _CapturingPipe(io.BytesIO):
"""ssh の標準入力の代わり．close 時点の内容を captured に残す．"""
captured: bytes | None = None
def close(self) -> None:
if not self.closed:
self.captured = self.getvalue()
super().close()
class _InterruptingStdout(io.BytesIO):
"""読むと KeyboardInterrupt（手元の Ctrl+C）を起こす標準出力．"""
def read1(self, size: int = -1) -> bytes:
raise KeyboardInterrupt
class FakeProc:
def __init__(
self,
argv: list[str],
kwargs: dict,
*,
rc: int = 0,
stdout: bytes = b"",
finished: bool = False,
interrupt: bool = False,
) -> None:
self.argv = argv
self.kwargs = kwargs
self._rc = rc
self.returncode = rc if finished else None
self.stdin = _CapturingPipe() if kwargs.get("stdin") == subprocess.PIPE else None
if kwargs.get("stdout") == subprocess.PIPE:
self.stdout = _InterruptingStdout(stdout) if interrupt else io.BytesIO(stdout)
else:
self.stdout = None
self.input: bytes | None = None
self.killed = False
def communicate(self, input: bytes | None = None): # noqa: A002
self.input = input
self.returncode = self._rc
return (self.stdout.read() if self.stdout is not None else None), None
def wait(self, timeout: float | None = None) -> int:
self.returncode = self._rc
return self._rc
def poll(self) -> int | None:
return self.returncode
def kill(self) -> None:
self.killed = True
@pytest.fixture
def ssh(monkeypatch):
"""subprocess.Popen を偽物にする．responses に積んだ応答を呼ばれた順に返す．"""
state = types.SimpleNamespace(calls=[], responses=[])
def fake_popen(argv, **kwargs):
spec = state.responses.pop(0) if state.responses else {}
proc = FakeProc(list(argv), kwargs, **spec)
state.calls.append(proc)
return proc
monkeypatch.setattr(remote_cli.subprocess, "Popen", fake_popen)
return state
@pytest.fixture(autouse=True)
def _isolated_environment(monkeypatch, tmp_path):
for key in (
"SM_REMOTE_HOST",
"SM_REMOTE_DIR",
"SM_REMOTE_UV",
"SM_REMOTE_VARIANT",
"SM_REMOTE_ALLOW_SELF",
):
monkeypatch.delenv(key, raising=False)
monkeypatch.setenv("SM_REMOTE_SSH", "fake-ssh")
monkeypatch.setattr(remote_cli.socket, "gethostname", lambda: "dev-pc")
cwd = tmp_path / "cwd"
cwd.mkdir()
monkeypatch.chdir(cwd)
# ---------------------------------------------------------------------------
# シェルの単語分割の再現
# ---------------------------------------------------------------------------
def _shell_words(text: str) -> list[str]:
lex = shlex.shlex(text, posix=True, punctuation_chars=";&|")
lex.whitespace_split = True
return list(lex)
def _statements(text: str) -> list[list[str]]:
result: list[list[str]] = [[]]
for word in _shell_words(text):
if word == ";":
result.append([])
else:
result[-1].append(word)
return [s for s in result if s]
def _has_sequence(words: list[str], seq: list[str]) -> bool:
n = len(seq)
return any(words[i : i + n] == seq for i in range(len(words) - n + 1))
def _job_script(run_command: str) -> str:
"""build_run_command の出力から，tmux → bash -c に渡るジョブ本体を取り出す．"""
words = _shell_words(run_command)
stmt = words[words.index("new-session") :]
if ";" in stmt:
stmt = stmt[: stmt.index(";")]
inner = shlex.split(stmt[-1])
assert inner[:2] == ["bash", "-c"]
return inner[2]
def _exports(script: str) -> dict[str, str]:
result = {}
for stmt in _statements(script):
if stmt[0] == "export" and len(stmt) == 2 and not stmt[1].startswith("PATH="):
key, _, value = stmt[1].partition("=")
result[key] = value
return result
def _run_statement(script: str) -> list[str]:
return next(s for s in _statements(script) if "/dev/null" in s and "tee" in s)
def _tar_bytes(members: dict[str, bytes], *, gz: bool = False) -> bytes:
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz" if gz else "w", format=tarfile.PAX_FORMAT) as tar:
for name, data in members.items():
info = tarfile.TarInfo(name)
info.size = len(data)
info.mode = 0o644
tar.addfile(info, io.BytesIO(data))
return buf.getvalue()
# ---------------------------------------------------------------------------
# 疑似プロジェクト（push 用）
# ---------------------------------------------------------------------------
_PROJECT = {
"pyproject.toml": "[project]\nname='x'\n",
"src/sm_tias_lightsim/__init__.py": "",
"src/sm_tias_lightsim/a.py": "x = 1\n",
"src/sm_tias_lightsim/__pycache__/a.cpython-312.pyc": "pyc",
"tests/test_a.py": "",
"scripts/s.py": "",
".venv/Scripts/python.exe": "exe",
".git/HEAD": "ref",
"output/r/luminance.npy": "npy",
"configs/scene.yaml": "geometry:\n meshes:\n - id: a\n path: meshes/a.stl\n",
"configs/meshes/前側.stl": "solid",
"configs/_tmp_x/s.yaml": "geometry:\n meshes:\n - id: a\n path: C:\\Users\\r\\a.stl\n",
}
_EXPECTED_PUSH = {
"pyproject.toml",
"src/sm_tias_lightsim/__init__.py",
"src/sm_tias_lightsim/a.py",
"tests/test_a.py",
"scripts/s.py",
"configs/scene.yaml",
"configs/meshes/前側.stl",
}
@pytest.fixture
def project(tmp_path) -> Path:
root = tmp_path / "proj"
for rel, content in _PROJECT.items():
path = root / rel
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
return root
# ---------------------------------------------------------------------------
# 設定の解決（仕様 5）
# ---------------------------------------------------------------------------
class TestResolveSettings:
def test_default_values(self):
assert DEFAULT_HOST == "Newton"
assert DEFAULT_REMOTE_DIR == "~/lab/SmTIAS-LightSim"
assert DEFAULT_UV == "~/.local/bin/uv"
settings = resolve_settings(None, None, None, {})
assert settings.host == "Newton"
assert settings.remote_dir == "~/lab/SmTIAS-LightSim"
assert settings.uv_path == "~/.local/bin/uv"
assert settings.ssh == default_ssh_executable({})
def test_env_overrides_defaults(self):
env = {
"SM_REMOTE_HOST": "Einstein",
"SM_REMOTE_DIR": "/srv/proj",
"SM_REMOTE_UV": "/opt/uv/uv",
"SM_REMOTE_SSH": "/usr/bin/ssh",
}
settings = resolve_settings(None, None, None, env)
assert (settings.host, settings.remote_dir, settings.uv_path, settings.ssh) == (
"Einstein",
"/srv/proj",
"/opt/uv/uv",
"/usr/bin/ssh",
)
def test_args_override_env(self):
env = {"SM_REMOTE_HOST": "Einstein", "SM_REMOTE_DIR": "/srv/a", "SM_REMOTE_UV": "/a/uv"}
settings = resolve_settings("Archimedes", "~/b", "/b/uv", env)
assert (settings.host, settings.remote_dir, settings.uv_path) == (
"Archimedes",
"~/b",
"/b/uv",
)
def test_blank_values_fall_back(self):
env = {"SM_REMOTE_HOST": " ", "SM_REMOTE_DIR": "", "SM_REMOTE_UV": " "}
settings = resolve_settings("", " ", None, env)
assert (settings.host, settings.remote_dir, settings.uv_path) == (
"Newton",
"~/lab/SmTIAS-LightSim",
"~/.local/bin/uv",
)
def test_main_uses_env_then_args_for_host_and_dir(self, ssh, monkeypatch):
monkeypatch.setenv("SM_REMOTE_HOST", "Einstein")
monkeypatch.setenv("SM_REMOTE_DIR", "/srv/env dir")
assert main(["status"]) == 0
argv = ssh.calls[0].argv
assert argv[:2] == ["fake-ssh", "Einstein"]
assert _statements(argv[-1])[0][:2] == ["cd", "/srv/env dir"]
assert main(["status", "--host", "Archimedes", "--remote-dir", "~/x"]) == 0
argv = ssh.calls[1].argv
assert argv[:2] == ["fake-ssh", "Archimedes"]
assert _statements(argv[-1])[0][:2] == ["cd", "$HOME/x"]
class TestDefaultSshExecutable:
def _make_ssh(self, root: Path) -> Path:
exe = root / "System32" / "OpenSSH" / "ssh.exe"
exe.parent.mkdir(parents=True)
exe.write_bytes(b"")
return exe
def test_windows_prefers_system32_openssh(self, tmp_path):
exe = self._make_ssh(tmp_path)
assert default_ssh_executable({"SystemRoot": str(tmp_path)}, "nt") == str(exe)
def test_windows_uppercase_systemroot(self, tmp_path):
exe = self._make_ssh(tmp_path)
assert default_ssh_executable({"SYSTEMROOT": str(tmp_path)}, "nt") == str(exe)
def test_windows_without_openssh_falls_back_to_path(self, tmp_path):
assert default_ssh_executable({"SystemRoot": str(tmp_path)}, "nt") == "ssh"
def test_posix_uses_ssh_on_path(self, tmp_path):
self._make_ssh(tmp_path)
assert default_ssh_executable({"SystemRoot": str(tmp_path)}, "posix") == "ssh"
@pytest.mark.skipif(
os.name != "nt" or not Path(r"C:\Windows\System32\OpenSSH\ssh.exe").is_file(),
reason="Windows 標準の OpenSSH がある環境のみ",
)
def test_real_windows_default(self):
result = default_ssh_executable(os.environ)
assert result.lower() == r"c:\windows\system32\openssh\ssh.exe"
# ---------------------------------------------------------------------------
# サーバ上での実行検出（仕様 6）
# ---------------------------------------------------------------------------
def _detect(**overrides):
kwargs = {
"host": "Newton",
"remote_dir": "~/lab/SmTIAS-LightSim",
"environ": {},
"local_hostname": "dev-pc",
"project_root": None,
"home": None,
"os_name": "posix",
}
kwargs.update(overrides)
host = kwargs.pop("host")
remote_dir = kwargs.pop("remote_dir")
return detect_running_on_server(host, remote_dir, **kwargs)
class TestDetectRunningOnServer:
@pytest.mark.parametrize(
("local", "host"),
[
("Newton", "Newton"),
("newton", "Newton"),
("NEWTON", "Newton"),
("newton.local", "Newton"),
("Newton", "Newton.local"),
("newton.lab.example.ac.jp", "Newton.local"),
(" newton ", "Newton"),
("192.168.23.13", "192.168.23.13"),
],
)
def test_hostname_match(self, local, host):
"""ホスト名は小文字化・ドメイン部除去で比べる（IPv4 はそのまま）．"""
reason = _detect(host=host, local_hostname=local)
assert isinstance(reason, str) and reason
@pytest.mark.parametrize(
("local", "host"),
[
("dev-pc", "Newton"),
("Newton2", "Newton"),
("newt", "Newton"),
("192.168.23.13", "192.168.23.15"),
("192", "192.168.23.13"), # IPv4 はドメイン部除去をしない
("192.168.23.13", "192"),
("", "Newton"),
],
)
def test_hostname_mismatch(self, local, host):
assert _detect(host=host, local_hostname=local) is None
def test_hostname_match_applies_on_windows_too(self):
assert _detect(local_hostname="newton", os_name="nt") is not None
def _server_layout(self, tmp_path: Path) -> tuple[Path, Path]:
home = tmp_path / "home"
root = home / "lab" / "SmTIAS-LightSim"
root.mkdir(parents=True)
return home, root
@pytest.mark.parametrize("remote_dir", ["~/lab/SmTIAS-LightSim", "lab/SmTIAS-LightSim"])
def test_posix_repo_root_match(self, tmp_path, remote_dir):
"""POSIX でリポジトリのルートが remote_dir（~ 展開・ホーム基準）と一致すれば検出．"""
home, root = self._server_layout(tmp_path)
assert _detect(remote_dir=remote_dir, project_root=root, home=home) is not None
def test_posix_absolute_remote_dir_match(self, tmp_path):
home, root = self._server_layout(tmp_path)
assert _detect(remote_dir=str(root), project_root=root, home=home) is not None
def test_posix_repo_root_mismatch(self, tmp_path):
home, _ = self._server_layout(tmp_path)
other = tmp_path / "work" / "SmTIAS-LightSim"
other.mkdir(parents=True)
assert _detect(project_root=other, home=home) is None
def test_windows_does_not_compare_directories(self, tmp_path):
"""Windows ではディレクトリを比べない（手元の PC を誤検出しない）．"""
home, root = self._server_layout(tmp_path)
assert _detect(project_root=root, home=home, os_name="nt") is None
def test_unknown_root_or_home_is_not_detected(self, tmp_path):
home, root = self._server_layout(tmp_path)
assert _detect(project_root=None, home=home) is None
assert _detect(project_root=root, home=None) is None
def test_tilde_user_remote_dir_does_not_crash(self, tmp_path):
home, root = self._server_layout(tmp_path)
assert _detect(remote_dir="~other/lab", project_root=root, home=home) is None
@pytest.mark.parametrize("value", ["1", "true", "yes", "TRUE", " Yes "])
def test_allow_self_disables_detection(self, tmp_path, value):
home, root = self._server_layout(tmp_path)
environ = {"SM_REMOTE_ALLOW_SELF": value}
assert (
_detect(environ=environ, local_hostname="newton", project_root=root, home=home) is None
)
@pytest.mark.parametrize("value", ["", "0", "no", "false", "off"])
def test_other_allow_self_values_keep_detection(self, value):
environ = {"SM_REMOTE_ALLOW_SELF": value}
assert _detect(environ=environ, local_hostname="newton") is not None
class TestMainStopsOnServer:
@pytest.mark.parametrize(
"argv",
[
["status"],
["logs", "job1"],
["logs", "job1", "-f"],
["attach", "job1"],
["kill", "job1"],
["run", "--name", "job1", "--", "echo", "hi"],
["push", "--dry-run", "--local-dir", "{project}"],
["pull", "foo", "--dest", "{dest}"],
],
)
def test_exits_1_before_starting_ssh(self, argv, ssh, monkeypatch, capsys, project, tmp_path):
"""サーバ上なら ssh を起動する前に案内を出して終了コード 1 で止まること．"""
monkeypatch.setattr(remote_cli.socket, "gethostname", lambda: "newton.local")
argv = [a.format(project=project, dest=tmp_path / "dest") for a in argv]
assert main(argv) == 1
assert ssh.calls == []
captured = capsys.readouterr()
assert "SM_REMOTE_ALLOW_SELF" in captured.err
assert "送信対象" not in captured.out
def test_allow_self_lets_command_run(self, ssh, monkeypatch):
monkeypatch.setattr(remote_cli.socket, "gethostname", lambda: "newton")
monkeypatch.setenv("SM_REMOTE_ALLOW_SELF", "1")
assert main(["status"]) == 0
assert len(ssh.calls) == 1
# ---------------------------------------------------------------------------
# run（仕様 2）
# ---------------------------------------------------------------------------
class TestBuildJobEnv:
def test_minimal_env_is_unbuffered_python(self):
assert build_job_env(gpu=None, variant=None, threads=None, extra=[]) == {
"PYTHONUNBUFFERED": "1"
}
def test_all_options(self):
env = build_job_env(gpu="0,2", variant="cuda_ad_rgb", threads=16, extra=["FOO=bar"])
assert env == {
"PYTHONUNBUFFERED": "1",
"CUDA_VISIBLE_DEVICES": "0,2",
"SM_RENDER_VARIANT": "cuda_ad_rgb",
"SM_DRJIT_THREADS": "16",
"FOO": "bar",
}
def test_extra_is_last_wins(self):
env = build_job_env(gpu=None, variant=None, threads=None, extra=["A=1", "A=2"])
assert env["A"] == "2"
@pytest.mark.parametrize(
"kwargs",
[
{"gpu": "0;rm"},
{"gpu": ""},
{"threads": 0},
{"threads": -4},
{"extra": ["NOEQUALS"]},
{"extra": ["1BAD=x"]},
],
)
def test_invalid_values_raise(self, kwargs):
base = {"gpu": None, "variant": None, "threads": None, "extra": []}
base.update(kwargs)
with pytest.raises(ValueError):
build_job_env(**base)
def _dry_run_script(capsys, *args: str) -> tuple[str, str]:
"""run --dry-run を実行し，(表示されたリモートコマンド, ジョブ本体) を返す．"""
argv = ["run", "--name", "offaxis", "--dry-run", *args]
if "--" not in args:
argv += ["--", "python", "scripts/sweep_offaxis_led.py"]
assert main(argv) == 0
remote_command = capsys.readouterr().out.strip()
return remote_command, _job_script(remote_command)
class TestRunCommand:
def test_dry_run_does_not_start_ssh(self, ssh, capsys):
remote_command, _ = _dry_run_script(capsys)
assert ssh.calls == []
assert "new-session" in remote_command
def test_default_env(self, capsys):
"""既定で SM_RENDER_VARIANT=cuda_ad_rgb と PYTHONUNBUFFERED=1 を渡すこと．"""
_, script = _dry_run_script(capsys)
env = _exports(script)
assert env["SM_RENDER_VARIANT"] == "cuda_ad_rgb"
assert env["PYTHONUNBUFFERED"] == "1"
assert "CUDA_VISIBLE_DEVICES" not in env
assert "SM_DRJIT_THREADS" not in env
def test_yaml_variant_does_not_set_variant(self, capsys):
remote_command, script = _dry_run_script(capsys, "--yaml-variant")
assert "SM_RENDER_VARIANT" not in _exports(script)
assert "SM_RENDER_VARIANT" not in remote_command
def test_gpu_and_threads(self, capsys):
_, script = _dry_run_script(capsys, "--gpu", "0,2", "--threads", "16")
env = _exports(script)
assert env["CUDA_VISIBLE_DEVICES"] == "0,2"
assert env["SM_DRJIT_THREADS"] == "16"
def test_variant_precedence(self, capsys, monkeypatch):
monkeypatch.setenv("SM_REMOTE_VARIANT", "llvm_ad_rgb")
_, script = _dry_run_script(capsys)
assert _exports(script)["SM_RENDER_VARIANT"] == "llvm_ad_rgb"
_, script = _dry_run_script(capsys, "--variant", "scalar_rgb")
assert _exports(script)["SM_RENDER_VARIANT"] == "scalar_rgb"
_, script = _dry_run_script(capsys, "--variant", "scalar_rgb", "--yaml-variant")
assert "SM_RENDER_VARIANT" not in _exports(script)
def test_extra_env(self, capsys):
_, script = _dry_run_script(capsys, "--env", "FOO=a b", "--env", "BAR=")
env = _exports(script)
assert env["FOO"] == "a b"
assert env["BAR"] == ""
def test_command_after_double_dash_survives_quoting(self, capsys):
command = ["python", "-c", "print('a b'); import os", "--x=$(id)", "it's"]
_, script = _dry_run_script(capsys, "--", *command)
run = _run_statement(script)
assert run[: 2 + len(command)] == ["$HOME/.local/bin/uv", "run", *command]
def test_uv_path_from_env_and_arg(self, capsys, monkeypatch):
monkeypatch.setenv("SM_REMOTE_UV", "/opt/uv/uv")
_, script = _dry_run_script(capsys)
assert _run_statement(script)[:2] == ["/opt/uv/uv", "run"]
_, script = _dry_run_script(capsys, "--uv", "/usr/local/bin/uv")
assert _run_statement(script)[:2] == ["/usr/local/bin/uv", "run"]
def test_no_uv(self, capsys):
_, script = _dry_run_script(capsys, "--no-uv", "--", "nvidia-smi")
assert _run_statement(script)[0] == "nvidia-smi"
def test_log_and_remote_dir(self, capsys):
remote_command, script = _dry_run_script(capsys, "--remote-dir", "~/lab/X")
statements = _statements(script)
assert statements[0][:2] == ["cd", "$HOME/lab/X"]
assert ["log=output/_remote_logs/offaxis.log"] in statements
assert _has_sequence(_run_statement(script), ["<", "/dev/null"])
assert _statements(remote_command)[0][:2] == ["cd", "$HOME/lab/X"]
def test_default_job_name_is_timestamp(self, capsys):
assert main(["run", "--dry-run", "--", "echo"]) == 0
script = _job_script(capsys.readouterr().out.strip())
logs = [s[0] for s in _statements(script) if s[0].startswith("log=")]
assert re.fullmatch(r"log=output/_remote_logs/job-\d{8}-\d{6}\.log", logs[0])
@pytest.mark.parametrize(
"argv",
[
["run", "--name", "x"], # コマンドなし
["run", "--name", "x", "--"],
["run", "--name", "bad name", "--", "echo"],
["run", "--name", "a.b", "--", "echo"],
["run", "--name", "x", "--gpu", "0;rm", "--", "echo"],
["run", "--name", "x", "--gpu", "gpu0", "--", "echo"],
["run", "--name", "x", "--threads", "0", "--", "echo"],
["run", "--name", "x", "--env", "BAD", "--", "echo"],
],
)
def test_invalid_input_exits_1_without_ssh(self, argv, ssh, capsys):
assert main(argv) == 1
assert ssh.calls == []
assert "エラー" in capsys.readouterr().err
def test_launch_success(self, ssh, capsys):
ssh.responses.append({"rc": 0})
rc = main(["run", "--name", "offaxis", "--gpu", "0", "--", "python", "a.py"])
assert rc == 0
proc = ssh.calls[0]
assert proc.argv[:2] == ["fake-ssh", "Newton"]
assert len(proc.argv) == 3
assert proc.kwargs["stdin"] == subprocess.DEVNULL
assert _exports(_job_script(proc.argv[-1]))["CUDA_VISIBLE_DEVICES"] == "0"
out = capsys.readouterr().out
assert "offaxis" in out
assert "output/_remote_logs/offaxis.log" in out
def test_existing_session_returns_91(self, ssh, capsys):
ssh.responses.append({"rc": 91})
assert main(["run", "--name", "offaxis", "--", "echo"]) == 91
err = capsys.readouterr().err
assert "offaxis" in err
assert "実行中" in err
def test_ssh_failure_is_reported(self, ssh, capsys):
ssh.responses.append({"rc": 255})
assert main(["run", "--name", "offaxis", "--", "echo"]) == 255
assert "ssh" in capsys.readouterr().err
def test_missing_remote_dir_is_reported(self, ssh, capsys):
ssh.responses.append({"rc": 90})
assert main(["run", "--name", "offaxis", "--", "echo"]) == 90
assert "push" in capsys.readouterr().err
# ---------------------------------------------------------------------------
# status / logs / attach / kill（仕様 3）
# ---------------------------------------------------------------------------
class TestStatus:
def test_prints_remote_output(self, ssh, capsys):
ssh.responses.append({"rc": 0, "stdout": "== tmux sessions ==\n(none)\n日本語\n".encode()})
assert main(["status"]) == 0
assert "(none)" in capsys.readouterr().out
words = _shell_words(ssh.calls[0].argv[-1])
assert "nvidia-smi" in words
assert ssh.calls[0].kwargs["stdin"] == subprocess.DEVNULL
def test_missing_remote_dir(self, ssh, capsys):
ssh.responses.append({"rc": 90})
assert main(["status"]) == 90
assert "push" in capsys.readouterr().err
class TestLogs:
def test_tail_n_lines(self, ssh, capsys):
ssh.responses.append({"rc": 0, "stdout": b"line1\nline2\n"})
assert main(["logs", "offaxis", "-n", "7"]) == 0
assert _statements(ssh.calls[0].argv[-1])[-1] == [
"tail",
"-n",
"7",
"output/_remote_logs/offaxis.log",
]
assert "line2" in capsys.readouterr().out
def test_missing_log_returns_92(self, ssh, capsys):
ssh.responses.append({"rc": 92})
assert main(["logs", "offaxis"]) == 92
assert "offaxis" in capsys.readouterr().err
@pytest.mark.parametrize("argv", [["logs", "offaxis", "-n", "-1"], ["logs", "a;b"]])
def test_invalid_input_exits_1(self, argv, ssh):
assert main(argv) == 1
assert ssh.calls == []
def test_follow_streams_and_closes_stdin(self, ssh, capsys):
"""-f はリモートの出力を逐次表示し，終了時に ssh の標準入力を閉じる（tail を止める）．"""
ssh.responses.append({"rc": 0, "stdout": b"start\n 50%\r100%\n"})
assert main(["logs", "offaxis", "-f"]) == 0
proc = ssh.calls[0]
assert proc.kwargs["stdin"] == subprocess.PIPE
assert proc.stdin.closed
assert "100%" in capsys.readouterr().out
assert "-F" in _shell_words(proc.argv[-1])
def test_follow_ctrl_c_returns_0_and_stops_remote_tail(self, ssh, capsys):
"""Ctrl+C は追従の正常な終わらせ方（0）．標準入力を閉じてリモートの tail を止める．"""
ssh.responses.append({"rc": 255, "interrupt": True})
assert main(["logs", "offaxis", "-f"]) == 0
assert ssh.calls[0].stdin.closed
assert "ジョブは継続" in capsys.readouterr().err
class TestAttach:
def test_ended_job_returns_93_with_last_line(self, ssh, capsys):
ssh.responses.append({"rc": 93, "stdout": b"=== [2026-09-13 12:00:00] exit 0\n"})
assert main(["attach", "offaxis"]) == 93
assert len(ssh.calls) == 1 # tmux には接続しない
out = capsys.readouterr().out
assert "既に終了しています" in out
assert "exit 0" in out
def test_unknown_job_returns_92(self, ssh, capsys):
ssh.responses.append({"rc": 92})
assert main(["attach", "offaxis"]) == 92
assert len(ssh.calls) == 1
assert "offaxis" in capsys.readouterr().err
def test_missing_remote_dir_returns_90(self, ssh, capsys):
ssh.responses.append({"rc": 90})
assert main(["attach", "offaxis"]) == 90
assert len(ssh.calls) == 1
assert "push" in capsys.readouterr().err
def test_running_job_attaches_with_tty(self, ssh):
ssh.responses += [{"rc": 0}, {"rc": 0}]
assert main(["attach", "offaxis"]) == 0
assert len(ssh.calls) == 2
attach = ssh.calls[1].argv
assert attach[:3] == ["fake-ssh", "-t", "Newton"]
assert shlex.split(attach[-1]) == ["tmux", "attach-session", "-t", "=offaxis"]
def test_invalid_name(self, ssh):
assert main(["attach", "a b"]) == 1
assert ssh.calls == []
class TestKill:
def test_success(self, ssh, capsys):
ssh.responses.append({"rc": 0})
assert main(["kill", "offaxis"]) == 0
assert "offaxis" in capsys.readouterr().out
assert _has_sequence(
_shell_words(ssh.calls[0].argv[-1]), ["kill-session", "-t", "=offaxis"]
)
def test_not_running_returns_93(self, ssh, capsys):
ssh.responses.append({"rc": 93})
assert main(["kill", "offaxis"]) == 93
assert "offaxis" in capsys.readouterr().err
# ---------------------------------------------------------------------------
# pull（仕様 4）
# ---------------------------------------------------------------------------
class TestPull:
@pytest.mark.parametrize("spec", ["foo", "output/foo", "output\\foo"])
def test_accepts_all_spellings_and_extracts(self, spec, ssh, tmp_path):
dest = tmp_path / "dest"
ssh.responses.append({"rc": 0, "stdout": _tar_bytes({"output/foo/a.txt": b"hello"})})
assert main(["pull", spec, "--dest", str(dest)]) == 0
assert (dest / "output/foo/a.txt").read_bytes() == b"hello"
proc = ssh.calls[0]
assert proc.kwargs["stdin"] == subprocess.DEVNULL
assert _statements(proc.argv[-1])[-1] == ["tar", "-cf", "-", "output/foo"]
def test_compress(self, ssh, tmp_path):
dest = tmp_path / "dest"
payload = _tar_bytes({"output/foo/前側.npy": b"\x00" * 100}, gz=True)
ssh.responses.append({"rc": 0, "stdout": payload})
assert main(["pull", "foo", "--compress", "--dest", str(dest)]) == 0
assert (dest / "output/foo/前側.npy").read_bytes() == b"\x00" * 100
assert _statements(ssh.calls[0].argv[-1])[-1] == ["tar", "-czf", "-", "output/foo"]
def test_multiple_paths_are_deduplicated(self, ssh, tmp_path):
payload = _tar_bytes({"output/foo/a": b"1", "output/bar/b": b"2"})
ssh.responses.append({"rc": 0, "stdout": payload})
assert main(["pull", "foo", "output/foo", "bar", "--dest", str(tmp_path / "d")]) == 0
assert _statements(ssh.calls[0].argv[-1])[-1] == [
"tar",
"-cf",
"-",
"output/foo",
"output/bar",
]
@pytest.mark.parametrize("spec", ["../src", "output", "/etc", "C:\\x", "x/../../y"])
def test_rejects_paths_outside_output(self, spec, ssh, tmp_path, capsys):
assert main(["pull", spec, "--dest", str(tmp_path / "d")]) == 1
assert ssh.calls == []
assert "エラー" in capsys.readouterr().err
def test_unrequested_member_is_rejected(self, ssh, tmp_path, capsys):
"""要求したパス以外のメンバーが届いたら展開を中止し，失敗を返すこと．"""
dest = tmp_path / "dest"
payload = _tar_bytes({"output/foo/a.txt": b"ok", "src/sm_tias_lightsim/evil.py": b"x"})
ssh.responses.append({"rc": 0, "stdout": payload})
assert main(["pull", "foo", "--dest", str(dest)]) == 1
assert not (dest / "src").exists()
assert "エラー" in capsys.readouterr().err
def test_remote_path_missing_returns_92(self, ssh, tmp_path, capsys):
ssh.responses.append({"rc": 92, "stdout": b"", "finished": True})
assert main(["pull", "nope", "--dest", str(tmp_path / "d")]) == 92
assert "サーバにありません" in capsys.readouterr().err
def test_ssh_failure_returns_255(self, ssh, tmp_path, capsys):
ssh.responses.append({"rc": 255, "stdout": b"", "finished": True})
assert main(["pull", "foo", "--dest", str(tmp_path / "d")]) == 255
assert "ssh" in capsys.readouterr().err
def test_tar_warning_rc1_still_succeeds(self, ssh, tmp_path, capsys):
"""GNU tar の終了コード 1（読み取り中に変化）は警告にとどめ，取得は成功とする．"""
dest = tmp_path / "dest"
ssh.responses.append({"rc": 1, "stdout": _tar_bytes({"output/foo/a": b"x"})})
assert main(["pull", "foo", "--dest", str(dest)]) == 0
assert (dest / "output/foo/a").is_file()
assert "警告" in capsys.readouterr().err
# ---------------------------------------------------------------------------
# push（仕様 1）
# ---------------------------------------------------------------------------
def _tar_names(data: bytes) -> list[str]:
if data[:2] == b"\x1f\x8b":
data = gzip.decompress(data)
with tarfile.open(fileobj=io.BytesIO(data), mode="r:", encoding="utf-8") as tar:
assert tar.format == tarfile.PAX_FORMAT
return tar.getnames()
class TestPush:
def test_dry_run_lists_without_ssh(self, ssh, capsys, project):
assert main(["push", "--dry-run", "--local-dir", str(project)]) == 0
assert ssh.calls == []
out = capsys.readouterr().out
assert "送信対象: 7 ファイル" in out
assert "src/sm_tias_lightsim/a.py" in out
assert ".venv" not in out
assert "output/r" not in out
assert "--dry-run" in out
def test_windows_path_yaml_is_warned(self, ssh, capsys, project):
assert (
main(["push", "--dry-run", "--local-dir", str(project), "--include", "configs/_tmp_x"])
== 0
)
out = capsys.readouterr().out
assert "警告" in out
assert "configs/_tmp_x/s.yaml" in out
def test_no_warning_for_portable_yaml(self, ssh, capsys, project):
assert main(["push", "--dry-run", "--local-dir", str(project)]) == 0
assert "警告" not in capsys.readouterr().out
@pytest.mark.parametrize(
"extra",
[
["--include", "configs/_tmp_x", "--only", "configs/_tmp_x"],
["--prune", "--only", "configs/_tmp_x"],
["--only", "output/r"],
["--include", "output"],
["--only", "configs/_tmp_missing"],
],
)
def test_invalid_options_exit_1(self, extra, ssh, capsys, project):
assert main(["push", "--local-dir", str(project), *extra]) == 1
assert ssh.calls == []
assert "エラー" in capsys.readouterr().err
def test_streams_tar_to_remote_tar_x(self, ssh, capsys, project):
ssh.responses.append({"rc": 0})
assert main(["push", "--local-dir", str(project)]) == 0
proc = ssh.calls[0]
assert proc.argv[:2] == ["fake-ssh", "Newton"]
assert proc.kwargs["stdin"] == subprocess.PIPE
assert proc.stdin.captured is not None
assert set(_tar_names(proc.stdin.captured)) == _EXPECTED_PUSH
words = _shell_words(proc.argv[-1])
assert _has_sequence(words, ["tar", "-xf", "-", "-C", "$HOME/lab/SmTIAS-LightSim"])
assert "送信完了" in capsys.readouterr().out
def test_only_streams_just_the_path(self, ssh, project):
ssh.responses.append({"rc": 0})
assert main(["push", "--local-dir", str(project), "--only", "configs/_tmp_x"]) == 0
assert _tar_names(ssh.calls[0].stdin.captured) == ["configs/_tmp_x/s.yaml"]
def test_compress(self, ssh, project):
ssh.responses.append({"rc": 0})
assert main(["push", "--local-dir", str(project), "--compress"]) == 0
proc = ssh.calls[0]
assert proc.stdin.captured[:2] == b"\x1f\x8b"
assert set(_tar_names(proc.stdin.captured)) == _EXPECTED_PUSH
assert "-xzf" in _shell_words(proc.argv[-1])
def test_ssh_failure_is_returned(self, ssh, capsys, project):
ssh.responses.append({"rc": 255})
assert main(["push", "--local-dir", str(project)]) == 255
assert "ssh" in capsys.readouterr().err
def test_prune_removes_only_stale_code_files(self, ssh, project):
"""--prune はサーバの src/ tests/ scripts/ にあって手元に無いファイルだけを消す．"""
remote_files = [
"src/sm_tias_lightsim/__init__.py",
"src/sm_tias_lightsim/a.py",
"src/sm_tias_lightsim/old.py",
"src/sm_tias_lightsim/__pycache__/old.cpython-312.pyc",
"tests/test_old.py",
"scripts/s.py",
"scripts/gone.py",
]
ssh.responses += [
{"rc": 0},
{"rc": 0, "stdout": b"".join(f.encode() + b"\0" for f in remote_files)},
{"rc": 0},
]
assert main(["push", "--local-dir", str(project), "--prune"]) == 0
assert len(ssh.calls) == 3
listing = _shell_words(ssh.calls[1].argv[-1])
assert _has_sequence(listing, ["in", "src", "tests", "scripts"])
assert "output" not in listing
assert "configs" not in listing
remove = ssh.calls[2]
assert _has_sequence(_shell_words(remove.argv[-1]), ["xargs", "-0", "rm", "-f", "--"])
assert remove.input == (
b"scripts/gone.py\0src/sm_tias_lightsim/old.py\0tests/test_old.py\0"
)
def test_prune_with_nothing_stale(self, ssh, capsys, project):
ssh.responses += [{"rc": 0}, {"rc": 0, "stdout": b"src/sm_tias_lightsim/a.py\0"}]
assert main(["push", "--local-dir", str(project), "--prune"]) == 0
assert len(ssh.calls) == 2
assert "削除対象はありません" in capsys.readouterr().out
def test_prune_dry_run_lists_targets_without_deleting(self, ssh, capsys, project):
ssh.responses.append({"rc": 0, "stdout": b"tests/test_old.py\0"})
assert main(["push", "--local-dir", str(project), "--prune", "--dry-run"]) == 0
assert len(ssh.calls) == 1 # 一覧の取得だけ（送信・削除はしない）
assert "tests/test_old.py" in capsys.readouterr().out
@pytest.mark.parametrize(
"listing",
[
b"src/x.py\0output/r/luminance.npy\0",
b"src/x.py\0configs/_tmp_x/s.yaml\0",
b"configs/scene.yaml\0output/_remote_logs/a.log\0",
b"src/../output/x\0",
b"/etc/passwd\0",
],
)
def test_prune_aborts_cleanly_on_unexpected_remote_paths(self, ssh, capsys, project, listing):
"""一覧に output/・configs/ 等のパスが紛れたら，削除コマンドを送らずに中止すること．
compute_prune_targets が RuntimeError を出す状況でも，トレースバックにせず
「削除を中止しました」旨を stderr に出して終了コード 1 で終わる．
（main が例外を送出した場合はこのテスト自体がエラーになる＝不合格）
"""
ssh.responses += [{"rc": 0}, {"rc": 0, "stdout": listing}, {"rc": 0}]
rc = main(["push", "--local-dir", str(project), "--prune"])
assert rc == 1
assert len(ssh.calls) == 2 # 送信と一覧取得だけ．削除（xargs rm）は送らない
assert all("xargs" not in _shell_words(c.argv[-1]) for c in ssh.calls)
assert "削除を中止" in capsys.readouterr().err
def test_prune_dry_run_aborts_cleanly_on_unexpected_remote_paths(self, ssh, capsys, project):
"""--dry-run --prune でも同じく，トレースバックにせず中止を伝えて 1 を返すこと．"""
ssh.responses.append({"rc": 0, "stdout": b"src/x.py\0output/r/luminance.npy\0"})
rc = main(["push", "--local-dir", str(project), "--prune", "--dry-run"])
assert rc == 1
assert len(ssh.calls) == 1
assert "削除を中止" in capsys.readouterr().err
