"""Bounded pack inflation and delta-resolution contracts for pushed objects."""

from __future__ import annotations

import hashlib
import io
import zlib

import pytest

from testing.tests.test_hermes_scm_broker_support import (
    _load,
    _object_entry,
    _pack_of,
)


def _delta_header(type_code: int, size: int, suffix: bytes = b"") -> bytes:
    byte = (type_code << 4) | (size & 0x0F)
    size >>= 4
    header = bytearray()
    while size:
        header.append(byte | 0x80)
        byte = size & 0x7F
        size >>= 7
    header.append(byte)
    return bytes(header) + suffix


def _size_varint(value: int) -> bytes:
    out = bytearray()
    while True:
        byte = value & 0x7F
        value >>= 7
        if value:
            out.append(byte | 0x80)
        else:
            out.append(byte)
            return bytes(out)


def _ofs_distance(value: int) -> bytes:
    out = [value & 0x7F]
    value >>= 7
    while value:
        value -= 1
        out.insert(0, 0x80 | (value & 0x7F))
        value >>= 7
    return bytes(out)


def _delta(source: bytes, target_size: int, instructions: bytes) -> bytes:
    return _size_varint(len(source)) + _size_varint(target_size) + instructions


def _blob_sha(payload: bytes) -> bytes:
    return hashlib.sha1(b"blob %d\x00" % len(payload) + payload).digest()


def _read_all(objects):
    try:
        return [(code, payload.read()) for code, payload in objects]
    finally:
        for _code, payload in objects:
            payload.close()


def test_unpack_returns_typed_payloads_for_every_base_object_kind():
    module = _load("git_pack_objects")
    entries = [
        _object_entry(1, b"tree 1\n"),
        _object_entry(2, b"\x00tree-bytes"),
        _object_entry(3, b"blob contents"),
        _object_entry(4, b"tag contents"),
    ]
    result = _read_all(module.unpack_objects(io.BytesIO(_pack_of(entries))))
    assert result == [
        (1, b"tree 1\n"),
        (2, b"\x00tree-bytes"),
        (3, b"blob contents"),
        (4, b"tag contents"),
    ]


def test_unpack_accepts_the_empty_pack():
    module = _load("git_pack_objects")
    assert module.unpack_objects(io.BytesIO(_pack_of([]))) == []


def test_ofs_delta_resolves_against_its_in_pack_base():
    module = _load("git_pack_objects")
    base = b"base payload for copying"
    base_entry = _object_entry(3, base)
    # Copy the first 4 bytes, then insert new bytes.
    instructions = bytes([0x80 | 0x10, 4]) + bytes([5]) + b"+tail"
    delta = _delta(base, 9, instructions)
    delta_entry = _delta_header(
        6, len(delta), _ofs_distance(len(base_entry))
    ) + zlib.compress(delta)
    result = _read_all(module.unpack_objects(io.BytesIO(_pack_of([base_entry, delta_entry]))))
    assert result == [(3, base), (3, b"base+tail")]


def test_ref_delta_resolves_by_sha_and_supports_chains():
    module = _load("git_pack_objects")
    base = b"chain base"
    first = _delta(base, 4, bytes([4]) + b"one!")
    second = _delta(b"one!", 4, bytes([4]) + b"two!")
    entries = [
        _object_entry(3, base),
        _delta_header(7, len(first), _blob_sha(base)) + zlib.compress(first),
        _delta_header(7, len(second), _blob_sha(b"one!")) + zlib.compress(second),
    ]
    result = _read_all(module.unpack_objects(io.BytesIO(_pack_of(entries))))
    assert [payload for _code, payload in result] == [base, b"one!", b"two!"]


def test_thin_pack_ref_delta_with_external_base_is_rejected():
    module = _load("git_pack_objects")
    delta = _delta(b"absent", 1, bytes([1]) + b"x")
    entry = _delta_header(7, len(delta), b"\xaa" * 20) + zlib.compress(delta)
    with pytest.raises(module.PolicyError, match="push full packs"):
        module.unpack_objects(io.BytesIO(_pack_of([entry])))


def test_ofs_delta_bad_offsets_fail_closed():
    module = _load("git_pack_objects")
    base_entry = _object_entry(3, b"base")
    delta = _delta(b"base", 1, bytes([1]) + b"x")
    # Distance beyond the pack start.
    early = _delta_header(6, len(delta), _ofs_distance(999)) + zlib.compress(delta)
    with pytest.raises(module.PolicyError, match="base offset is invalid"):
        module.unpack_objects(io.BytesIO(_pack_of([base_entry, early])))
    # Distance landing between object boundaries.
    misaligned = _delta_header(6, len(delta), _ofs_distance(3)) + zlib.compress(delta)
    with pytest.raises(module.PolicyError, match="base offset is invalid"):
        module.unpack_objects(io.BytesIO(_pack_of([base_entry, misaligned])))


@pytest.mark.parametrize(
    ("mutate", "match"),
    [
        (lambda pack: b"JUNK" + pack[4:], "header is invalid"),
        (lambda pack: pack[:4] + (9).to_bytes(4, "big") + pack[8:], "header is invalid"),
        (lambda pack: pack[:8] + (2).to_bytes(4, "big") + pack[12:], "Git"),
        (lambda pack: pack[:-1], "ended early"),
        (lambda pack: pack[:-20] + b"\x00" * 20, "checksum is invalid"),
        (lambda pack: pack[:11], "ended early"),
    ],
)
def test_malformed_pack_containers_fail_closed(mutate, match):
    module = _load("git_pack_objects")
    pack = _pack_of([_object_entry(3, b"payload")])
    with pytest.raises(module.PolicyError, match=match):
        module.unpack_objects(io.BytesIO(mutate(pack)))


def test_object_count_and_type_bounds_fail_closed(monkeypatch):
    module = _load("git_pack_objects")
    body = b"PACK" + (2).to_bytes(4, "big") + (module.MAX_PACK_OBJECTS + 1).to_bytes(
        4, "big"
    )
    with pytest.raises(module.PolicyError, match="too many objects"):
        module.unpack_objects(io.BytesIO(body + hashlib.sha1(body).digest()))
    reserved = _delta_header(5, 1) + zlib.compress(b"x")
    with pytest.raises(module.PolicyError, match="object type is invalid"):
        module.unpack_objects(io.BytesIO(_pack_of([reserved])))


def test_declared_size_mismatches_and_zlib_bombs_fail_closed(monkeypatch):
    module = _load("git_pack_objects")
    longer = _delta_header(3, 2) + zlib.compress(b"longer-than-two")
    with pytest.raises(module.PolicyError, match="does not match its header"):
        module.unpack_objects(io.BytesIO(_pack_of([longer])))
    shorter = _delta_header(3, 9) + zlib.compress(b"x")
    with pytest.raises(module.PolicyError, match="does not match its header"):
        module.unpack_objects(io.BytesIO(_pack_of([shorter])))
    corrupt = _delta_header(3, 4) + b"\x00not-zlib\x00"
    with pytest.raises(module.PolicyError, match="corrupt"):
        module.unpack_objects(io.BytesIO(_pack_of([corrupt])))

    monkeypatch.setattr(module, "MAX_OBJECT_BYTES", 4)
    with pytest.raises(module.PolicyError, match="exceeds the safe size limit"):
        module.unpack_objects(io.BytesIO(_pack_of([_object_entry(3, b"12345")])))
    monkeypatch.setattr(module, "MAX_OBJECT_BYTES", 1024)
    monkeypatch.setattr(module, "MAX_TOTAL_BYTES", 8)
    with pytest.raises(module.PolicyError, match="contents exceed"):
        module.unpack_objects(
            io.BytesIO(_pack_of([_object_entry(3, b"12345"), _object_entry(3, b"12345")]))
        )


def test_object_and_distance_varints_are_bounded():
    module = _load("git_pack_objects")
    with pytest.raises(module.PolicyError, match="object header is invalid"):
        module._object_header(io.BytesIO(b"\xff" * 12))
    with pytest.raises(module.PolicyError, match="ended early"):
        module._object_header(io.BytesIO(b"\x80"))
    with pytest.raises(module.PolicyError, match="distance is invalid"):
        module._delta_distance(io.BytesIO(b"\xff" * 12))


@pytest.mark.parametrize(
    ("base", "delta", "match"),
    [
        (b"base", b"", "delta header is invalid"),
        (b"base", _size_varint(4), "delta header is invalid"),
        (b"base", _size_varint(9) + _size_varint(1) + bytes([1]) + b"x", "sizes are invalid"),
        (b"base", _size_varint(4) + _size_varint(2) + bytes([0]), "instruction is invalid"),
        (b"base", _size_varint(4) + _size_varint(2) + bytes([5]) + b"ab", "insert is truncated"),
        (b"base", _size_varint(4) + _size_varint(2) + bytes([0x80 | 0x10]), "copy is truncated"),
        (
            b"base",
            _size_varint(4) + _size_varint(2) + bytes([0x80 | 0x01 | 0x10, 3, 9]),
            "copy is out of bounds",
        ),
        (b"base", _size_varint(4) + _size_varint(1) + bytes([2]) + b"ab", "exceeds its declared size"),
        (b"base", _size_varint(4) + _size_varint(3) + bytes([2]) + b"ab", "result size is invalid"),
    ],
)
def test_delta_application_rejects_malformed_instructions(base, delta, match):
    module = _load("git_pack_objects")
    with pytest.raises(module.PolicyError, match=match):
        module._apply_delta(base, delta)


def test_delta_copy_with_zero_size_field_means_the_git_default():
    module = _load("git_pack_objects")
    base = b"z" * 0x10000
    delta = _delta(base, 0x10000, bytes([0x80 | 0x10 | 0x20, 0, 0]))
    assert module._apply_delta(base, delta) == base


def test_delta_target_size_cap_applies(monkeypatch):
    module = _load("git_pack_objects")
    monkeypatch.setattr(module, "MAX_OBJECT_BYTES", 4)
    delta = _delta(b"base", 5, bytes([5]) + b"abcde")
    with pytest.raises(module.PolicyError, match="sizes are invalid"):
        module._apply_delta(b"base", delta)


def test_resolution_pass_and_total_caps_apply(monkeypatch):
    module = _load("git_pack_objects")
    base = b"pass base"
    first = _delta(base, 4, bytes([4]) + b"one!")
    second = _delta(b"one!", 4, bytes([4]) + b"two!")
    entries = [
        _delta_header(7, len(second), _blob_sha(b"one!")) + zlib.compress(second),
        _object_entry(3, base),
        _delta_header(7, len(first), _blob_sha(base)) + zlib.compress(first),
    ]
    pack = _pack_of(entries)
    monkeypatch.setattr(module, "MAX_RESOLVE_PASSES", 1)
    with pytest.raises(module.PolicyError, match="resolution limit"):
        module.unpack_objects(io.BytesIO(pack))
    monkeypatch.setattr(module, "MAX_RESOLVE_PASSES", 64)
    result = _read_all(module.unpack_objects(io.BytesIO(pack)))
    assert [payload for _code, payload in result] == [b"two!", base, b"one!"]

    monkeypatch.setattr(module, "MAX_TOTAL_BYTES", len(base) + len(first) + len(second) + 1)
    with pytest.raises(module.PolicyError, match="contents exceed"):
        module.unpack_objects(io.BytesIO(pack))


def test_parse_loose_object_round_trip_and_bounds(monkeypatch):
    module = _load("git_pack_objects")
    payload = b"loose blob"
    value = zlib.compress(b"blob %d\x00" % len(payload) + payload)
    assert module.parse_loose_object(value) == (3, payload)

    with pytest.raises(module.PolicyError, match="corrupt"):
        module.parse_loose_object(b"\x00junk")
    with pytest.raises(module.PolicyError, match="framing is invalid"):
        module.parse_loose_object(value + b"trailing")
    with pytest.raises(module.PolicyError, match="framing is invalid"):
        module.parse_loose_object(value[:-4])
    for header in (
        b"noodle 3\x00abc",
        b"blob3\x00abc",
        b"blob x\x00abc",
        b"blob 4\x00abc",
        b"blob 3",
    ):
        with pytest.raises(module.PolicyError, match="header is invalid"):
            module.parse_loose_object(zlib.compress(header))
    monkeypatch.setattr(module, "MAX_OBJECT_BYTES", 4)
    with pytest.raises(module.PolicyError, match="exceeds the safe size limit"):
        module.parse_loose_object(b"12345")