Skip to content
Open
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
3 changes: 2 additions & 1 deletion test/test_datasource.py
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down
127 changes: 65 additions & 62 deletions test/test_fileuploads.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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())
Loading