diff --git a/test/test_dqw.py b/test/test_dqw.py index 6d1219f6..5cb17221 100644 --- a/test/test_dqw.py +++ b/test/test_dqw.py @@ -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 diff --git a/test/test_endpoint.py b/test/test_endpoint.py index ff1ef0f7..0b852ab0 100644 --- a/test/test_endpoint.py +++ b/test/test_endpoint.py @@ -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")