Skip to content
Merged
Show file tree
Hide file tree
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
14 changes: 6 additions & 8 deletions test/test_dqw.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,9 @@
import unittest
import tableauserverclient as TSC


class DQWTests(unittest.TestCase):
def test_existence(self):
dqw: TSC.DQWItem = TSC.DQWItem()
dqw.message = "message"
dqw.warning_type = TSC.DQWItem.WarningType.STALE
dqw.active = True
dqw.severe = True
def test_dqw_existence():
dqw: TSC.DQWItem = TSC.DQWItem()
dqw.message = "message"
dqw.warning_type = TSC.DQWItem.WarningType.STALE
dqw.active = True
dqw.severe = True
146 changes: 78 additions & 68 deletions test/test_endpoint.py
Original file line number Diff line number Diff line change
@@ -1,83 +1,93 @@
from pathlib import Path
import pytest
import requests
import unittest

import tableauserverclient as TSC
from tableauserverclient.server.endpoint import Endpoint

import requests_mock

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


class TestEndpoint(unittest.TestCase):
def setUp(self) -> None:
self.server = TSC.Server("http://test/", use_server_version=False)
@pytest.fixture(scope="function")
def server():
"""Fixture to create a TSC.Server instance for testing."""
server = TSC.Server("http://test", False)

# Fake signin
self.server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
self.server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM"
return super().setUp()
# Fake signin
server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
server._auth_token = "j80k54ll2lfMZ0tv97mlPvvS"

def test_fallback_request_logic(self) -> None:
url = "http://test/"
endpoint = TSC.server.Endpoint(self.server)
with requests_mock.mock() as m:
m.get(url)
response = endpoint.get_request(url=url)
self.assertIsNotNone(response)
return server

def test_user_friendly_request_returns(self) -> None:
url = "http://test/"
endpoint = TSC.server.Endpoint(self.server)
with requests_mock.mock() as m:
m.get(url)
response = endpoint.send_request_while_show_progress_threaded(
endpoint.parent_srv.session.get, url=url, request_timeout=2
)
self.assertIsNotNone(response)

def test_blocking_request_raises_request_error(self) -> None:
with pytest.raises(requests.exceptions.ConnectionError):
url = "http://test/"
endpoint = TSC.server.Endpoint(self.server)
response = endpoint._blocking_request(endpoint.parent_srv.session.get, url=url)
self.assertIsNotNone(response)

def test_get_request_stream(self) -> None:

def test_fallback_request_logic(server: TSC.Server) -> None:
url = "http://test/"
endpoint = Endpoint(server)
with requests_mock.mock() as m:
m.get(url)
response = endpoint.get_request(url=url)
assert response is not None


def test_user_friendly_request_returns(server: TSC.Server) -> None:
url = "http://test/"
endpoint = Endpoint(server)
with requests_mock.mock() as m:
m.get(url)
response = endpoint.send_request_while_show_progress_threaded(
endpoint.parent_srv.session.get, url=url, request_timeout=2
)
assert response is not None


def test_blocking_request_raises_request_error(server: TSC.Server) -> None:
with pytest.raises(requests.exceptions.ConnectionError):
url = "http://test/"
endpoint = TSC.server.Endpoint(self.server)
with requests_mock.mock() as m:
m.get(url, headers={"Content-Type": "application/octet-stream"})
response = endpoint.get_request(url, parameters={"stream": True})

self.assertFalse(response._content_consumed)

def test_binary_log_truncated(self):
class FakeResponse:
headers = {"Content-Type": "application/octet-stream"}
content = b"\x1337" * 1000
status_code = 200

endpoint = TSC.server.Endpoint(self.server)
server_response = FakeResponse()
log = endpoint.log_response_safely(server_response)
self.assertTrue(log.find("[Truncated File Contents]") > 0, log)

def test_set_user_agent_from_options_headers(self):
params = {"User-Agent": "1", "headers": {"User-Agent": "2"}}
result = TSC.server.Endpoint.set_user_agent(params)
# it should use the value under 'headers' if more than one is given
print(result)
print(result["headers"]["User-Agent"])
self.assertTrue(result["headers"]["User-Agent"] == "2")

def test_set_user_agent_from_options(self):
params = {"headers": {"User-Agent": "2"}}
result = TSC.server.Endpoint.set_user_agent(params)
self.assertTrue(result["headers"]["User-Agent"] == "2")

def test_set_user_agent_when_blank(self):
params = {"headers": {}}
result = TSC.server.Endpoint.set_user_agent(params)
self.assertTrue(result["headers"]["User-Agent"].startswith("Tableau Server Client"))
endpoint = Endpoint(server)
response = endpoint._blocking_request(endpoint.parent_srv.session.get, url=url)
assert response is not None


def test_get_request_stream(server: TSC.Server) -> None:
url = "http://test/"
endpoint = Endpoint(server)
with requests_mock.mock() as m:
m.get(url, headers={"Content-Type": "application/octet-stream"})
response = endpoint.get_request(url, parameters={"stream": True})

assert response._content_consumed is False


def test_binary_log_truncated(server: TSC.Server) -> None:
class FakeResponse:
headers = {"Content-Type": "application/octet-stream"}
content = b"\x1337" * 1000
status_code = 200

endpoint = Endpoint(server)
server_response = FakeResponse()
log = endpoint.log_response_safely(server_response) # type: ignore
assert log.find("[Truncated File Contents]") > 0


def test_set_user_agent_from_options_headers(server: TSC.Server) -> None:
params = {"User-Agent": "1", "headers": {"User-Agent": "2"}}
result = Endpoint.set_user_agent(params)
# it should use the value under 'headers' if more than one is given
print(result)
print(result["headers"]["User-Agent"])
assert result["headers"]["User-Agent"] == "2"


def test_set_user_agent_from_options(server: TSC.Server) -> None:
params = {"headers": {"User-Agent": "2"}}
result = Endpoint.set_user_agent(params)
assert result["headers"]["User-Agent"] == "2"


def test_set_user_agent_when_blank(server: TSC.Server) -> None:
params = {"headers": {}} # type: ignore
result = Endpoint.set_user_agent(params)
assert result["headers"]["User-Agent"].startswith("Tableau Server Client")
Loading