test(e2e): add e2e tests for workbooks, datasources, views, pagination, jobs, tagging, projects, favorites, metadata, site-admin

pyproject.toml

@@ -70,6 +70,7 @@ testpaths = ["test"]
 addopts = "--junitxml=./test.junit.xml -n auto"
 markers = [
     "e2e: mark test as end-to-end (requires a real Tableau server)",
+    "e2e_admin: mark test as end-to-end requiring SiteAdmin credentials",
     "packaging: mark test as requiring a built wheel in dist/",
 ]
 

test_e2e/conftest.py

@@ -1,10 +1,22 @@
 import os
+import warnings
 import pytest
 import tableauserverclient as TSC
 
 
 def pytest_configure(config):
     config.addinivalue_line("markers", "e2e: mark test as end-to-end (requires a real Tableau server)")
+    config.addinivalue_line("markers", "e2e_admin: mark test as end-to-end requiring SiteAdmin credentials")
+
+
+def _http_options() -> dict:
+    verify = os.environ.get("TABLEAU_VERIFY_SSL", "true").lower() != "false"
+    if not verify:
+        warnings.warn(
+            "TABLEAU_VERIFY_SSL=false — TLS certificate verification is disabled for e2e tests.",
+            stacklevel=2,
+        )
+    return {"verify": verify}
 
 
 @pytest.fixture(scope="session")
@@ -26,7 +38,60 @@ def server():
     if not all([url, token, token_name]):
         pytest.skip("E2E tests require TABLEAU_SERVER, TABLEAU_TOKEN, and TABLEAU_TOKEN_NAME env vars")
 
-    server = TSC.Server(url, use_server_version=True)
+    server = TSC.Server(url, use_server_version=True, http_options=_http_options())
+    auth = TSC.PersonalAccessTokenAuth(token_name, token, site)
+    with server.auth.sign_in(auth):
+        yield server
+
+
+@pytest.fixture(scope="session")
+def server_admin():
+    """Authenticated TSC session using SiteAdmin credentials."""
+    url = os.environ.get("TABLEAU_SERVER")
+    site = os.environ.get("TABLEAU_SITE", "")
+    token = os.environ.get("TABLEAU_SITEADMIN_TOKEN")
+    token_name = os.environ.get("TABLEAU_SITEADMIN_TOKEN_NAME")
+
+    if not all([url, token, token_name]):
+        pytest.skip(
+            "Admin e2e tests require TABLEAU_SERVER, TABLEAU_SITEADMIN_TOKEN, and TABLEAU_SITEADMIN_TOKEN_NAME env vars"
+        )
+
+    server = TSC.Server(url, use_server_version=True, http_options=_http_options())
     auth = TSC.PersonalAccessTokenAuth(token_name, token, site)
     with server.auth.sign_in(auth):
         yield server
+
+
+@pytest.fixture(scope="session")
+def default_project(server):
+    """Return a project to publish into, shared across all test modules.
+
+    Uses TABLEAU_PROJECT env var if set (default: "Default"). Falls back to
+    "Personal Work" then the first available project.
+    """
+    project_name = os.environ.get("TABLEAU_PROJECT", "Default")
+    opts = TSC.RequestOptions()
+    opts.filter.add(TSC.Filter(TSC.RequestOptions.Field.Name, TSC.RequestOptions.Operator.Equals, project_name))
+    projects, _ = server.projects.get(opts)
+    if projects:
+        return projects[0]
+
+    for fallback in ("Personal Work",):
+        opts2 = TSC.RequestOptions()
+        opts2.filter.add(TSC.Filter(TSC.RequestOptions.Field.Name, TSC.RequestOptions.Operator.Equals, fallback))
+        fallback_projects, _ = server.projects.get(opts2)
+        if fallback_projects:
+            return fallback_projects[0]
+
+    all_projects, _ = server.projects.get()
+    if all_projects:
+        return all_projects[0]
+
+    pytest.skip(f"Project {project_name!r} not found — set TABLEAU_PROJECT env var")
+
+
+@pytest.fixture(scope="session")
+def project_id(default_project):
+    """Convenience fixture returning just the project ID string."""
+    return default_project.id

test_e2e/test_datasources_crud.py

@@ -0,0 +1,115 @@
+"""
+E2E tests for datasource CRUD operations against a real Tableau server.
+
+Run with:
+    TABLEAU_SERVER=https://... TABLEAU_SITE=mysite TABLEAU_TOKEN=... TABLEAU_TOKEN_NAME=... \
+    pytest test_e2e/test_datasources_crud.py -v
+"""
+
+import os
+import warnings
+from pathlib import Path
+
+import pytest
+import tableauserverclient as TSC
+from tableauserverclient.server.endpoint.exceptions import ServerResponseError
+
+ASSETS_DIR = Path(__file__).parent / "assets"
+SAMPLE_DATASOURCE = ASSETS_DIR / "WorldIndicators.tdsx"
+
+pytestmark = pytest.mark.e2e
+
+
+@pytest.fixture(scope="module")
+def datasource(server, project_id):
+    """Publish a datasource for CRUD tests; clean up after."""
+    ds_item = TSC.DatasourceItem(project_id=project_id, name="tsc-e2e-datasource-crud")
+    ds_item = server.datasources.publish(ds_item, SAMPLE_DATASOURCE, TSC.Server.PublishMode.Overwrite)
+    yield ds_item
+    server.datasources.delete(ds_item.id)
+
+
+def test_datasource_publish(server, datasource):
+    """datasources.publish() returns a DatasourceItem with an id."""
+    assert datasource.id is not None
+    assert datasource.name == "tsc-e2e-datasource-crud"
+
+
+def test_datasources_get(server, datasource):
+    """datasources.get() with a name filter returns the published datasource."""
+    opts = TSC.RequestOptions()
+    opts.filter.add(
+        TSC.Filter(TSC.RequestOptions.Field.Name, TSC.RequestOptions.Operator.Equals, "tsc-e2e-datasource-crud")
+    )
+    results, pagination = server.datasources.get(opts)
+    assert len(results) >= 1
+    assert any(ds.id == datasource.id for ds in results)
+
+
+def test_datasources_get_by_id(server, datasource):
+    """datasources.get_by_id() returns the correct DatasourceItem."""
+    fetched = server.datasources.get_by_id(datasource.id)
+    assert fetched.id == datasource.id
+    assert fetched.name == datasource.name
+
+
+def test_datasources_update_description(server, datasource):
+    """datasources.update() persists a description change."""
+    original_description = datasource.description
+    datasource.description = "e2e-test description"
+    server.datasources.update(datasource)
+    fetched = server.datasources.get_by_id(datasource.id)
+    try:
+        assert fetched.description == "e2e-test description"
+    finally:
+        datasource.description = original_description
+        server.datasources.update(datasource)
+
+
+def test_datasources_populate_connections(server, datasource):
+    """datasources.populate_connections() populates the connections list without error."""
+    fetched = server.datasources.get_by_id(datasource.id)
+    server.datasources.populate_connections(fetched)
+    connections = fetched.connections
+    assert isinstance(connections, list)
+
+
+def test_datasources_add_tags(server, datasource):
+    """datasources.add_tags() adds a tag that is visible via get_by_id()."""
+    server.datasources.add_tags(datasource, ["e2e-test"])
+    try:
+        fetched = server.datasources.get_by_id(datasource.id)
+        assert "e2e-test" in fetched.tags
+    finally:
+        server.datasources.delete_tags(datasource, ["e2e-test"])
+
+
+def test_datasources_delete_tags(server, datasource):
+    """datasources.delete_tags() removes a previously added tag."""
+    server.datasources.add_tags(datasource, ["e2e-test-delete"])
+    server.datasources.delete_tags(datasource, ["e2e-test-delete"])
+    fetched = server.datasources.get_by_id(datasource.id)
+    assert "e2e-test-delete" not in fetched.tags
+
+
+def test_datasources_download(server, datasource, tmp_path):
+    """datasources.download() writes a non-empty file."""
+    result_path = server.datasources.download(datasource.id, filepath=str(tmp_path))
+    assert os.path.isfile(result_path)
+    assert os.path.getsize(result_path) > 0
+
+
+def test_datasources_delete(server, project_id):
+    """datasources.delete() removes a datasource; subsequent get_by_id raises ServerResponseError."""
+    ds_item = TSC.DatasourceItem(project_id=project_id, name="tsc-e2e-datasource-delete")
+    published = server.datasources.publish(ds_item, SAMPLE_DATASOURCE, TSC.Server.PublishMode.Overwrite)
+    ds_id = published.id
+    try:
+        server.datasources.delete(ds_id)
+        with pytest.raises(ServerResponseError):
+            server.datasources.get_by_id(ds_id)
+    finally:
+        try:
+            server.datasources.delete(ds_id)
+        except Exception as exc:
+            warnings.warn(f"Failed to delete datasource: {exc}")

test_e2e/test_favorites.py

@@ -0,0 +1,79 @@
+"""
+E2E tests for favorites operations against a real Tableau server.
+
+Run with:
+    TABLEAU_SERVER=https://... TABLEAU_SITE=mysite TABLEAU_TOKEN=... TABLEAU_TOKEN_NAME=... \
+    pytest test_e2e/test_favorites.py -v
+"""
+
+from pathlib import Path
+
+import pytest
+import tableauserverclient as TSC
+from tableauserverclient.models import Resource
+
+ASSETS_DIR = Path(__file__).parent / "assets"
+SAMPLE_WORKBOOK = ASSETS_DIR / "WorkbookWithoutExtract.twbx"
+SAMPLE_DATASOURCE = ASSETS_DIR / "WorldIndicators.tdsx"
+
+pytestmark = pytest.mark.e2e
+
+
+@pytest.fixture(scope="module")
+def workbook(server, project_id):
+    wb = TSC.WorkbookItem(name="tsc-e2e-favorites-wb", project_id=project_id)
+    wb = server.workbooks.publish(wb, SAMPLE_WORKBOOK, TSC.Server.PublishMode.Overwrite)
+    try:
+        yield wb
+    finally:
+        server.workbooks.delete(wb.id)
+
+
+@pytest.fixture(scope="module")
+def datasource(server, project_id):
+    ds = TSC.DatasourceItem(project_id=project_id, name="tsc-e2e-favorites-ds")
+    ds = server.datasources.publish(ds, SAMPLE_DATASOURCE, TSC.Server.PublishMode.Overwrite)
+    try:
+        yield ds
+    finally:
+        server.datasources.delete(ds.id)
+
+
+def test_favorites_workbook(server, workbook):
+    """A workbook can be added to and removed from favorites."""
+    user = TSC.UserItem()
+    user.id = server.user_id
+    server.favorites.add_favorite(user, Resource.Workbook, workbook)
+    server.favorites.get(user)
+    try:
+        assert any(f.id == workbook.id for f in user.favorites.get("workbooks", []))
+    finally:
+        server.favorites.delete_favorite_workbook(user, workbook)
+
+
+def test_favorites_view(server, workbook):
+    """A view can be added to and removed from favorites."""
+    server.workbooks.populate_views(workbook)
+    if not workbook.views:
+        pytest.skip(f"Published workbook {workbook.name!r} has no views")
+    view = workbook.views[0]
+    user = TSC.UserItem()
+    user.id = server.user_id
+    server.favorites.add_favorite_view(user, view)
+    server.favorites.get(user)
+    try:
+        assert any(f.id == view.id for f in user.favorites.get("views", []))
+    finally:
+        server.favorites.delete_favorite_view(user, view)
+
+
+def test_favorites_datasource(server, datasource):
+    """A datasource can be added to and removed from favorites."""
+    user = TSC.UserItem()
+    user.id = server.user_id
+    server.favorites.add_favorite_datasource(user, datasource)
+    server.favorites.get(user)
+    try:
+        assert any(f.id == datasource.id for f in user.favorites.get("datasources", []))
+    finally:
+        server.favorites.delete_favorite_datasource(user, datasource)