from __future__ import annotations

import hashlib
import zlib
from pathlib import Path

import pytest

from osm_lead_source_service.errors import (
    ContractViolation,
    PbfResourceLimitExceeded,
)
from osm_lead_source_service.pbf import (
    OsmiumPbfParser,
    ParsedOsmObject,
    PbfParseLimits,
    PbfParseSummary,
    VerifiedPbfInput,
)


class _Sink:
    def accept(self, parsed_object: ParsedOsmObject) -> None:
        del parsed_object

    def finish(self, summary: PbfParseSummary) -> None:
        del summary

    def abort(self) -> None:
        return None


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


def _bytes_field(field_number: int, value: bytes) -> bytes:
    return _varint((field_number << 3) | 2) + _varint(len(value)) + value


def _int_field(field_number: int, value: int) -> bytes:
    return _varint(field_number << 3) + _varint(value)


def _block(blob_type: str, blob: bytes) -> bytes:
    header = _bytes_field(1, blob_type.encode("ascii")) + _int_field(3, len(blob))
    return len(header).to_bytes(4, "big") + header + blob


def _compressed_blob(payload: bytes) -> bytes:
    return _int_field(2, len(payload)) + _bytes_field(3, zlib.compress(payload))


def _approved(path: Path) -> VerifiedPbfInput:
    data = path.read_bytes()
    return VerifiedPbfInput(
        local_path=path,
        content_sha256=hashlib.sha256(data).hexdigest(),
        size_bytes=len(data),
        provenance="synthetic cumulative envelope test",
    )


def test_total_uncompressed_limit_has_hard_ceiling() -> None:
    with pytest.raises(ContractViolation, match="max_total_uncompressed_bytes"):
        PbfParseLimits(max_total_uncompressed_bytes=(16 * 1024 * 1024 * 1024) + 1)


def test_blob_count_limit_has_hard_ceiling() -> None:
    with pytest.raises(ContractViolation, match="max_blob_count"):
        PbfParseLimits(max_blob_count=500_001)


def test_per_blob_limit_cannot_exceed_total_limit() -> None:
    with pytest.raises(ContractViolation, match="cannot exceed"):
        PbfParseLimits(
            max_uncompressed_blob_bytes=2048,
            max_total_uncompressed_bytes=1024,
        )


def test_cumulative_uncompressed_bytes_are_bounded_across_blobs(tmp_path: Path) -> None:
    header_blob = _bytes_field(1, b"x")
    data_blob = _compressed_blob(b"x" * 700)
    path = tmp_path / "cumulative.osm.pbf"
    path.write_bytes(
        _block("OSMHeader", header_blob)
        + _block("OSMData", data_blob)
        + _block("OSMData", data_blob)
    )
    parser = OsmiumPbfParser(
        PbfParseLimits(
            max_uncompressed_blob_bytes=1024,
            max_total_uncompressed_bytes=1200,
        )
    )
    with pytest.raises(PbfResourceLimitExceeded, match="max_total_uncompressed_bytes"):
        parser.parse(_approved(path), _Sink())


def test_blob_count_is_bounded_across_the_entire_file(tmp_path: Path) -> None:
    header_blob = _bytes_field(1, b"x")
    data_blob = _compressed_blob(b"x")
    path = tmp_path / "many-blocks.osm.pbf"
    path.write_bytes(
        _block("OSMHeader", header_blob)
        + _block("OSMData", data_blob)
        + _block("OSMData", data_blob)
    )
    parser = OsmiumPbfParser(PbfParseLimits(max_blob_count=2))
    with pytest.raises(PbfResourceLimitExceeded, match="max_blob_count"):
        parser.parse(_approved(path), _Sink())