# -*- coding: utf-8 -*-
# FILE: _St/Cache8701/test_cache8701_module.py | ROLE: Cache8701 v304 자동 저장·선택 로그 정적·단위 테스트
import importlib.util
from pathlib import Path
import json

PATH=Path(__file__).with_name("cache8701_main.py")
spec=importlib.util.spec_from_file_location("cache8701_main",PATH); mod=importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)

def test_normalize_url_keeps_query_and_drops_fragment():
    assert mod.normalize_url("HTTPS://A.COM/x.jpg?a=1#z")=="https://a.com/x.jpg?a=1"

def test_page_type_mapping():
    assert mod.page_type_from_url("https://www.instagram.com/reel/ABC/")=="single_post"
    assert mod.page_type_from_url("https://www.instagram.com/user/reels/")=="profile_reels"
    assert mod.page_type_from_url("https://www.instagram.com/user/")=="profile"

def test_iter_observations_uses_img_origin():
    payload={"source_url":"https://www.instagram.com/user/reels/","info":[{"code":"ABC","username":"user","img_origin":"https://cdn/x.jpg","play_count":7}]}
    obs=list(mod.iter_observations(payload)); assert len(obs)==1
    assert obs[0]["shortcode"]=="ABC" and obs[0]["page_type"]=="profile_reels" and obs[0]["metrics"]["views"]==7

def test_extension_signature_fallback():
    mod.CONFIG=mod.load_config()
    assert mod.extension_for("",b"\xff\xd8\xff"+b"x"*100)==".jpg"
    webp = b"RIFF" + (80).to_bytes(4, "little") + b"WEBP" + b"x" * 80
    assert mod.extension_for("image/jpeg", webp) == ".webp"


def test_auto8700_connection_url_and_signature():
    mod.CONFIG = mod.load_config()
    assert mod.auto8700_connection_url() == "http://127.0.0.1:8707/api/cdp/connection"
    connection = {"cdp_host":"127.0.0.1","cdp_port":8700,"owner_target_id":"A","owner_generation":3,"targets":[{"id":"B"},{"id":"A"}]}
    sig1 = mod.connection_signature(connection)
    connection["targets"] = [{"id":"A"},{"id":"B"}]
    assert mod.connection_signature(connection) == sig1


def test_cdp_command_can_carry_flattened_session_id():
    sent = []
    cdp = mod.SimpleCDPWebSocket("ws://127.0.0.1:8700/devtools/browser/test")
    cdp.send_text = lambda text: sent.append(json.loads(text))
    command_id = cdp.send_cmd("Network.enable", {}, session_id="SESSION-1")
    assert command_id == 1
    assert sent[0]["sessionId"] == "SESSION-1"
    assert sent[0]["method"] == "Network.enable"


def test_config_uses_auto8700_and_real_cdp_default():
    config = mod.load_config()
    assert config["schema"] == "cache8701_config_v2"
    assert config["version"] == "2.3.8.324"
    assert config["auto8700_port"] == 8707
    assert config["cdp_port"] == 8700
    assert config["run_timeout_sec"] == 90
    assert config["file_logging"] is False
    assert config["log_keep_files"] == 20
    assert all(config["page_types"].get(key) is True for key in ("explore", "single_post", "profile", "profile_reels"))


def test_wait_result_preserves_interleaved_session_event():
    class FakeCDP:
        def __init__(self):
            self.rows = [
                json.dumps({"method":"Network.loadingFinished","sessionId":"S1","params":{"requestId":"R1"}}),
                json.dumps({"id":7,"result":{"ok":True}}),
            ]
            self.kept = []
        def recv_text(self, timeout=1.0):
            return self.rows.pop(0) if self.rows else None
        def keep_message(self, data):
            self.kept.append(data)
    fake = FakeCDP()
    result = mod.wait_result(fake, 7, timeout=0.5)
    assert result["id"] == 7
    assert fake.kept[0]["method"] == "Network.loadingFinished"


def test_save_body_uses_shortcode_actual_extension_and_registers_duplicate(tmp_path):
    original_root = mod.ROOT
    original_config = mod.CONFIG
    try:
        st_root = tmp_path / "_St"
        st_root.mkdir()
        mod.ROOT = st_root
        mod.CONFIG = mod.load_config()
        raw = b"RIFF" + (80).to_bytes(4, "little") + b"WEBP" + b"x" * 80
        obs = {
            "account": "ameliauusa",
            "shortcode": "ABC123",
            "post_url": "https://www.instagram.com/reel/ABC123/",
            "metrics": {},
        }
        first = mod.save_body(obs, "image/webp", raw, "https://cdn.example/ABC123.webp", "disk_cache")
        assert first["ok"] is True and first["registered"] is True
        assert Path(first["file_path"]).name == "ABC123.webp"
        assert first["post_id"] > 0 and first["asset_id"] > 0
        from Shared.storage.sort_storage import connect_database, default_database_path
        with connect_database(default_database_path(st_root)) as conn:
            assert conn.execute("SELECT COUNT(*) FROM post_stat_10m").fetchone()[0] == 0
        second = mod.save_body(obs, "image/webp", raw, "https://cdn.example/ABC123.webp", "disk_cache")
        assert second["duplicate"] is True and second["registered"] is True
        assert second["asset_id"] == first["asset_id"]
    finally:
        mod.ROOT = original_root
        mod.CONFIG = original_config


def test_cache_run_counts_new_duplicate_and_failure():
    original_runs = mod.CACHE_RUNS
    try:
        mod.CACHE_RUNS = {}
        base = {"run_id": "RUN303", "shortcode": "A"}
        mod.cache_run_register_target(base)
        mod.cache_run_register_target(base)
        mod.cache_run_register_target({"run_id": "RUN303", "shortcode": "B"})
        snap = mod.cache_run_snapshot("RUN303")
        assert snap["target"] == 2 and snap["processed"] == 0
        mod.cache_run_record(base, {"ok": True})
        mod.cache_run_record({"run_id": "RUN303", "shortcode": "B"}, {"ok": True, "duplicate": True})
        snap = mod.cache_run_snapshot("RUN303")
        assert snap["complete"] is True
        assert snap["new_saved"] == 1 and snap["duplicate_skipped"] == 1 and snap["failed"] == 0
    finally:
        mod.CACHE_RUNS = original_runs


def test_cache_save_creates_only_shortcode_file_and_minimal_index(tmp_path):
    original_root = mod.ROOT
    original_config = mod.CONFIG
    try:
        st_root = tmp_path / "_St"
        st_root.mkdir()
        mod.ROOT = st_root
        mod.CONFIG = mod.load_config()
        raw = b"\xff\xd8\xff" + b"x" * 200
        obs = {"account": "livingdaum", "shortcode": "CODE303", "post_url": "https://www.instagram.com/reel/CODE303/", "metrics": {}}
        result = mod.save_body(obs, "image/jpeg", raw, "https://cdn.example/original_name.jpg", "disk_cache")
        account_dir = tmp_path / "_StDown" / "instagram" / "livingdaum"
        assert Path(result["file_path"]).name == "CODE303.jpg"
        assert (account_dir / "index.htm").is_file()
        assert not list(account_dir.glob("*.md5"))
        assert not list(account_dir.glob("*.json"))
        assert not list(account_dir.glob("*.html"))
    finally:
        mod.ROOT = original_root
        mod.CONFIG = original_config


def test_owner_inactive_does_not_stop_cache_target_watch():
    connection = {
        "cdp_alive": True,
        "owner_active": False,
        "owner_target_id": "OWNER-A",
        "targets": [{"id": "OWNER-A", "url": "https://www.instagram.com/user/"}],
    }
    assert mod.connection_can_watch(connection) is True
    assert mod.connection_can_watch({"cdp_alive": True, "owner_active": False, "targets": []}) is False
    assert mod.connection_can_watch({"cdp_alive": False, "owner_active": True, "targets": [{"id": "A"}]}) is False


def test_file_logging_creates_one_file_and_keeps_total_20(tmp_path, monkeypatch):
    original_config = mod.CONFIG
    try:
        mod._close_file_log()
        log_dir = tmp_path / "logs"
        log_dir.mkdir()
        for index in range(20):
            path = log_dir / f"cache8701_log_20260101_0000{index:02d}_pid{index}.log"
            path.write_text(str(index), encoding="utf-8")
            # 가장 오래된 파일이 index=0이 되도록 mtime을 순서대로 둔다.
            import os, time
            os.utime(path, (time.time() - (1000 - index), time.time() - (1000 - index)))
        monkeypatch.setattr(mod, "_cache_log_dir", lambda: log_dir)
        mod.CONFIG = {"file_logging": True, "log_keep_files": 20}
        created = Path(mod.set_file_logging(True, "pytest"))
        files = sorted(log_dir.glob("cache8701_log_*.log"))
        assert created.is_file()
        assert len(files) == 20
        assert not (log_dir / "cache8701_log_20260101_000000_pid0.log").exists()
        mod.check_log("CI-T", "체크완료", "테스트 로그 기록", "once")
        assert "[캐시이미지][CI-T][체크완료]" in created.read_text(encoding="utf-8")
    finally:
        mod._close_file_log()
        mod.CONFIG = original_config


def test_tab_scoped_observation_prevents_cross_tab_mix():
    mod.CONFIG = mod.load_config()
    mod.OBSERVED.clear(); mod.CACHE_RUNS.clear(); mod.LOG_ONCE.clear()
    a={"account":"moyasi_jm","shortcode":"AAA","image_url":"https://cdn.example/x.jpg","post_url":"","page_type":"profile","metrics":{},"tab_id":101,"run_id":"side-101-1","observed_at":1}
    b={"account":"other","shortcode":"BBB","image_url":"https://cdn.example/x.jpg","post_url":"","page_type":"profile","metrics":{},"tab_id":202,"run_id":"side-202-1","observed_at":1}
    assert mod.accept_observation(a)
    assert mod.accept_observation(b)
    assert mod._observation_for_session_url(101, a["image_url"], "https://www.instagram.com/moyasi_jm/")["shortcode"] == "AAA"
    assert mod._observation_for_session_url(202, b["image_url"], "https://www.instagram.com/other/")["shortcode"] == "BBB"
    assert mod._observation_for_session_url(0, a["image_url"], "https://www.instagram.com/unknown/") is None