"""Opt-in live MinIO compatibility tests for the S3 backend."""

from __future__ import annotations

from collections.abc import AsyncIterator
from os import environ
from uuid import uuid4

import pytest

from artifactstore.identity import ContentAddress, digest_bytes
from artifactstore.storage import ObjectNotFoundError, S3Backend, S3BackendConfig

pytestmark = pytest.mark.integration

_MINIO_ENV_VARS = (
    "ARTIFACTSTORE_MINIO_ENDPOINT_URL",
    "ARTIFACTSTORE_MINIO_ACCESS_KEY",
    "ARTIFACTSTORE_MINIO_SECRET_KEY",
    "ARTIFACTSTORE_MINIO_BUCKET",
)
_MIB = 1024 * 1024


async def _stream(data: bytes, chunk_size: int = _MIB) -> AsyncIterator[bytes]:
    for offset in range(0, len(data), chunk_size):
        yield data[offset : offset + chunk_size]


async def _consume(stream: AsyncIterator[bytes]) -> bytes:
    out = bytearray()
    async for chunk in stream:
        out.extend(chunk)
    return bytes(out)


def _ca(data: bytes) -> ContentAddress:
    return digest_bytes(data).primary.content_address


@pytest.fixture
def minio_backend() -> S3Backend:
    missing = [name for name in _MINIO_ENV_VARS if not environ.get(name)]
    if missing:
        pytest.skip(f"set {', '.join(missing)} to run live MinIO compatibility tests")

    config = S3BackendConfig(
        endpoint_url=environ["ARTIFACTSTORE_MINIO_ENDPOINT_URL"],
        region=environ.get("ARTIFACTSTORE_MINIO_REGION", "us-east-1"),
        bucket=environ["ARTIFACTSTORE_MINIO_BUCKET"],
        key_prefix=environ.get("ARTIFACTSTORE_MINIO_KEY_PREFIX", f"compat/{uuid4()}"),
        access_key_id=environ["ARTIFACTSTORE_MINIO_ACCESS_KEY"],
        secret_access_key=environ["ARTIFACTSTORE_MINIO_SECRET_KEY"],
        storage_class=environ.get("ARTIFACTSTORE_MINIO_STORAGE_CLASS") or None,
        sse=environ.get("ARTIFACTSTORE_MINIO_SSE") or None,
        multipart_threshold_bytes=5 * _MIB,
        multipart_chunk_bytes=5 * _MIB,
    )
    return S3Backend(config, backend_id="minio-live", chunk_size=512 * 1024)


async def test_live_minio_round_trip_with_range(minio_backend: S3Backend) -> None:
    data = f"artifact-store minio compatibility {uuid4()}".encode()
    content_address = _ca(data)
    try:
        status = await minio_backend.health()
        assert status.healthy, status.detail

        receipt = await minio_backend.put(
            content_address,
            _stream(data, chunk_size=7),
            size_hint=len(data),
        )
        assert receipt.backend_id == "minio-live"
        assert receipt.size_bytes == len(data)

        metadata = await minio_backend.head(content_address)
        assert metadata.size_bytes == len(data)

        stream = await minio_backend.get(content_address)
        assert await _consume(stream) == data

        ranged = await minio_backend.get(content_address, byte_range=(1, 9))
        assert await _consume(ranged) == data[1:10]
    finally:
        await minio_backend.delete(content_address)

    with pytest.raises(ObjectNotFoundError):
        await minio_backend.head(content_address)


async def test_live_minio_multipart_upload(minio_backend: S3Backend) -> None:
    data = (b"artifact-store-multipart-" + uuid4().bytes) * 200_000
    content_address = _ca(data)
    try:
        receipt = await minio_backend.put(content_address, _stream(data), size_hint=len(data))
        assert receipt.backend_id == "minio-live"
        assert receipt.size_bytes == len(data)

        tail = await minio_backend.get(content_address, byte_range=(len(data) - 16, len(data) - 1))
        assert await _consume(tail) == data[-16:]
    finally:
        await minio_backend.delete(content_address)