thousandeyes-sdk-python/thousandeyes-sdk-core/test/test_pagination_iterator.py

from types import SimpleNamespace

from thousandeyes_sdk.core.pagination_iterator import PaginatorIterator


def test_iterator_uses_cursor_from_next_href():
    calls = []

    def method(**params):
        calls.append(params.copy())
        if params.get("cursor") is None:
            links = SimpleNamespace(next="https://example.com/items?cursor=abc")
            items = ["first", "second"]
        else:
            links = SimpleNamespace(next=None)
            items = ["third"]
        data = SimpleNamespace(links=links)
        return SimpleNamespace(data=data, items=items)

    responses = list(PaginatorIterator(method, lambda response: response.items))

    assert responses == ["first", "second", "third"]
    assert calls == [{}, {"cursor": "abc"}]


def test_iterator_reads_cursor_from_links_mapping():
    calls = []

    def method(**params):
        calls.append(params.copy())
        if params.get("pageCursor") is None:
            data = {"_links": {"next": {"href": "https://example.com?foo=1&pageCursor=xyz"}}}
            items = ["alpha"]
        else:
            data = {"_links": {"next": None}}
            items = ["beta"]
        return SimpleNamespace(data=data, items=items)

    responses = list(PaginatorIterator(method, lambda response: response.items, cursor_param="pageCursor"))

    assert responses == ["alpha", "beta"]
    assert calls == [{}, {"pageCursor": "xyz"}]


def test_iterator_stops_when_no_cursor_param_present():
    calls = []

    def method(**params):
        calls.append(params.copy())
        if params.get("cursor") is None:
            data = {"links": {"next": "/next/page"}}
            items = ["one"]
        else:
            data = {"links": {"next": None}}
            items = ["two"]
        return SimpleNamespace(data=data, items=items)

    responses = list(PaginatorIterator(method, lambda response: response.items))

    assert responses == ["one"]
    assert calls == [{}]


def test_iterator_stops_on_repeated_cursor():
    calls = []

    def method(**params):
        calls.append(params.copy())
        data = {"links": {"next": "https://example.com?cursor=same"}}
        return SimpleNamespace(data=data, items=["only"])

    responses = list(PaginatorIterator(method, lambda response: response.items, cursor="same"))

    assert responses == ["only"]
    assert calls == [{"cursor": "same"}]