"""Tests for the read-only Content.ggpk extract pipeline.

Live cases read the installed Path of Exile 2 client (Content.ggpk / HashCache.dat).
"""
from __future__ import annotations

import sqlite3
import struct
from pathlib import Path

import pytest

from src.assistant_tool.ggpk_extract.bundle import parse_bundle_header
from src.assistant_tool.ggpk_extract.datc64 import cell_size, parse_datc64
from src.assistant_tool.ggpk_extract.ggpk_fs import GgpkFs
from src.assistant_tool.ggpk_extract.hashcache import parse_hashcache
from src.assistant_tool.ggpk_extract.murmur import hash_bundle_path, murmurhash64a
from src.assistant_tool.ggpk_extract.paths import find_content_ggpk, find_hashcache
from src.assistant_tool.ggpk_extract.pipeline import extract
from src.assistant_tool.ggpk_extract.sqlite_export import snake

REPO = Path(__file__).resolve().parent.parent


def test_murmurhash64a_is_stable():
    sample = b"data/baseitemtypes.datc64"
    assert murmurhash64a(sample) == murmurhash64a(sample)
    assert murmurhash64a(sample) != murmurhash64a(sample + b"x")
    assert hash_bundle_path("Data/BaseItemTypes.datc64") == hash_bundle_path("data/baseitemtypes.datc64")


def test_snake_table_names():
    assert snake("BaseItemTypes") == "base_item_types"
    assert snake("WorldAreas") == "world_areas"
    assert snake("UniqueStashLayout") == "unique_stash_layout"


def test_cell_size_datc64_layout():
    assert cell_size({"type": "i32", "array": False}) == 4
    assert cell_size({"type": "string", "array": False}) == 8
    assert cell_size({"type": "foreignrow", "array": False}) == 16
    assert cell_size({"type": "i32", "array": True}) == 16


def test_datc64_rowcount_and_bb_marker_contract():
    schema = {"name": "ContractI32", "columns": [{"name": "Val", "type": "i32", "array": False}]}
    raw = struct.pack("<I", 2) + struct.pack("<ii", 7, 11) + (b"\xBB" * 8)
    rows = parse_datc64(raw, schema)
    assert [r["Val"] for r in rows] == [7, 11]


def test_datc64_null_array_offset_does_not_overflow():
    schema = {
        "name": "Arr",
        "columns": [{"name": "Refs", "type": "row", "array": True}],
    }
    count_rel = struct.pack("<QQ", 3, 0xFEFEFEFEFEFEFEFE)
    raw = struct.pack("<I", 1) + count_rel + (b"\xBB" * 8)
    rows = parse_datc64(raw, schema)
    assert rows[0]["Refs"] == []


def test_datc64_trims_trailing_schema_wider_than_055_row():
    schema = {
        "name": "Wide",
        "columns": [
            {"name": "Val", "type": "i32", "array": False},
            {"name": None, "type": "i32", "array": False},
        ],
    }
    raw = struct.pack("<I", 1) + struct.pack("<i", 9) + (b"\xBB" * 8)
    rows = parse_datc64(raw, schema)
    assert rows[0]["Val"] == 9
    assert "col_0" not in rows[0]


def test_hashcache_live_client_file():
    hc = find_hashcache()
    if hc is None:
        pytest.skip("HashCache.dat not installed")
    recs = parse_hashcache(hc)
    assert hc.stat().st_size == 2846
    assert len(recs) == 33
    names = {r.name.replace("\\", "/") for r in recs}
    assert any(n.endswith("PathOfExile.exe") or n == "PathOfExile.exe" for n in names)
    exe = next(r for r in recs if r.name.endswith("PathOfExile.exe") or r.name == "PathOfExile.exe")
    assert exe.size in (78319448, 78325080, 78329176)
    assert len(exe.sha256) == 32


def test_ggpk_root_and_bundle_headers_live():
    ggpk = find_content_ggpk()
    if ggpk is None:
        pytest.skip("Content.ggpk not installed")
    assert ggpk.stat().st_size == 161_927_115_259
    with GgpkFs(ggpk) as fs:
        names = [e.name for e in fs.iter_dir_children(fs.root_offset)]
        assert names.count("Bundles2") == 1
        assert "gateway_list.txt" in names
        index = fs.find("Bundles2/_.index.bin")
        data = fs.find("Bundles2/Folders/data.dat.bundle.bin")
        assert index is not None and data is not None
        ih = parse_bundle_header(fs.read_payload_prefix(index, 12 + 4096))
        dh = parse_bundle_header(fs.read_payload_prefix(data, 12 + 4096))
        assert ih.uncompressed_size in (149_123_238, 149_180_854, 149_294_814)
        assert ih.block_count in (569, 570, 571)
        assert fs.read_payload_prefix(index, 2)  # payload not header; check magic via full prefix
        payload = fs.read_payload_prefix(index, ih.payload_offset + 2)
        assert payload[ih.payload_offset : ih.payload_offset + 2] == b"\x8c\x0c"
        assert dh.uncompressed_size in (9_667_076, 9_673_698, 9_677_308)
        assert dh.block_count == 37


def test_catalog_extract_writes_sqlite(tmp_path: Path):
    ggpk = find_content_ggpk()
    if ggpk is None:
        pytest.skip("Content.ggpk not installed")
    db_path = tmp_path / "poe2_ggpk.db"
    schema_path = REPO / "assets" / "game_data" / "schema.min.json"
    stats = extract(
        ggpk_path=ggpk,
        db_path=db_path,
        schema_path=schema_path,
        catalog_only=True,
        log=lambda *_a, **_k: None,
    )
    assert stats["status"] == "catalog-only"
    assert stats["ggpk_records"] >= 10
    conn = sqlite3.connect(str(db_path))
    n = conn.execute("SELECT COUNT(*) FROM ggpk_files").fetchone()[0]
    assert n == stats["ggpk_records"]
    row = conn.execute(
        "SELECT data_size FROM ggpk_files WHERE path = ?",
        ("Bundles2/_.index.bin",),
    ).fetchone()
    assert row is not None and row[0] in (114_953_184, 114_989_715, 115_073_230)
    meta = dict(conn.execute("SELECT key, value FROM extract_meta"))
    assert meta["ggpk_size"] == str(161_927_115_259)
    assert int(meta["index_uncompressed"]) in (149_123_238, 149_180_854, 149_294_814)
    hc_n = conn.execute("SELECT COUNT(*) FROM hashcache_files").fetchone()[0]
    assert hc_n == 33
    if (REPO / "assets" / "game_data" / "schema.min.json").is_file():
        assert meta.get("schema_version") == "7"
    conn.close()


def test_select_canonical_prefers_data_datc64():
    from src.assistant_tool.ggpk_extract.index_bin import LogicalFile
    from src.assistant_tool.ggpk_extract.pipeline import select_canonical_dat_files

    files = [
        LogicalFile("french/data/baseitemtypes.datc64", 1, 0, 0, 10, "fr"),
        LogicalFile("data/baseitemtypes.datc64", 2, 0, 0, 10, "data"),
        LogicalFile("data/tags.datc64", 3, 0, 0, 10, "data"),
    ]
    picked = select_canonical_dat_files(files, ("BaseItemTypes", "Tags"))
    assert [f.path for f in picked] == ["data/baseitemtypes.datc64", "data/tags.datc64"]
    from src.assistant_tool.ggpk_extract.ggpk_fs import GgpkEntry
    from src.assistant_tool.ggpk_extract.pipeline import find_bundle_entry

    entry = GgpkEntry(
        name="data.dat.bundle.bin",
        tag="FILE",
        offset=1,
        record_length=10,
        sha256=b"",
        path="Bundles2/Folders/data.dat.bundle.bin",
        data_size=100,
    )
    assert find_bundle_entry([entry], "data") is entry
    assert find_bundle_entry([entry], "Data/data") is entry


def test_unique_projection_joins_words_key(tmp_path: Path):
    from src.assistant_tool.ggpk_extract.sqlite_export import (
        connect,
        init_catalog_schema,
        write_dat_table,
        write_projections,
    )

    db_path = tmp_path / "proj.db"
    conn = connect(db_path)
    init_catalog_schema(conn)
    write_dat_table(conn, "Words", [{"_row": 0, "Text": "Headhunter"}])
    write_dat_table(
        conn,
        "UniqueStashLayout",
        [{"_row": 0, "WordsKey": 0, "ItemVisualIdentityKey": 7}],
    )
    write_projections(conn)
    row = conn.execute("SELECT id, name, base_item FROM uniques").fetchone()
    assert row == ("0", "Headhunter", "")
    conn.close()