"""deploy.py 的单测：只测纯逻辑与发布包结构，不碰真实主机。

主机侧 host_deploy.sh 的仿真见 test_host_deploy.py（假 systemctl/curl）。
"""
import hashlib
import os
import subprocess
import sys
import tarfile
import tempfile
import unittest
from unittest import mock

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
    os.path.abspath(__file__))))
sys.path.insert(0, os.path.join(REPO_ROOT, "deploy"))
import deploy as deploy_mod  # noqa: E402


class TestDeployPure(unittest.TestCase):
    def test_release_tarball_name_is_commit_scoped(self):
        name = deploy_mod.release_tarball_name("abc123")
        self.assertEqual(name, "codex-app-server-abc123.tar.gz")
        # 绝不再用通用名：不同 commit 绝不可能撞到同一个远端路径。
        self.assertNotEqual(
            deploy_mod.release_tarball_name("abc123"),
            deploy_mod.release_tarball_name("def456"))

    def test_parse_release_ok_requires_matching_commit(self):
        out = "部署成功\nRELEASE_OK commit=abc sha=dead backup=/tmp/x.tar.gz\n"
        self.assertTrue(deploy_mod.parse_release_ok(out, "abc"))
        self.assertFalse(deploy_mod.parse_release_ok(out, "other"))
        self.assertFalse(deploy_mod.parse_release_ok("部署成功，health 通过\n", "abc"))

    def test_sha256_of(self):
        with tempfile.NamedTemporaryFile(delete=False) as f:
            f.write(b"hello")
            path = f.name
        try:
            self.assertEqual(deploy_mod.sha256_of(path),
                             hashlib.sha256(b"hello").hexdigest())
        finally:
            os.unlink(path)

    def test_tarball_commit_missing_manifest_refuses(self):
        with tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False) as f:
            path = f.name
        try:
            with tarfile.open(path, "w:gz") as tf:
                pass
            with self.assertRaises(SystemExit):
                deploy_mod.tarball_commit(path)
        finally:
            os.unlink(path)

    def test_remote_sha_ignores_ssh_warning_lines(self):
        # 2026-10-27 新链路首推实测：pty helper 把 ssh 的 Warning 行
        # 合并进 stdout 放在 sha 行前面，旧 split()[0] 取到 Warning:
        # 导致传输明明成功却被判验真失败、两路都“失败”。
        sha = "5d0645b85a65b2d58d0d5812f9e8020bdafe15d60dc440e763cfcc7cd178148b"
        fake = subprocess.CompletedProcess(
            args=[], returncode=0,
            stdout="Warning: Permanently added 'x' (ED25519) to the list of known hosts.\n"
                   + sha + "  /tmp/x.tar.gz\n", stderr="")
        with mock.patch.object(deploy_mod.subprocess, "run",
                               return_value=fake):
            self.assertEqual(deploy_mod.remote_sha("h", "u", "/tmp/x"), sha)

    def test_remote_sha_none_without_hash(self):
        fake = subprocess.CompletedProcess(args=[], returncode=0,
                                           stdout="Warning: something\n",
                                           stderr="")
        with mock.patch.object(deploy_mod.subprocess, "run",
                               return_value=fake):
            self.assertIsNone(deploy_mod.remote_sha("h", "u", "/tmp/x"))

    def test_live_commit_ignores_ssh_warning_lines(self):
        commit = "32125f7a980e24fd28664a60701d7c1123beb63e"
        out = "Warning: Permanently added 'x' (ED25519) to the list of known hosts.\n" \
            + commit + "\n"
        self.assertEqual(deploy_mod.live_commit(out), commit)
        self.assertIsNone(deploy_mod.live_commit("Warning: something\n"))


class TestBuildRelease(unittest.TestCase):
    def test_build_from_temp_git_repo(self):
        with tempfile.TemporaryDirectory() as tmp:
            repo = os.path.join(tmp, "repo")
            os.makedirs(repo)
            subprocess.run(["git", "init", "-q"], cwd=repo, check=True)
            subprocess.run(["git", "config", "user.email", "t@t"],
                           cwd=repo, check=True)
            subprocess.run(["git", "config", "user.name", "t"],
                           cwd=repo, check=True)
            with open(os.path.join(repo, "hello.py"), "w") as f:
                f.write("x = 1\n")
            subprocess.run(["git", "add", "."], cwd=repo, check=True)
            subprocess.run(["git", "commit", "-qm", "init"],
                           cwd=repo, check=True)
            commit = subprocess.run(
                ["git", "rev-parse", "HEAD"], cwd=repo, capture_output=True,
                text=True, check=True).stdout.strip()
            # 把 deploy 模块的 REPO_ROOT 临时指到这个小仓库。
            old = deploy_mod.REPO_ROOT
            deploy_mod.REPO_ROOT = repo
            try:
                out_dir = os.path.join(tmp, "out")
                tarball, sha, got = deploy_mod.build_release("HEAD", out_dir)
            finally:
                deploy_mod.REPO_ROOT = old
            self.assertEqual(got, commit)
            self.assertEqual(deploy_mod.sha256_of(tarball), sha)
            self.assertEqual(deploy_mod.tarball_commit(tarball), commit)
            with tarfile.open(tarball, "r:gz") as tf:
                names = tf.getnames()
            self.assertIn("DEPLOYED_COMMIT", names)
            self.assertIn("RELEASE.json", names)
            self.assertIn("hello.py", names)


if __name__ == "__main__":
    unittest.main()