from __future__ import annotations

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

ROOT = Path(__file__).resolve().parent
SOURCE = (ROOT / "auto_service8700_manager.py").read_text(encoding="utf-8")


def _config(name: str) -> dict:
    return json.loads((ROOT / name).read_text(encoding="utf-8"))


def _load_module():
    spec = importlib.util.spec_from_file_location("auto8700_v235_runtime", ROOT / "auto_service8700_manager.py")
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def test_version_and_retired_service_removed() -> None:
    assert '"version": "2.3.8.235"' in SOURCE
    assert 'direct_schedule_activation' in SOURCE
    assert 'ws_user_mouse_idle_10s' not in SOURCE
    for cfg_name in ["auto_service8700_manager.default.json", "auto_service8700_manager.json"]:
        cfg = _config(cfg_name)
        assert cfg["schema"] == "auto_service8700_manager_v235"
        assert cfg["version"] == "2.3.8.235"
        assert cfg["version_name"] == "2.3.8.235"
        assert cfg["build_tag"] == "v235"
        names = {str(x.get("name", "")).upper() for x in cfg.get("services", [])}
        assert "HEART8772" not in names
        ws = next(x for x in cfg["services"] if str(x.get("name", "")).upper() == "WS8771")
        forbidden = {"ws8771_heart_client.py", "ws8771_ig_auto_click_download.py", "ws8771_ig_auto_download.py", "ws8771_origin_crosscheck.py"}
        assert not forbidden.intersection(ws.get("watch_files", []))
        assert "Heart8772" not in (cfg.get("bat_catalog") or {})
        assert "Heart8772" not in (cfg.get("shared_watch") or {}).get("restart_services", [])


def test_analyze8780_is_auto_started_and_http_health_is_one_time() -> None:
    for cfg_name in ["auto_service8700_manager.default.json", "auto_service8700_manager.json"]:
        cfg = _config(cfg_name)
        service = next(x for x in cfg["services"] if str(x.get("name", "")).upper() == "ANALYZE8780")
        assert service["enabled"] is True
        assert service["start_on_manager_start"] is True
        assert service["ports"] == [8780] and service["health_ports"] == [8780]
        assert service["health_mode"] == "listen_only"
        assert service["startup_api_health_once"] is True
        assert service["runtime_health_check"] is False
        assert service["health_path"] == "/api/health"
        assert service["health_expect_key"] == "ok"
        assert service["health_expect_value"] is True
        assert service["work_dir"] == r"D:\_St\Analyze8780"
        assert "analyze8780_main.py" in service["bat_text"]
        assert "Analyze8780" in cfg["bat_catalog"]
        assert "Analyze8780" in cfg["shared_watch"]["restart_services"]
    for token in [
        'analyze8780_api_checked_once',
        'analyze8780_api_checked_at',
        'analyze8780_port_ok',
        'analyze8780_api_ok',
        '8780:{analyze_text}',
        '확인 완료 뒤에는 1초 상태 루프에서 HTTP·TCP health를 반복하지 않는다',
        'runtime_health_check',
    ]:
        assert token in SOURCE


def test_analyze8780_http_health_called_once_while_port_stays_open(monkeypatch) -> None:
    module = _load_module()
    cfg = module.default_config()
    calls = []
    tcp_calls = []

    monkeypatch.setattr(module, "chrome_debug_alive", lambda _cfg: True)
    monkeypatch.setattr(module, "ws_service_alive", lambda _cfg: True)
    monkeypatch.setattr(module, "ig_chrome_configured", lambda _cfg: True)
    monkeypatch.setattr(module, "chrome_cfg", lambda _cfg: {"profiles": [{"name": "P1"}]})
    def fake_tcp_connect_ok(host, port, timeout=0.5):
        tcp_calls.append((host, int(port)))
        return int(port) == 8780

    monkeypatch.setattr(module, "tcp_connect_ok", fake_tcp_connect_ok)
    monkeypatch.setattr(module, "chrome_cdp_mode_status", lambda _cfg: {})

    def fake_http_json_get(url, timeout=1.0):
        calls.append(url)
        return {"ok": True, "service": "Analyze8780", "version": "2.3.8.228"}

    monkeypatch.setattr(module, "http_json_get", fake_http_json_get)
    with module.AUTO8700_STATUS_CACHE_LOCK:
        module.AUTO8700_STATUS_CACHE["analyze8780_api_checked_once"] = False
        module.AUTO8700_STATUS_CACHE["analyze8780_api_checked_at"] = 0.0
        module.AUTO8700_STATUS_CACHE["analyze8780_api_ok"] = False
        module.AUTO8700_STATUS_CACHE["analyze8780_version"] = ""
        module.AUTO8700_STATUS_CACHE["analyze8780_error"] = ""

    module.refresh_auto8700_status_cache(cfg)
    module.refresh_auto8700_status_cache(cfg)
    module.refresh_auto8700_status_cache(cfg)
    assert len(calls) == 1
    assert len(tcp_calls) == 1
    assert calls[0].endswith("/api/health")


def test_chrome_and_instagram_tabs_are_not_terminated() -> None:
    for token in [
        "kill_all_chrome_for_cdp_mode",
        "cleanup_remembered_procs",
        "remember_launched_proc",
        "AUTO8700_LAUNCHED_PROCS",
        'kill_ports({"name": "IG_CHROME_DEBUG"',
    ]:
        assert token not in SOURCE
    assert "Chrome/CDP 유지: 탭·창·프로세스 종료 안 함" in SOURCE


def test_source_does_not_restore_retired_heart_service() -> None:
    for token in ['"name": "Heart8772"', 'D:\\_St\\Heart8772', 'heart8772_position.py', 'ws8771_heart_client.py', 'ws8771_ig_auto_click_download.py']:
        assert token not in SOURCE


def test_generic_http_json_health_mode_runtime(monkeypatch) -> None:
    module = _load_module()
    cfg = module.default_config()
    assert cfg["version"] == "2.3.8.235"
    assert any(str(x.get("name", "")).upper() == "ANALYZE8780" for x in cfg["services"])
    monkeypatch.setattr(module, "http_json_get", lambda url, timeout=1.2: {"ok": True, "service": "Analyze8780"})
    service = {
        "health_ports": [8780],
        "health_mode": "http_json",
        "health_path": "/api/health",
        "health_expect_key": "ok",
        "health_expect_value": True,
    }
    assert module.health_ok(service) is True
    monkeypatch.setattr(module, "http_json_get", lambda url, timeout=1.2: {"ok": False})
    assert module.health_ok(service) is False


def test_cdp_download_root_is_fixed_sibling_stdown_and_failure_blocks_download(tmp_path, monkeypatch) -> None:
    module = _load_module()
    auto_dir = tmp_path / "bundle" / "_St" / "Auto8700"
    auto_dir.mkdir(parents=True)
    monkeypatch.setattr(module, "script_dir", lambda: auto_dir)
    monkeypatch.setattr(module, "chrome_browser_websocket_url", lambda _cfg: "ws://127.0.0.1/devtools/browser/test")
    monkeypatch.setattr(module, "log", lambda *args, **kwargs: None)
    calls = []

    def fake_cdp_call(ws_url, method, params, timeout=4.0):
        calls.append((ws_url, method, params, timeout))
        return {"ok": True, "result": {}}

    monkeypatch.setattr(module, "cdp_call", fake_cdp_call)
    state = module.apply_instagram_download_behavior(module.default_config(), reason="test")
    expected = tmp_path / "bundle" / "_StDown" / "instagram"
    assert state["ok"] is True
    assert Path(state["download_path"]) == expected
    assert calls == [(
        "ws://127.0.0.1/devtools/browser/test",
        "Browser.setDownloadBehavior",
        {"behavior": "allow", "downloadPath": str(expected), "eventsEnabled": True},
        4.0,
    )]
    assert expected.is_dir()
    assert '/api/cdp/download-root' in SOURCE
    assert '/api/cdp/download-root/ensure' in SOURCE
    for cfg_name in ["auto_service8700_manager.default.json", "auto_service8700_manager.json"]:
        assert _config(cfg_name)["cdp_mode_policy"]["auto_download_allowed_without_cdp"] is False


def test_auto8700_ui_runs_on_main_thread_without_daemon_ui_thread(monkeypatch) -> None:
    module = _load_module()
    calls = []

    class DummyUI:
        def __init__(self, cfg):
            calls.append(("init", module.threading.current_thread().name))

        def run(self):
            calls.append(("run", module.threading.current_thread().name))

    monkeypatch.setattr(module, "tk", object())
    monkeypatch.setattr(module, "auto8700_ui_cfg", lambda cfg: {"enabled": True})
    monkeypatch.setattr(module, "Auto8700FloatingUI", DummyUI)
    module.run_auto8700_floating_ui_main_thread(module.default_config())
    assert calls == [("init", "MainThread"), ("run", "MainThread")]
    assert 'name="auto8700-floating-ui"' not in SOURCE
    assert 'name="auto8700-manager-worker"' in SOURCE
    assert '[UI] WINDOW_VISIBLE' in SOURCE
    assert '[UI] GEOMETRY_APPLY' in SOURCE
    assert '[UI] VERIFY_BEGIN' in SOURCE
    assert '[UI] FORCE_VISIBLE' in SOURCE


def test_close_button_requests_master_shutdown(monkeypatch) -> None:
    module = _load_module()
    calls = []

    class Root:
        def winfo_x(self): return 10
        def winfo_y(self): return 20
        def destroy(self): calls.append("destroy")

    class FakeUI:
        cfg = {}
        root = Root()
        stage = 0
        def ui_cfg(self): return {"enabled": True}

    monkeypatch.setattr(module, "save_config", lambda cfg: calls.append("save"))
    monkeypatch.setattr(module, "request_manager_shutdown", lambda cfg, reason="": calls.append(reason))
    module.Auto8700FloatingUI.close_ui(FakeUI())
    assert "ui_x_close" in calls
    assert "destroy" in calls


def test_master_shutdown_closes_services_but_keeps_chrome(monkeypatch) -> None:
    module = _load_module()
    stopped = []
    killed = []
    monkeypatch.setattr(module, "stop_process", lambda proc, name, cfg: stopped.append(name))
    monkeypatch.setattr(module, "kill_ports", lambda service, cfg, reason: killed.append((service["name"], reason)))
    monkeypatch.setattr(module, "log", lambda *args, **kwargs: None)

    services = {
        "WS8771": {"name": "WS8771", "ports": [8771]},
        "Analyze8780": {"name": "Analyze8780", "ports": [8780]},
        "RTC8790_8791": {"name": "RTC8790_8791", "ports": [8790, 8791]},
        "AUTO_SERVICE8700_MANAGER": {"name": "AUTO_SERVICE8700_MANAGER", "ports": []},
        "IG_CHROME_DEBUG": {"name": "IG_CHROME_DEBUG", "ports": [8700]},
    }
    managed = {name: module.ManagedProc(name=name, process=None) for name in services}
    module.stop_all_managed_services(services, managed, {})
    assert stopped == ["WS8771", "Analyze8780", "RTC8790_8791"]
    assert [name for name, _ in killed] == stopped
    assert "IG_CHROME_DEBUG" not in stopped


def test_startup_geometry_is_forced_inside_primary_work_area(monkeypatch) -> None:
    module = _load_module()
    monkeypatch.setattr(module, "_igblue_primary_work_area", lambda root=None: {"left": 0, "top": 0, "right": 1920, "bottom": 1040})
    w, h, x, y, area = module.clamp_tk_window_geometry_primary(None, 520, 42, 262, 1080, 360, 42, margin=10)
    assert area == {"left": 0, "top": 0, "right": 1920, "bottom": 1040}
    assert (w, h, x, y) == (520, 42, 262, 988)
    assert y + h <= area["bottom"] - 10


def test_ui_runtime_diagnostic_tokens_are_persisted() -> None:
    for token in [
        "[UI] CONFIG",
        "[UI] LAYOUT_BEGIN",
        "[UI] IDLETASKS_DONE",
        "[UI] DEICONIFY_DONE",
        "[UI] LIFT_DONE",
        "[UI] WINDOW_VISIBLE snapshot=",
        "[UI] VERIFY_BEGIN",
        "[UI] VERIFY_OK",
        "[UI] FORCE_VISIBLE",
        "win32_rect",
        "primary_work_area",
    ]:
        assert token in SOURCE
