# FILE: _St/Auto8700/test_auto8700_update_v320.py | ROLE: build 기반 시작 자동업데이트·역행 방지 회귀
from __future__ import annotations

import hashlib
import importlib.util
import sys
import tempfile
import zipfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent
ST_ROOT = ROOT.parent


def _load(name: str, path: Path):
    spec = importlib.util.spec_from_file_location(name, path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


update = _load("st_update8700_test", ROOT / "st_update8700.py")
apply = _load("st_update_apply_test", ROOT / "st_update_apply.py")


def _html(version: str, build: int, sha256: str = "", file: str = "_St.zip") -> str:
    return "\n".join([
        '<meta name="st-version" content="%s">' % version,
        '<meta name="st-build" content="%s">' % build,
        '<meta name="st-file" content="%s">' % file,
        '<meta name="st-sha256" content="%s">' % sha256,
    ])


def test_build_integer_comparison_and_local_newer_blocks_reverse_update() -> None:
    local = update.VersionInfo("2.3.8.400", 400, "_St.zip")
    assert update.decide_update(local, update.VersionInfo("x", 401, "_St.zip"), publisher=False).action == "UPDATE"
    assert update.decide_update(local, update.VersionInfo("x", 400, "_St.zip"), publisher=False).action == "SAME"
    assert update.decide_update(local, update.VersionInfo("x", 399, "_St.zip"), publisher=False).action == "LOCAL_NEWER"


def test_publisher_marker_never_auto_overwrites_even_when_server_is_newer(tmp_path: Path) -> None:
    marker = tmp_path / "AutoDistribute" / "AutoDistribute.py"
    marker.parent.mkdir(parents=True)
    marker.write_text("# publisher", encoding="utf-8")
    local = update.VersionInfo("dev", 100, "_St.zip")
    remote = update.VersionInfo("server", 999, "_St.zip")
    assert update.is_publisher(tmp_path)
    decision = update.decide_update(local, remote, publisher=True)
    assert decision.action == "PUBLISHER_SKIP"


def test_startup_user_update_downloads_verified_zip_and_spawns_temp_updater(tmp_path: Path) -> None:
    st = tmp_path / "_St"
    (st / "Auto8700").mkdir(parents=True)
    (st / "Auto8700" / "st_update_apply.py").write_text((ROOT / "st_update_apply.py").read_text(encoding="utf-8"), encoding="utf-8")
    (st / "Auto8700" / "auto_service8700_manager.py").write_text("# manager", encoding="utf-8")
    (st / "_St_ver.html").write_text(_html("2.3.8.321", 320), encoding="utf-8")

    payload = tmp_path / "payload.zip"
    with zipfile.ZipFile(payload, "w") as zf:
        zf.writestr("_St/_St_ver.html", _html("2.3.8.321", 321))
        zf.writestr("_St/Auto8700/new.py", "# FILE: _St/Auto8700/new.py\n")
    digest = hashlib.sha256(payload.read_bytes()).hexdigest()
    remote_text = _html("2.3.8.321", 321, digest)
    calls: list[tuple[list[str], Path]] = []

    def fake_download(url: str, target: Path, timeout: float) -> Path:
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(payload.read_bytes())
        return target

    def fake_spawn(cmd: list[str], cwd: Path):
        calls.append((cmd, cwd))
        return object()

    logs: list[str] = []
    cfg = {"auto_update": {"enabled": True, "check_on_start": True, "version_url": "http://server/_St_ver.html"}}
    should_exit = update.startup_update_check(
        cfg,
        st_root=st,
        logger=logs.append,
        remote_getter=lambda _url, _timeout: remote_text,
        downloader=fake_download,
        spawner=fake_spawn,
    )
    assert should_exit is True
    assert len(calls) == 1
    assert "--expected-build" in calls[0][0]
    assert "321" in calls[0][0]
    assert any("UP-04" in line and "체크완료" in line for line in logs)


def test_startup_publisher_does_not_call_downloader(tmp_path: Path) -> None:
    st = tmp_path / "_St"
    (st / "AutoDistribute").mkdir(parents=True)
    (st / "AutoDistribute" / "AutoDistribute.py").write_text("# publisher", encoding="utf-8")
    (st / "_St_ver.html").write_text(_html("dev", 1), encoding="utf-8")
    called = {"download": False}

    def no_download(url: str, target: Path, timeout: float):
        called["download"] = True
        raise AssertionError("publisher must not download")

    result = update.startup_update_check(
        {"auto_update": {"version_url": "http://server/_St_ver.html"}},
        st_root=st,
        logger=lambda _msg: None,
        remote_getter=lambda _url, _timeout: _html("server", 999, "a" * 64),
        downloader=no_download,
        spawner=lambda *_args: None,
    )
    assert result is False
    assert called["download"] is False


def test_apply_zip_overlays_code_but_preserves_existing_data(tmp_path: Path) -> None:
    st = tmp_path / "_St"
    (st / "Shared" / "data").mkdir(parents=True)
    data_file = st / "Shared" / "data" / "user.sqlite3"
    data_file.write_text("KEEP", encoding="utf-8")
    old = st / "Auto8700" / "old.py"
    old.parent.mkdir(parents=True)
    old.write_text("old", encoding="utf-8")

    package = tmp_path / "_St.zip"
    with zipfile.ZipFile(package, "w") as zf:
        zf.writestr("_St/_St_ver.html", _html("2.3.8.321", 321))
        zf.writestr("_St/Auto8700/old.py", "new")
        zf.writestr("_St/WS8771/new.py", "new")
    copied = apply.apply_zip(package, st, staging_parent=tmp_path)
    assert copied == 3
    assert old.read_text(encoding="utf-8") == "new"
    assert data_file.read_text(encoding="utf-8") == "KEEP"
    assert apply.read_build(st) == 321



def test_apply_zip_rolls_back_partial_copy_failure(tmp_path: Path, monkeypatch) -> None:
    st = tmp_path / "_St"
    target = st / "Auto8700" / "a.py"
    target.parent.mkdir(parents=True)
    target.write_text("OLD", encoding="utf-8")
    package = tmp_path / "_St.zip"
    with zipfile.ZipFile(package, "w") as zf:
        zf.writestr("_St/Auto8700/a.py", "NEW")
        zf.writestr("_St/WS8771/b.py", "NEW")

    real_copy2 = apply.shutil.copy2
    calls = {"source_copies": 0}

    def flaky_copy2(src, dst, *args, **kwargs):
        src_path = Path(src)
        if "st_update_apply_" in str(src_path) and "__backup__" not in str(src_path):
            calls["source_copies"] += 1
            if calls["source_copies"] >= 2:
                raise OSError("forced copy failure")
        return real_copy2(src, dst, *args, **kwargs)

    monkeypatch.setattr(apply.shutil, "copy2", flaky_copy2)
    try:
        apply.apply_zip(package, st, staging_parent=tmp_path)
    except OSError as exc:
        assert "forced copy failure" in str(exc)
    else:
        raise AssertionError("copy failure must propagate")
    assert target.read_text(encoding="utf-8") == "OLD"
    assert not (st / "WS8771" / "b.py").exists()


def test_server_meta_build_must_match_zip_embedded_build(tmp_path: Path) -> None:
    package = tmp_path / "_St.zip"
    with zipfile.ZipFile(package, "w") as zf:
        zf.writestr("_St/_St_ver.html", _html("2.3.8.321", 320))
        zf.writestr("_St/Auto8700/a.py", "x")
    embedded = update.read_zip_version(package)
    assert embedded.build == 320
    assert embedded.build != 321

def test_update_zip_rejects_path_traversal(tmp_path: Path) -> None:
    package = tmp_path / "bad.zip"
    with zipfile.ZipFile(package, "w") as zf:
        zf.writestr("_St/../evil.txt", "x")
    try:
        update.validate_update_zip(package)
    except update.UpdateError as exc:
        assert "위험 경로" in str(exc)
    else:
        raise AssertionError("path traversal must fail")


def test_manager_checks_update_before_lock_and_reexec_skips_repeat_check() -> None:
    source = (ROOT / "auto_service8700_manager.py").read_text(encoding="utf-8")
    assert '"--skip-update-check"' in source
    main_source = source[source.index("def main("):]
    assert main_source.index("startup_update_check(") < main_source.index("acquire_manager_lock(cfg")
    assert 'args = ["--takeover-pid", str(os.getpid()), "--reason", reason, "--skip-update-check"]' in source