"""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