Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
228 changes: 115 additions & 113 deletions test/test_pager.py
Original file line number Diff line number Diff line change
@@ -1,19 +1,32 @@
import contextlib
import os
import unittest
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 = os.path.join(os.path.dirname(__file__), "assets")
TEST_ASSET_DIR = Path(__file__).parent / "assets"

GET_VIEW_XML = os.path.join(TEST_ASSET_DIR, "view_get.xml")
GET_XML_PAGE1 = os.path.join(TEST_ASSET_DIR, "workbook_get_page_1.xml")
GET_XML_PAGE2 = os.path.join(TEST_ASSET_DIR, "workbook_get_page_2.xml")
GET_XML_PAGE3 = os.path.join(TEST_ASSET_DIR, "workbook_get_page_3.xml")
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
Expand All @@ -27,110 +40,99 @@ def set_env(**environ):
os.environ.update(old_environ)


class PagerTests(unittest.TestCase):
def setUp(self):
self.server = TSC.Server("http://test", False)

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

self.baseurl = self.server.workbooks.baseurl

def test_pager_with_no_options(self) -> None:
with open(GET_XML_PAGE1, "rb") as f:
page_1 = f.read().decode("utf-8")
with open(GET_XML_PAGE2, "rb") as f:
page_2 = f.read().decode("utf-8")
with open(GET_XML_PAGE3, "rb") as f:
page_3 = f.read().decode("utf-8")
with requests_mock.mock() as m:
# Register Pager with default request options
m.get(self.baseurl, text=page_1)

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

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

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

def test_pager_with_options(self) -> None:
with open(GET_XML_PAGE1, "rb") as f:
page_1 = f.read().decode("utf-8")
with open(GET_XML_PAGE2, "rb") as f:
page_2 = f.read().decode("utf-8")
with open(GET_XML_PAGE3, "rb") as f:
page_3 = f.read().decode("utf-8")
with requests_mock.mock() as m:
# Register Pager with some pages
m.get(self.baseurl + "?pageNumber=1&pageSize=1", complete_qs=True, text=page_1)
m.get(self.baseurl + "?pageNumber=2&pageSize=1", complete_qs=True, text=page_2)
m.get(self.baseurl + "?pageNumber=3&pageSize=1", complete_qs=True, text=page_3)
m.get(self.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(self.server.workbooks, opts))
self.assertTrue(len(workbooks) == 2)

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

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

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

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

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

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

def test_queryset_no_matches(self) -> 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(self.server.groups.baseurl, text=xml)
all_groups = self.server.groups.all()
groups = list(all_groups)
assert len(groups) == 0
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
Loading