cmautobuy/client/test/test_packaging.py

"""Client 打包目录和发布清单测试。"""

import hashlib
import json
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from build_tools.launcher import (
    UpdateApplyError,
    app_executable,
    apply_pending_update,
    finalize_applied_update,
    install_root,
    is_main_program_running,
)
from build_tools.release_manifest import (
    MANIFEST_FILE_NAME,
    versioned_manifest_file_name,
    write_manifest,
)
from src.db import data_dir


class LauncherPathTest(unittest.TestCase):
    def test_main_program_is_inside_app_directory(self):
        root = Path("D:/CMAutoBuy")
        self.assertEqual(
            app_executable(root),
            root / "app" / "CMAutoBuy.exe",
        )

    def test_install_root_is_launcher_parent(self):
        self.assertEqual(
            install_root("D:/CMAutoBuy/Launcher.exe"),
            Path("D:/CMAutoBuy"),
        )


class LauncherUpdateTest(unittest.TestCase):
    def setUp(self):
        self.temporary_directory = tempfile.TemporaryDirectory()
        self.root = Path(self.temporary_directory.name)
        self.current_app = self.root / "app"
        self.staged_app = self.root / "data" / "update" / "app.new"
        self.current_app.mkdir(parents=True)
        self.staged_app.mkdir(parents=True)
        (self.current_app / "CMAutoBuy.exe").write_bytes(b"old")
        (self.staged_app / "CMAutoBuy.exe").write_bytes(b"new")
        self.pending_path = self.root / "data" / "update" / "pending.json"
        self.pending_path.write_text(
            json.dumps(
                {
                    "schema_version": 1,
                    "state": "ready",
                    "version": "0.2.0",
                }
            ),
            encoding="utf-8",
        )

    def tearDown(self):
        self.temporary_directory.cleanup()

    def test_apply_keeps_old_app_until_health_is_confirmed(self):
        version = apply_pending_update(self.root)

        self.assertEqual(version, "0.2.0")
        self.assertEqual((self.root / "app" / "CMAutoBuy.exe").read_bytes(), b"new")
        self.assertEqual((self.root / "app.old" / "CMAutoBuy.exe").read_bytes(), b"old")
        pending = json.loads(self.pending_path.read_text(encoding="utf-8"))
        self.assertEqual(pending["state"], "applied")

        healthy_path = self.root / "data" / "update" / "healthy.json"
        healthy_path.write_text('{"version":"0.2.0"}', encoding="utf-8")
        finalize_applied_update(self.root)
        self.assertFalse(self.pending_path.exists())
        self.assertTrue((self.root / "app.old").exists())

    def test_applied_without_health_rolls_back_on_next_launch(self):
        apply_pending_update(self.root)

        result = apply_pending_update(self.root)

        self.assertIsNone(result)
        self.assertEqual((self.root / "app" / "CMAutoBuy.exe").read_bytes(), b"old")
        self.assertFalse(self.pending_path.exists())
        self.assertTrue((self.root / "data" / "update" / "app.failed").exists())

    def test_move_failure_restores_current_app(self):
        real_replace = os.replace

        def fail_for_staged_app(source, destination):
            if Path(source).name == "app.new":
                raise OSError("simulated lock")
            return real_replace(source, destination)

        with patch("build_tools.launcher.os.replace", side_effect=fail_for_staged_app):
            with self.assertRaises(UpdateApplyError):
                apply_pending_update(self.root)

        self.assertEqual((self.root / "app" / "CMAutoBuy.exe").read_bytes(), b"old")
        self.assertEqual((self.staged_app / "CMAutoBuy.exe").read_bytes(), b"new")

    def test_running_pid_must_belong_to_expected_executable(self):
        running_path = self.root / "data" / "update" / "running.json"
        running_path.write_text('{"pid":123}', encoding="utf-8")

        self.assertTrue(
            is_main_program_running(
                self.root,
                checker=lambda pid, executable: pid == 123
                and executable == self.root / "app" / "CMAutoBuy.exe",
            )
        )
        self.assertFalse(
            is_main_program_running(self.root, checker=lambda _pid, _path: False)
        )
        self.assertFalse(running_path.exists())

    def test_corrupt_pending_state_stops_update_instead_of_being_ignored(self):
        self.pending_path.write_text("not-json", encoding="utf-8")

        with self.assertRaises(UpdateApplyError):
            apply_pending_update(self.root)

        self.assertEqual((self.current_app / "CMAutoBuy.exe").read_bytes(), b"old")


class PackagedDataPathTest(unittest.TestCase):
    def test_packaged_main_program_uses_root_data_directory(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            executable = root / "app" / "CMAutoBuy.exe"
            with patch("src.db.sys.frozen", True, create=True), patch(
                "src.db.sys.executable", str(executable)
            ):
                self.assertEqual(data_dir(), root / "data")


class ReleaseManifestTest(unittest.TestCase):
    def test_manifest_name_and_hash_match_release_files(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            update_zip = root / "CMAutoBuy_0.1.0.zip"
            portable_zip = root / "自动采集采购工具.zip"
            update_zip.write_bytes(b"update")
            portable_zip.write_bytes(b"portable")
            output = root / MANIFEST_FILE_NAME

            write_manifest("0.1.0", update_zip, portable_zip, output)

            self.assertEqual(output.name, "autobuy_manifest.json")
            manifest = json.loads(output.read_text(encoding="utf-8"))
            self.assertEqual(manifest["version"], "0.1.0")
            self.assertEqual(manifest["update"]["size"], len(b"update"))
            self.assertEqual(
                manifest["update"]["sha256"],
                hashlib.sha256(b"update").hexdigest(),
            )
            self.assertEqual(manifest["update"]["file"], "CMAutoBuy_0.1.0.zip")
            self.assertEqual(manifest["portable"]["file"], portable_zip.name)

    def test_versioned_manifest_name_is_stable_and_validated(self):
        self.assertEqual(
            versioned_manifest_file_name("0.1.0"),
            "autobuy_manifest_0.1.0.json",
        )
        with self.assertRaises(ValueError):
            versioned_manifest_file_name("../latest")


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