comet-ml--opik/sdks/opik_optimizer/tests/unit/utils/test_dataset_fetching.py

from __future__ import annotations

from collections.abc import Callable
from typing import Any

import pytest

import httpx  # type: ignore[import-not-found]

from opik_optimizer.utils import dataset as dataset_utils


class DummyDataset:
    def __init__(self, records: list[dict[str, int]]) -> None:
        self._records = list(records)

    def __len__(self) -> int:
        return len(self._records)

    def shuffle(self, *, seed: int) -> DummyDataset:
        return self

    def filter(self, function: Callable[[dict[str, int]], bool]) -> DummyDataset:
        self._records = [record for record in self._records if function(record)]
        return self

    def select(self, indices: range) -> DummyDataset:
        return DummyDataset([self._records[i] for i in indices])

    def to_list(self) -> list[dict[str, int]]:
        return list(self._records)


def test_stream_records_for_slice_filters_and_slices() -> None:
    records = [{"value": i} for i in range(6)]

    def load_fn(*, streaming: bool, **kwargs: object) -> list[dict[str, int]]:
        del kwargs
        assert streaming
        return records

    result = dataset_utils.stream_records_for_slice(
        load_fn=load_fn,
        load_kwargs={},
        start=1,
        count=2,
        filter_by={"value": lambda value: value >= 2},
    )

    assert result == [{"value": 3}, {"value": 4}]


def test_download_and_slice_hf_dataset_raises_when_slice_exceeds_total() -> None:
    def load_fn(**kwargs: object) -> DummyDataset:
        del kwargs
        return DummyDataset([{"value": 0}, {"value": 1}, {"value": 2}])

    with pytest.raises(ValueError):
        dataset_utils.download_and_slice_hf_dataset(
            load_fn=load_fn,
            load_kwargs={},
            start=2,
            count=5,
            seed=42,
        )


def test_download_and_slice_hf_dataset_respects_filter_by(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    def load_fn(**kwargs: object) -> DummyDataset:
        del kwargs
        return DummyDataset([{"value": 0}, {"value": 1}, {"value": 2}, {"value": 3}])

    result = dataset_utils.download_and_slice_hf_dataset(
        load_fn=load_fn,
        load_kwargs={},
        start=0,
        count=10,
        seed=42,
        filter_by={"value": lambda value: value >= 2},
    )

    assert result == [{"value": 2}, {"value": 3}]


def test_fetch_records_for_slice_falls_back_to_download(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    slice_request = dataset_utils.SliceRequest(
        source_split="train",
        start=0,
        count=2,
        dataset_name="test",
    )

    monkeypatch.setattr(
        dataset_utils,
        "stream_records_for_slice",
        lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("stream disabled")),
    )

    download_called: list[dict[str, object]] = []

    def fake_download(
        *,
        load_fn: Callable[..., object],
        load_kwargs: dict[str, object],
        start: int,
        count: int | None,
        seed: int,
        filter_by: dict[str, object] | None = None,
    ) -> list[dict[str, object]]:
        download_called.append(load_kwargs)
        return [{"downloaded": True}]

    monkeypatch.setattr(dataset_utils, "download_and_slice_hf_dataset", fake_download)

    records = dataset_utils.fetch_records_for_slice(
        slice_request=slice_request,
        load_kwargs_resolver=lambda split: {"split": split},
        seed=42,
        filter_by={"value": 1},
        custom_loader=None,
        load_fn=lambda **kwargs: (),
    )

    assert records == [{"downloaded": True}]
    assert download_called


def test_load_hf_dataset_slice_handles_stream_errors(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    def fake_fetch(*_args: Any, **_kwargs: Any) -> list[dict[str, Any]]:
        raise httpx.RemoteProtocolError("boom")

    monkeypatch.setattr(dataset_utils, "fetch_records_for_slice", fake_fetch)

    with pytest.raises(RuntimeError, match="Dataset download interrupted"):
        dataset_utils.load_hf_dataset_slice(
            base_name="test",
            requested_split="train",
            presets={},
            default_source_split="train",
            load_kwargs_resolver=lambda split: {"path": "dummy", "split": split},
            start=0,
            count=1,
        )