test_pager.py

import contextlib
import os
from pathlib import Path
import xml.etree.ElementTree as ET

import pytest
import requests_mock

import tableauserverclient as TSC
from tableauserverclient.config import config

TEST_ASSET_DIR = Path(__file__).parent / "assets"

GET_VIEW_XML = TEST_ASSET_DIR / "view_get.xml"
GET_XML_PAGE1 = TEST_ASSET_DIR / "workbook_get_page_1.xml"
GET_XML_PAGE2 = TEST_ASSET_DIR / "workbook_get_page_2.xml"
GET_XML_PAGE3 = TEST_ASSET_DIR / "workbook_get_page_3.xml"


@pytest.fixture(scope="function")
def server():
    """Fixture to create a TSC.Server instance for testing."""
    server = TSC.Server("http://test", False)

    # Fake signin
    server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
    server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM"

    return server


@contextlib.contextmanager
def set_env(**environ):
    old_environ = dict(os.environ)
    os.environ.update(environ)
    try:
        yield
    finally:
        os.environ.clear()
        os.environ.update(old_environ)


def test_pager_with_no_options(server: TSC.Server) -> None:
    page_1 = GET_XML_PAGE1.read_text()
    page_2 = GET_XML_PAGE2.read_text()
    page_3 = GET_XML_PAGE3.read_text()
    with requests_mock.mock() as m:
        # Register Pager with default request options
        m.get(server.workbooks.baseurl, text=page_1)

        # Register Pager with some pages
        m.get(server.workbooks.baseurl + "?pageNumber=1&pageSize=1", text=page_1)
        m.get(server.workbooks.baseurl + "?pageNumber=2&pageSize=1", text=page_2)
        m.get(server.workbooks.baseurl + "?pageNumber=3&pageSize=1", text=page_3)

        # No options should get all 3
        workbooks = list(TSC.Pager(server.workbooks))
        assert len(workbooks) == 3

        # Let's check that workbook items aren't duplicates
        wb1, wb2, wb3 = workbooks
        assert wb1.name == "Page1Workbook"
        assert wb2.name == "Page2Workbook"
        assert wb3.name == "Page3Workbook"


def test_pager_with_options(server: TSC.Server) -> None:
    page_1 = GET_XML_PAGE1.read_text()
    page_2 = GET_XML_PAGE2.read_text()
    page_3 = GET_XML_PAGE3.read_text()
    with requests_mock.mock() as m:
        # Register Pager with some pages
        m.get(server.workbooks.baseurl + "?pageNumber=1&pageSize=1", complete_qs=True, text=page_1)
        m.get(server.workbooks.baseurl + "?pageNumber=2&pageSize=1", complete_qs=True, text=page_2)
        m.get(server.workbooks.baseurl + "?pageNumber=3&pageSize=1", complete_qs=True, text=page_3)
        m.get(server.workbooks.baseurl + "?pageNumber=1&pageSize=3", complete_qs=True, text=page_1)

        # Starting on page 2 should get 2 out of 3
        opts = TSC.RequestOptions(2, 1)
        workbooks = list(TSC.Pager(server.workbooks, opts))
        assert len(workbooks) == 2

        # Check that the workbooks are the 2 we think they should be
        wb2, wb3 = workbooks
        assert wb2.name == "Page2Workbook"
        assert wb3.name == "Page3Workbook"

        # Starting on 1 with pagesize of 3 should get all 3
        opts = TSC.RequestOptions(1, 3)
        workbooks = list(TSC.Pager(server.workbooks, opts))
        assert len(workbooks) == 3
        wb1, wb2, wb3 = workbooks
        assert wb1.name == "Page1Workbook"
        assert wb2.name == "Page2Workbook"
        assert wb3.name == "Page3Workbook"

        # Starting on 3 with pagesize of 1 should get the last item
        opts = TSC.RequestOptions(3, 1)
        workbooks = list(TSC.Pager(server.workbooks, opts))
        assert len(workbooks) == 1
        # Should have the last workbook
        wb3 = workbooks.pop()
        assert wb3.name == "Page3Workbook"


def test_pager_with_env_var(server: TSC.Server) -> None:
    with set_env(TSC_PAGE_SIZE="1000"):
        assert config.PAGE_SIZE == 1000
        loop = TSC.Pager(server.workbooks)
        assert loop._options.pagesize == 1000


def test_queryset_with_env_var(server: TSC.Server) -> None:
    with set_env(TSC_PAGE_SIZE="1000"):
        assert config.PAGE_SIZE == 1000
        loop = server.workbooks.all()
        assert loop.request_options.pagesize == 1000


def test_pager_view(server: TSC.Server) -> None:
    with open(GET_VIEW_XML, "rb") as f:
        view_xml = f.read().decode("utf-8")
    with requests_mock.mock() as m:
        m.get(server.views.baseurl, text=view_xml)
        for view in TSC.Pager(server.views):
            assert view.name is not None


def test_queryset_no_matches(server: TSC.Server) -> None:
    elem = ET.Element("tsResponse", xmlns="http://tableau.com/api")
    ET.SubElement(elem, "pagination", totalAvailable="0")
    ET.SubElement(elem, "groups")
    xml = ET.tostring(elem).decode("utf-8")
    with requests_mock.mock() as m:
        m.get(server.groups.baseurl, text=xml)
        all_groups = server.groups.all()
        groups = list(all_groups)
    assert len(groups) == 0