# FILE: _St/Auto8700/test_auto8700_module.py | ROLE: Auto8700 정적·단위 회귀 테스트
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_v250_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_current_services() -> None:
    assert '"version": "2.3.8.281"' 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_v250"
        assert cfg["version"] == "2.3.8.281"
        assert cfg["version_name"] == "2.3.8.281"
        assert cfg["build_tag"] == "v281"
        names = {str(x.get("name", "")).upper() for x in cfg.get("services", [])}
        assert names == {"WS8771", "ANALYZE8780", "RTC8790_8791", "AUTO_SERVICE8700_MANAGER"}
        ws = next(x for x in cfg["services"] if str(x.get("name", "")).upper() == "WS8771")
        forbidden = {"ws8771_ig_auto_click_download.py", "ws8771_ig_auto_download.py", "ws8771_origin_crosscheck.py"}
        assert not forbidden.intersection(ws.get("watch_files", []))


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)
    analyze_calls = [url for url in calls if url.endswith("/api/health")]
    assert len(analyze_calls) == 1
    assert len(tcp_calls) == 1
    assert analyze_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_generic_http_json_health_mode_runtime(monkeypatch) -> None:
    module = _load_module()
    cfg = module.default_config()
    assert cfg["version"] == "2.3.8.281"
    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_and_analyze(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", "RTC8790_8791"]
    assert "Analyze8780" not in stopped
    assert [name for name, _ in killed] == stopped
    assert "IG_CHROME_DEBUG" not in stopped
    assert "상시 분석 서비스 유지" in SOURCE


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


def test_managed_services_always_restart_while_auto8700_is_alive() -> None:
    assert "manual_stopped_services" not in SOURCE
    assert "ws8771_intentional_stop.json" not in SOURCE
    assert "consume_ws_intentional_stop_marker" not in SOURCE
    assert "manager 재실행 전까지 자동 재시작 보류" not in SOURCE
    assert 'restart_service(name, services_by_name, managed, cfg, reason=f"process_exit:{code}")' in SOURCE
    assert 'restart_service(name, services_by_name, managed, cfg, reason="port_health_down")' in SOURCE
    assert "[자동복구][완료]" in SOURCE
    assert "[자동복구][실패]" in SOURCE
    for cfg_name in ["auto_service8700_manager.default.json", "auto_service8700_manager.json"]:
        cfg = _config(cfg_name)
        assert cfg["restart_if_process_exits"] is True
        assert cfg["restart_if_ports_down"] is True
        for service in cfg["services"]:
            if str(service.get("name", "")).upper() == "AUTO_SERVICE8700_MANAGER":
                continue
            assert service.get("start_on_manager_start", True) is True


def test_process_exit_recovery_records_new_pid_and_health(monkeypatch) -> None:
    module = _load_module()
    logs = []

    class DummyProc:
        pid = 4321
        def poll(self): return None

    service = {
        "name": "WS8771",
        "restart_strategy": "kill_then_start",
        "ports": [8771],
        "health_ports": [8771],
        "restart_cooldown_sec": 0,
    }
    managed = {"WS8771": module.ManagedProc(name="WS8771")}
    monkeypatch.setattr(module, "stop_process", lambda *args, **kwargs: None)
    monkeypatch.setattr(module, "kill_ports", lambda *args, **kwargs: None)
    monkeypatch.setattr(module, "start_service", lambda *args, **kwargs: DummyProc())
    monkeypatch.setattr(module, "wait_for_health", lambda *args, **kwargs: None)
    monkeypatch.setattr(module, "health_ok", lambda _service: True)
    monkeypatch.setattr(module, "log", lambda message, cfg=None: logs.append(str(message)))

    module.restart_service("WS8771", {"WS8771": service}, managed, module.default_config(), reason="process_exit:0")
    assert managed["WS8771"].process.pid == 4321
    assert any("[자동복구][완료] service=WS8771 pid=4321 reason=process_exit:0" in line for line in logs)


def test_cdp_new_target_uses_put_and_manual_target_is_brought_to_front(monkeypatch) -> None:
    module = _load_module()
    calls = []
    monkeypatch.setattr(module, "chrome_debug_url", lambda _cfg, path="/json/version": f"http://127.0.0.1:8700{path}")

    def fake_request(url, timeout=1.2, method="GET"):
        calls.append((url, method))
        return {"id": "new-target", "url": "https://www.instagram.com/explore/"}

    monkeypatch.setattr(module, "http_json_request", fake_request)
    assert module.open_chrome_url_via_debug({}, "https://www.instagram.com/explore/") is True
    assert calls and calls[0][1] == "PUT"
    assert "/json/new?" in calls[0][0]


def test_ensure_instagram_tab_requires_runtime_and_can_bring_visible(monkeypatch) -> None:
    module = _load_module()
    tab = {
        "id": "IG1",
        "url": "https://www.instagram.com/explore/",
        "webSocketDebuggerUrl": "ws://127.0.0.1/devtools/page/IG1",
    }
    events = []
    monkeypatch.setattr(module, "chrome_debug_alive", lambda _cfg: True)
    monkeypatch.setattr(module, "chrome_tabs", lambda _cfg: [tab])
    monkeypatch.setattr(module, "choose_instagram_control_tab", lambda _cfg: tab)
    monkeypatch.setattr(module, "chrome_tab_runtime_ready", lambda _tab, timeout=1.8: True)
    monkeypatch.setattr(module, "mark_control_tab_titles", lambda _cfg, _tab, devtools_ok=True: events.append("mark"))
    monkeypatch.setattr(module, "bring_chrome_tab_to_front", lambda _cfg, _tab: events.append("front") or True)
    monkeypatch.setattr(module, "log", lambda *args, **kwargs: None)
    assert module.ensure_instagram_tab(module.default_config(), bring_to_front=True, wait_sec=0.5) == tab
    assert events == ["mark", "front"]


def test_existing_cdp_without_instagram_runtime_is_not_reported_as_success(monkeypatch, tmp_path) -> None:
    module = _load_module()
    chrome = tmp_path / "chrome.exe"
    chrome.write_text("", encoding="utf-8")
    cfg = module.default_config()
    monkeypatch.setattr(module, "chrome_exe_path", lambda _cfg: chrome)
    monkeypatch.setattr(module, "chrome_debug_alive", lambda _cfg: True)
    monkeypatch.setattr(module, "ensure_instagram_tab", lambda _cfg, bring_to_front=False, wait_sec=5.0: None)
    monkeypatch.setattr(module, "stop_dedicated_cdp_chrome", lambda _cfg, reason: False)
    monkeypatch.setattr(module, "log", lambda *args, **kwargs: None)
    assert module.launch_chrome_cdp_user_data_dir(cfg, reason="manual_ui_cdp_user_data_dir", bring_to_front=True) is False
    assert module.AUTO8700_SCHEDULE_STATE["cdp_error"] == "instagram_runtime_unresponsive"


def test_dedicated_cdp_process_filter_never_selects_normal_chrome(monkeypatch) -> None:
    module = _load_module()
    cfg = module.default_config()
    cfg["ig_chrome"]["cdp_user_data_dir"] = r"D:\cdp-data-dir"
    monkeypatch.setattr(module, "is_windows", lambda: True)
    monkeypatch.setattr(module, "query_chrome_processes_windows", lambda: [
        {"ProcessId": 10, "CommandLine": r'chrome.exe --user-data-dir="D:\cdp-data-dir" --remote-debugging-port=8700'},
        {"ProcessId": 11, "CommandLine": r'chrome.exe --user-data-dir="C:\Users\u\AppData\Local\Google\Chrome\User Data"'},
        {"ProcessId": 12, "CommandLine": r'chrome.exe --type=renderer --user-data-dir="D:\cdp-data-dir"'},
    ])
    items = module.dedicated_cdp_chrome_main_processes(cfg)
    assert [int(item["ProcessId"]) for item in items] == [10]


def test_manual_auto_run_requests_visible_cdp_chrome(monkeypatch) -> None:
    module = _load_module()
    calls = []
    monkeypatch.setattr(
        module,
        "launch_chrome_cdp_user_data_dir",
        lambda cfg, reason="", bring_to_front=None: calls.append((reason, bring_to_front)) or True,
    )
    assert module.ensure_ig_chrome_for_schedule(module.default_config(), reason="manual_ui_run_now") is True
    assert calls == [("manual_ui_run_now", True)]


def test_bring_to_front_restores_minimized_cdp_window(monkeypatch) -> None:
    module = _load_module()
    tab = {
        "id": "IG-WINDOW",
        "webSocketDebuggerUrl": "ws://127.0.0.1/devtools/page/IG-WINDOW",
    }
    calls = []
    monkeypatch.setattr(module, "chrome_browser_websocket_url", lambda _cfg: "ws://127.0.0.1/devtools/browser/B1")
    monkeypatch.setattr(module, "activate_chrome_tab", lambda _cfg, _tab_id: False)

    def fake_cdp_call(ws_url, method, params=None, timeout=4.0):
        calls.append((ws_url, method, params or {}))
        if method == "Browser.getWindowForTarget":
            return {"ok": True, "result": {"windowId": 7, "bounds": {"windowState": "minimized"}}}
        return {"ok": True, "result": {}}

    monkeypatch.setattr(module, "cdp_call", fake_cdp_call)
    assert module.bring_chrome_tab_to_front(module.default_config(), tab) is True
    assert ("ws://127.0.0.1/devtools/browser/B1", "Browser.setWindowBounds", {"windowId": 7, "bounds": {"windowState": "normal"}}) in calls
    assert ("ws://127.0.0.1/devtools/page/IG-WINDOW", "Page.bringToFront", {}) in calls


def test_cdp_lifecycle_preload_source_blocks_only_unload_registration() -> None:
    module = _load_module()
    cfg = module.default_config()
    policy = module.cdp_instagram_lifecycle_cfg(cfg)
    assert policy["enabled"] is True
    assert policy["block_unload_registration"] is True
    source = module.cdp_instagram_lifecycle_preload_source({**policy, "session_id": "test-session"})
    assert "Page.addScriptToEvaluateOnNewDocument" not in source
    assert "BLOCKED_REGISTER" in source
    assert "BLOCKED_PROPERTY" in source
    assert "eventType === 'unload'" in source
    assert "['beforeunload', 'pagehide']" in source
    assert "visibilitychange" in source
    assert "yellow_cdp_lifecycle_diag" in source
    assert "session_id\": \"test-session" in source


def test_cdp_lifecycle_preload_payload_registers_before_runtime(monkeypatch) -> None:
    module = _load_module()
    cfg = module.default_config()
    tab = {
        "id": "TARGET-1",
        "url": "http://127.0.0.1:8707/cdp-bootstrap?token=yellow_cdp_bootstrap_v281_test",
        "title": "",
        "type": "page",
        "webSocketDebuggerUrl": "ws://127.0.0.1/devtools/page/TARGET-1",
    }
    calls = []
    monkeypatch.setattr(module, "cdp_find_target_by_exact_url", lambda _cfg, exact_url, timeout=3.0: tab)
    monkeypatch.setattr(module, "log", lambda *args, **kwargs: None)

    def fake_call(ws_url, method, params=None, timeout=4.0):
        calls.append((method, params or {}))
        if method == "Page.addScriptToEvaluateOnNewDocument":
            return {"ok": True, "result": {"identifier": "script-1"}}
        if method == "Runtime.evaluate":
            return {"ok": True, "result": {"result": {"value": {"ok": True, "installed": True}}}}
        return {"ok": False, "error": "unexpected_method"}

    monkeypatch.setattr(module, "cdp_call", fake_call)
    result = module.cdp_ig_install_lifecycle_preload_payload(cfg, {
        "bootstrap_url": tab["url"],
        "target_url": "https://www.instagram.com/example/",
        "session_id": "test-session",
    })
    assert result["ok"] is True
    assert result["identifier"] == "script-1"
    assert [method for method, _params in calls] == ["Page.addScriptToEvaluateOnNewDocument", "Runtime.evaluate"]
    assert "BLOCKED_REGISTER" in calls[0][1]["source"]
    assert calls[1][1]["expression"] == calls[0][1]["source"]


def test_cdp_lifecycle_preload_endpoint_is_local_and_explicit() -> None:
    assert '/api/cdp/ig/install_lifecycle_preload' in SOURCE
    assert 'invalid_bootstrap_url' in SOURCE
    assert '[CDP수명주기][CL-02][체크완료]' in SOURCE
