diff --git a/test/test_datasource.py b/test/test_datasource.py index 7f4cca75..56eb11ab 100644 --- a/test/test_datasource.py +++ b/test/test_datasource.py @@ -895,7 +895,8 @@ def test_publish_description(server: TSC.Server) -> None: ds_elem = body.find(".//datasource") assert ds_elem is not None assert ds_elem.attrib["description"] == "Sample description" - + + def test_get_datasource_no_owner(server: TSC.Server) -> None: with requests_mock.mock() as m: m.get(server.datasources.baseurl, text=GET_NO_OWNER.read_text()) diff --git a/test/test_fileuploads.py b/test/test_fileuploads.py index 9567bc3a..2e69b588 100644 --- a/test/test_fileuploads.py +++ b/test/test_fileuploads.py @@ -1,17 +1,18 @@ import contextlib import io import os -import unittest +from pathlib import Path +import pytest import requests_mock +import tableauserverclient as TSC from tableauserverclient.config import BYTES_PER_MB, config -from tableauserverclient.server import Server -from ._utils import asset -TEST_ASSET_DIR = os.path.join(os.path.dirname(__file__), "assets") -FILEUPLOAD_INITIALIZE = os.path.join(TEST_ASSET_DIR, "fileupload_initialize.xml") -FILEUPLOAD_APPEND = os.path.join(TEST_ASSET_DIR, "fileupload_append.xml") +TEST_ASSET_DIR = Path(__file__).parent / "assets" +FILEUPLOAD_INITIALIZE = TEST_ASSET_DIR / "fileupload_initialize.xml" +FILEUPLOAD_APPEND = TEST_ASSET_DIR / "fileupload_append.xml" +SAMPLE_WB = TEST_ASSET_DIR / "SampleWB.twbx" @contextlib.contextmanager @@ -25,65 +26,67 @@ def set_env(**environ): os.environ.update(old_environ) -class FileuploadsTests(unittest.TestCase): - def setUp(self): - self.server = Server("http://test", False) +@pytest.fixture(scope="function") +def server(): + """Fixture to create a TSC.Server instance for testing.""" + server = TSC.Server("http://test", False) - # Fake sign in - self.server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67" - self.server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM" + # Fake signin + server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67" + server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM" - self.baseurl = f"{self.server.baseurl}/sites/{self.server.site_id}/fileUploads" + return server - def test_read_chunks_file_path(self): - file_path = asset("SampleWB.twbx") - chunks = self.server.fileuploads._read_chunks(file_path) + +def test_read_chunks_file_path(server: TSC.Server) -> None: + file_path = str(SAMPLE_WB) + chunks = server.fileuploads._read_chunks(file_path) + for chunk in chunks: + assert chunk is not None + + +def test_read_chunks_file_object(server: TSC.Server) -> None: + with SAMPLE_WB.open("rb") as f: + chunks = server.fileuploads._read_chunks(f) for chunk in chunks: - self.assertIsNotNone(chunk) - - def test_read_chunks_file_object(self): - with open(asset("SampleWB.twbx"), "rb") as f: - chunks = self.server.fileuploads._read_chunks(f) - for chunk in chunks: - self.assertIsNotNone(chunk) - - def test_upload_chunks_file_path(self): - file_path = asset("SampleWB.twbx") - upload_id = "7720:170fe6b1c1c7422dadff20f944d58a52-1:0" - - with open(FILEUPLOAD_INITIALIZE, "rb") as f: - initialize_response_xml = f.read().decode("utf-8") - with open(FILEUPLOAD_APPEND, "rb") as f: - append_response_xml = f.read().decode("utf-8") + assert chunk is not None + + +def test_upload_chunks_file_path(server: TSC.Server) -> None: + file_path = str(SAMPLE_WB) + upload_id = "7720:170fe6b1c1c7422dadff20f944d58a52-1:0" + + initialize_response_xml = FILEUPLOAD_INITIALIZE.read_text() + append_response_xml = FILEUPLOAD_APPEND.read_text() + with requests_mock.mock() as m: + m.post(server.fileuploads.baseurl, text=initialize_response_xml) + m.put(f"{server.fileuploads.baseurl}/{upload_id}", text=append_response_xml) + actual = server.fileuploads.upload(file_path) + + assert upload_id == actual + + +def test_upload_chunks_file_object(server: TSC.Server) -> None: + upload_id = "7720:170fe6b1c1c7422dadff20f944d58a52-1:0" + + with SAMPLE_WB.open("rb") as file_content: + initialize_response_xml = FILEUPLOAD_INITIALIZE.read_text() + append_response_xml = FILEUPLOAD_APPEND.read_text() with requests_mock.mock() as m: - m.post(self.baseurl, text=initialize_response_xml) - m.put(f"{self.baseurl}/{upload_id}", text=append_response_xml) - actual = self.server.fileuploads.upload(file_path) - - self.assertEqual(upload_id, actual) - - def test_upload_chunks_file_object(self): - upload_id = "7720:170fe6b1c1c7422dadff20f944d58a52-1:0" - - with open(asset("SampleWB.twbx"), "rb") as file_content: - with open(FILEUPLOAD_INITIALIZE, "rb") as f: - initialize_response_xml = f.read().decode("utf-8") - with open(FILEUPLOAD_APPEND, "rb") as f: - append_response_xml = f.read().decode("utf-8") - with requests_mock.mock() as m: - m.post(self.baseurl, text=initialize_response_xml) - m.put(f"{self.baseurl}/{upload_id}", text=append_response_xml) - actual = self.server.fileuploads.upload(file_content) - - self.assertEqual(upload_id, actual) - - def test_upload_chunks_config(self): - data = io.BytesIO() - data.write(b"1" * (config.CHUNK_SIZE_MB * BYTES_PER_MB + 1)) + m.post(server.fileuploads.baseurl, text=initialize_response_xml) + m.put(f"{server.fileuploads.baseurl}/{upload_id}", text=append_response_xml) + actual = server.fileuploads.upload(file_content) + + assert upload_id == actual + + +def test_upload_chunks_config(server: TSC.Server) -> None: + data = io.BytesIO() + data.write(b"1" * (config.CHUNK_SIZE_MB * BYTES_PER_MB + 1)) + data.seek(0) + with set_env(TSC_CHUNK_SIZE_MB="1"): + chunker = server.fileuploads._read_chunks(data) + chunk = next(chunker) + assert len(chunk) == config.CHUNK_SIZE_MB * BYTES_PER_MB data.seek(0) - with set_env(TSC_CHUNK_SIZE_MB="1"): - chunker = self.server.fileuploads._read_chunks(data) - chunk = next(chunker) - assert len(chunk) == config.CHUNK_SIZE_MB * BYTES_PER_MB - data.seek(0) - assert len(chunk) < len(data.read()) + assert len(chunk) < len(data.read())