# FILE: _St/Auto8700/test_auto8700_update_hourly_v324.py | ROLE: 시작 1회 + 매 정시 build 자동업데이트 회귀
from __future__ import annotations

import importlib.util
import sys
from datetime import datetime
from pathlib import Path

ROOT = Path(__file__).resolve().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_hourly_test", ROOT / "st_update8700.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_next_hour_uses_local_clock_and_never_rechecks_same_exact_hour() -> None:
    assert update.seconds_until_next_hour(datetime(2026, 8, 14, 17, 2, 0)) == 58 * 60
    assert update.seconds_until_next_hour(datetime(2026, 8, 14, 17, 59, 59, 500000)) == 0.5
    assert update.seconds_until_next_hour(datetime(2026, 8, 14, 18, 0, 0)) == 60 * 60


def test_hourly_check_runs_even_when_startup_check_was_skipped(tmp_path: Path) -> None:
    st = tmp_path / "_St"
    st.mkdir()
    (st / "_St_ver.html").write_text(_html("2.3.8.324", 324), encoding="utf-8")
    logs: list[str] = []
    result = update.hourly_update_check(
        {
            "auto_update": {
                "enabled": True,
                "check_on_start": False,
                "check_hourly": True,
                "version_url": "http://server/_St_ver.html",
            }
        },
        st_root=st,
        logger=logs.append,
        remote_getter=lambda _url, _timeout: _html("2.3.8.324", 324),
        downloader=lambda *_args: (_ for _ in ()).throw(AssertionError("same build must not download")),
        spawner=lambda *_args: (_ for _ in ()).throw(AssertionError("same build must not spawn updater")),
    )
    assert result is False
    assert any("source=hourly" in line and "server_build=324" in line for line in logs)
    assert any("판단=SAME" in line for line in logs)


def test_hourly_publisher_checks_server_but_never_downloads(tmp_path: Path) -> None:
    st = tmp_path / "_St"
    marker = st / "AutoDistribute" / "AutoDistribute.py"
    marker.parent.mkdir(parents=True)
    marker.write_text("# publisher", encoding="utf-8")
    (st / "_St_ver.html").write_text(_html("dev", 324), encoding="utf-8")
    logs: list[str] = []
    result = update.hourly_update_check(
        {"auto_update": {"check_hourly": True, "version_url": "http://server/_St_ver.html"}},
        st_root=st,
        logger=logs.append,
        remote_getter=lambda _url, _timeout: _html("server", 999, "a" * 64),
        downloader=lambda *_args: (_ for _ in ()).throw(AssertionError("publisher must not download")),
        spawner=lambda *_args: (_ for _ in ()).throw(AssertionError("publisher must not spawn updater")),
    )
    assert result is False
    assert any("판단=PUBLISHER_SKIP" in line and "source=hourly" in line for line in logs)


def test_hourly_check_can_be_disabled_without_server_request(tmp_path: Path) -> None:
    st = tmp_path / "_St"
    st.mkdir()
    called = {"remote": False}

    def remote_getter(_url: str, _timeout: float) -> str:
        called["remote"] = True
        return _html("server", 999)

    result = update.hourly_update_check(
        {"auto_update": {"enabled": True, "check_hourly": False}},
        st_root=st,
        remote_getter=remote_getter,
    )
    assert result is False
    assert called["remote"] is False


def test_manager_starts_single_hourly_watch_after_lock_and_shutdowns_for_update() -> None:
    source = (ROOT / "auto_service8700_manager.py").read_text(encoding="utf-8")
    main = source[source.index("def main("):]
    assert '"check_hourly": True' in source
    assert 'name="auto8700-hourly-update-watch"' in source
    assert main.index("acquire_manager_lock(cfg") < main.index("start_auto8700_hourly_update_watch(cfg)")
    assert 'request_manager_shutdown(cfg, reason="hourly_auto_update")' in source
    assert 'AUTO8700_SHUTDOWN_EVENT.wait(wait_sec)' in source
    assert '"--skip-update-check"' in source


def test_hourly_watch_runtime_requests_normal_shutdown_when_update_is_ready(monkeypatch) -> None:
    sys.path.insert(0, str(ROOT))
    try:
        manager = _load("auto_service8700_manager_hourly_test", ROOT / "auto_service8700_manager.py")
    finally:
        try:
            sys.path.remove(str(ROOT))
        except ValueError:
            pass
    manager.AUTO8700_SHUTDOWN_EVENT.clear()
    manager.DEVTOOLS_WATCH_STATE["stop"] = False
    logs: list[str] = []
    monkeypatch.setattr(manager, "log", lambda message, _cfg=None: logs.append(str(message)))
    monkeypatch.setattr(manager, "seconds_until_next_hour", lambda: 0.001)
    monkeypatch.setattr(manager, "hourly_update_check", lambda *_args, **_kwargs: True)

    manager.auto8700_hourly_update_watch_loop({"auto_update": {"enabled": True, "check_hourly": True}})

    assert manager.AUTO8700_SHUTDOWN_EVENT.is_set()
    assert manager.DEVTOOLS_WATCH_STATE["stop"] is True
    assert any("UP-H-02" in line and "updater로 제어 이관" in line for line in logs)
    assert any("reason=hourly_auto_update" in line for line in logs)
    manager.AUTO8700_SHUTDOWN_EVENT.clear()