diff --git a/test_e2e/assets/WorkbookWithExtract.twbx b/test_e2e/assets/WorkbookWithExtract.twbx new file mode 100644 index 000000000..b6025b36c Binary files /dev/null and b/test_e2e/assets/WorkbookWithExtract.twbx differ diff --git a/test_e2e/assets/WorldIndicators.tdsx b/test_e2e/assets/WorldIndicators.tdsx new file mode 100644 index 000000000..6e041442b Binary files /dev/null and b/test_e2e/assets/WorldIndicators.tdsx differ diff --git a/test_e2e/conftest.py b/test_e2e/conftest.py index 376d41a09..38e2d6835 100644 --- a/test_e2e/conftest.py +++ b/test_e2e/conftest.py @@ -5,6 +5,7 @@ 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") @pytest.fixture(scope="session") @@ -26,7 +27,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) + http_options = {"verify": os.environ.get("TABLEAU_VERIFY_SSL", "true").lower() != "false"} + 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") + + http_options = {"verify": os.environ.get("TABLEAU_VERIFY_SSL", "true").lower() != "false"} + 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 diff --git a/test_e2e/test_datasources_crud.py b/test_e2e/test_datasources_crud.py new file mode 100644 index 000000000..ffb475fc9 --- /dev/null +++ b/test_e2e/test_datasources_crud.py @@ -0,0 +1,113 @@ +""" +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 +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: + pass diff --git a/test_e2e/test_favorites.py b/test_e2e/test_favorites.py new file mode 100644 index 000000000..b9c3e8714 --- /dev/null +++ b/test_e2e/test_favorites.py @@ -0,0 +1,76 @@ +""" +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, str(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) + 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) diff --git a/test_e2e/test_jobs.py b/test_e2e/test_jobs.py new file mode 100644 index 000000000..253a8bf7d --- /dev/null +++ b/test_e2e/test_jobs.py @@ -0,0 +1,154 @@ +""" +E2E tests for Jobs and async operations against a real Tableau server. + +Run with: + TABLEAU_SERVER=https://... TABLEAU_SITE=mysite TABLEAU_TOKEN=... TABLEAU_TOKEN_NAME=... \ + pytest test_e2e/test_jobs.py -v +""" +from pathlib import Path + +import pytest +import tableauserverclient as TSC +from tableauserverclient.server.endpoint.exceptions import JobCancelledException, JobFailedException, ServerResponseError + +ASSETS_DIR = Path(__file__).parent / "assets" +SAMPLE_WORKBOOK = ASSETS_DIR / "WorkbookWithoutExtract.twbx" +EXTRACT_WORKBOOK = ASSETS_DIR / "WorkbookWithExtract.twbx" +SAMPLE_DATASOURCE = ASSETS_DIR / "WorldIndicators.tdsx" + +pytestmark = pytest.mark.e2e + + +@pytest.fixture(scope="module") +def workbook(server, project_id): + """Publish a workbook for jobs tests, clean up after.""" + wb = TSC.WorkbookItem(name="tsc-e2e-jobs-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): + """Publish a datasource for jobs tests, clean up after.""" + ds = TSC.DatasourceItem(project_id=project_id, name="tsc-e2e-jobs-ds") + ds = server.datasources.publish(ds, str(SAMPLE_DATASOURCE), TSC.Server.PublishMode.Overwrite) + try: + yield ds + finally: + server.datasources.delete(ds.id) + + +@pytest.fixture(scope="module") +def extract_workbook(server, project_id): + """Publish a workbook with an extract for refresh tests, clean up after.""" + if not EXTRACT_WORKBOOK.exists(): + pytest.skip(f"Extract workbook asset not found: {EXTRACT_WORKBOOK}") + wb = TSC.WorkbookItem(name="tsc-e2e-extract-wb", project_id=project_id) + wb = server.workbooks.publish(wb, EXTRACT_WORKBOOK, TSC.Server.PublishMode.Overwrite) + try: + yield wb + finally: + server.workbooks.delete(wb.id) + + +def test_jobs_get_returns_list(server): + """jobs.get() returns a list of background jobs and a pagination item.""" + jobs, pagination = server.jobs.get() + assert isinstance(jobs, list) + assert pagination is not None + assert pagination.total_available >= 0 + + +def test_workbook_refresh_returns_job(server, workbook): + """workbooks.refresh() returns a JobItem with a valid id.""" + try: + job = server.workbooks.refresh(workbook) + except ServerResponseError as e: + if "403180" in str(e) or "refresh" in str(e).lower(): + pytest.skip(f"Workbook does not support refresh on this server: {e}") + raise + if job is None: + pytest.skip("Duplicate refresh job already queued — skipping") + assert job.id is not None + assert isinstance(job.id, str) + + +def test_jobs_get_by_id(server, project_id): + """jobs.get_by_id() returns the correct JobItem for a running job.""" + wb = TSC.WorkbookItem(name="tsc-e2e-jobs-get-by-id-wb", project_id=project_id) + wb = server.workbooks.publish(wb, SAMPLE_WORKBOOK, TSC.Server.PublishMode.Overwrite) + try: + try: + job = server.workbooks.refresh(wb) + except ServerResponseError as e: + if "403180" in str(e) or "refresh" in str(e).lower(): + pytest.skip(f"Workbook does not support refresh on this server: {e}") + raise + if job is None: + pytest.skip("Duplicate refresh job already queued — skipping") + fetched = server.jobs.get_by_id(job.id) + assert fetched is not None + assert fetched.id == job.id + finally: + server.workbooks.delete(wb.id) + + +def test_workbook_refresh_job_completes(server, workbook): + """wait_for_job() on a workbook refresh completes without raising.""" + try: + job = server.workbooks.refresh(workbook) + except ServerResponseError as e: + if "403180" in str(e) or "refresh" in str(e).lower(): + pytest.skip(f"Workbook does not support refresh on this server: {e}") + raise + if job is None: + pytest.skip("Duplicate refresh job already queued — skipping") + try: + completed_job = server.jobs.wait_for_job(job, timeout=300) + except JobFailedException as e: + pytest.fail(f"Job failed — workbook must contain an extract for this test: {e}") + except JobCancelledException as e: + pytest.skip(f"Job was cancelled: {e}") + assert completed_job.finish_code in ( + TSC.JobItem.FinishCode.Success, TSC.JobItem.FinishCode.Completed + ) + assert completed_job.completed_at is not None + + +def test_datasource_refresh_job_completes(server, datasource): + """datasources.refresh() returns a job that completes successfully.""" + job = server.datasources.refresh(datasource) + if job is None: + pytest.skip("Duplicate refresh job already queued — skipping") + assert job is not None + assert job.id is not None + try: + completed_job = server.jobs.wait_for_job(job, timeout=300) + except JobFailedException as e: + pytest.skip(f"Datasource refresh job failed: {e}") + except JobCancelledException as e: + pytest.skip(f"Job was cancelled: {e}") + assert completed_job.finish_code in ( + TSC.JobItem.FinishCode.Success, TSC.JobItem.FinishCode.Completed + ) + assert completed_job.completed_at is not None + + +def test_extract_workbook_refresh_completes(server, extract_workbook): + """workbooks.refresh() on a workbook with an extract completes successfully.""" + job = server.workbooks.refresh(extract_workbook) + if job is None: + pytest.skip("Duplicate refresh job already queued — skipping") + try: + completed_job = server.jobs.wait_for_job(job, timeout=300) + except JobFailedException as e: + pytest.skip(f"Extract workbook refresh job failed (check asset compatibility): {e}") + except JobCancelledException as e: + pytest.skip(f"Job was cancelled: {e}") + assert completed_job.finish_code in ( + TSC.JobItem.FinishCode.Success, TSC.JobItem.FinishCode.Completed + ) + assert completed_job.completed_at is not None diff --git a/test_e2e/test_metadata.py b/test_e2e/test_metadata.py new file mode 100644 index 000000000..cb29d4b70 --- /dev/null +++ b/test_e2e/test_metadata.py @@ -0,0 +1,48 @@ +""" +E2E tests for the Metadata API (GraphQL) against a real Tableau server. + +Run with: + TABLEAU_SERVER=https://... TABLEAU_SITE=mysite TABLEAU_TOKEN=... TABLEAU_TOKEN_NAME=... \ + pytest test_e2e/test_metadata.py -v +""" +import pytest + +pytestmark = pytest.mark.e2e + + +def test_metadata_query_published_datasources(server): + """Metadata GraphQL API returns a valid response for publishedDatasourcesConnection.""" + result = server.metadata.query( + """ + { + publishedDatasourcesConnection(first: 5) { + nodes { + luid + name + } + } + } + """ + ) + assert "data" in result + assert "publishedDatasourcesConnection" in result["data"] + nodes = result["data"]["publishedDatasourcesConnection"]["nodes"] + assert isinstance(nodes, list) + + +def test_metadata_query_workbooks(server): + """Metadata GraphQL API returns workbook nodes.""" + result = server.metadata.query( + """ + { + workbooksConnection(first: 5) { + nodes { + luid + name + } + } + } + """ + ) + assert "data" in result + assert "workbooksConnection" in result["data"] diff --git a/test_e2e/test_pagination.py b/test_e2e/test_pagination.py new file mode 100644 index 000000000..e67e3067b --- /dev/null +++ b/test_e2e/test_pagination.py @@ -0,0 +1,123 @@ +""" +E2E tests for pagination via TSC.Pager and QuerySet against a real Tableau server. + +Run with: + TABLEAU_SERVER=https://... TABLEAU_SITE=mysite TABLEAU_TOKEN=... TABLEAU_TOKEN_NAME=... \ + pytest test_e2e/test_pagination.py -v +""" +import warnings +from pathlib import Path + +import pytest +import tableauserverclient as TSC + +ASSETS_DIR = Path(__file__).parent / "assets" +SAMPLE_WORKBOOK = ASSETS_DIR / "WorkbookWithoutExtract.twbx" +SAMPLE_DATASOURCE = ASSETS_DIR / "WorldIndicators.tdsx" + +PAGINATION_WB_NAMES = ["tsc-e2e-pagination-alpha", "tsc-e2e-pagination-zeta"] + +pytestmark = pytest.mark.e2e + + +@pytest.fixture(scope="module") +def workbooks_for_pagination(server, project_id): + published = [] + for name in PAGINATION_WB_NAMES: + wb = TSC.WorkbookItem(name=name, project_id=project_id) + wb = server.workbooks.publish(wb, SAMPLE_WORKBOOK, TSC.Server.PublishMode.Overwrite) + published.append(wb) + yield published + for wb in published: + try: + server.workbooks.delete(wb.id) + except Exception as exc: + warnings.warn(f"Failed to delete workbook: {exc}") + + +@pytest.fixture(scope="module") +def datasource_for_pagination(server, project_id): + ds = TSC.DatasourceItem(project_id=project_id, name="tsc-e2e-pagination-ds") + ds = server.datasources.publish(ds, SAMPLE_DATASOURCE, TSC.Server.PublishMode.Overwrite) + yield ds + try: + server.datasources.delete(ds.id) + except Exception: + pass + + +def test_pager_workbooks_count_matches_get(server, workbooks_for_pagination): + """TSC.Pager iterates all workbooks; count is at least the number of fixture workbooks.""" + _, pagination_item = server.workbooks.get() + total_declared = pagination_item.total_available + assert total_declared > 0, "Server has no workbooks — fixture likely failed" + + pager_count = sum(1 for _ in TSC.Pager(server.workbooks)) + + assert pager_count >= 2, ( + f"Pager yielded {pager_count} workbooks but expected at least 2 fixture workbooks" + ) + + +def test_pager_datasources_returns_items(server, datasource_for_pagination): + """TSC.Pager iterates all datasources; count matches total_available.""" + _, pagination_item = server.datasources.get() + total_declared = pagination_item.total_available + assert total_declared > 0, "Server has no datasources — fixture likely failed" + + pager_count = sum(1 for _ in TSC.Pager(server.datasources)) + + assert pager_count == total_declared, ( + f"Pager yielded {pager_count} datasources but server reported {total_declared}" + ) + + +def test_queryset_all_workbooks_matches_pager(server, workbooks_for_pagination): + """server.workbooks.all() QuerySet yields same count as Pager.""" + pager_count = sum(1 for _ in TSC.Pager(server.workbooks)) + queryset_count = sum(1 for _ in server.workbooks.all()) + + assert queryset_count > 0 + assert queryset_count == pager_count, ( + f"QuerySet yielded {queryset_count} but Pager yielded {pager_count}" + ) + + +def test_queryset_filter_by_name(server, workbooks_for_pagination): + """server.workbooks.filter(name=...) returns only workbooks with that exact name.""" + known_name = PAGINATION_WB_NAMES[0] + + results = list(server.workbooks.filter(name=known_name)) + + assert len(results) >= 1, f"No workbook named {known_name!r} returned by filter" + for wb in results: + assert wb.name == known_name, ( + f"Filter returned {wb.name!r}, expected {known_name!r}" + ) + + +def test_queryset_order_by_name_ascending(server, workbooks_for_pagination): + """server.workbooks.order_by('name') returns fixture workbooks sorted A to Z by name.""" + all_results = list(server.workbooks.order_by("name")) + fixture_name_set = set(PAGINATION_WB_NAMES) + fixture_results = [wb for wb in all_results if wb.name in fixture_name_set] + + assert len(fixture_results) >= 2, ( + f"Expected at least 2 fixture workbooks in results, got {len(fixture_results)}" + ) + names = [wb.name for wb in fixture_results] + assert names == sorted(names, key=str.casefold), ( + f"Fixture workbooks not in ascending case-insensitive name order: {names}" + ) + + +def test_pager_small_pagesize(server, workbooks_for_pagination): + """TSC.Pager with pagesize=1 correctly iterates all workbooks one page at a time.""" + request_opts = TSC.RequestOptions(pagesize=1) + items = list(TSC.Pager(server.workbooks.get, request_opts=request_opts)) + + assert len(items) >= 1, "Pager with pagesize=1 returned no items" + for item in items: + assert isinstance(item, TSC.WorkbookItem), ( + f"Expected WorkbookItem, got {type(item).__name__}" + ) diff --git a/test_e2e/test_projects_admin.py b/test_e2e/test_projects_admin.py new file mode 100644 index 000000000..5701ac717 --- /dev/null +++ b/test_e2e/test_projects_admin.py @@ -0,0 +1,150 @@ +""" +E2E tests for Projects CRUD operations against a real Tableau server (SiteAdmin). + +Run with: + TABLEAU_SERVER=https://... TABLEAU_SITE=mysite \ + TABLEAU_SITEADMIN_TOKEN_NAME=... TABLEAU_SITEADMIN_TOKEN=... \ + pytest test_e2e/test_projects_admin.py -v +""" +import uuid +import pytest +import tableauserverclient as TSC +from tableauserverclient.models.permissions_item import Permission, PermissionsRule +from tableauserverclient.models.reference_item import ResourceReference + +pytestmark = pytest.mark.e2e_admin + + +def _name(base: str) -> str: + return f"{base}-{uuid.uuid4().hex[:8]}" + + +def test_create_project(server_admin): + """projects.create() returns a ProjectItem with the correct name.""" + project_name = _name("tsc-e2e-proj") + project = TSC.ProjectItem(name=project_name) + created = server_admin.projects.create(project) + try: + assert created.id is not None + assert created.name == project_name + finally: + server_admin.projects.delete(created.id) + + +def test_get_includes_created_project(server_admin): + """After create, filter(name=...) includes the new project's ID.""" + project_name = _name("tsc-e2e-proj-list") + project = TSC.ProjectItem(name=project_name) + created = server_admin.projects.create(project) + try: + matches = list(server_admin.projects.filter(name=project_name)) + assert any(p.id == created.id for p in matches), ( + f"Newly created project {created.id!r} not found in projects.filter(name={project_name!r}) result" + ) + finally: + server_admin.projects.delete(created.id) + + +def test_update_project_description(server_admin): + """projects.update() with a new description persists on a subsequent get_by_id.""" + project = TSC.ProjectItem(name=_name("tsc-e2e-proj-update")) + created = server_admin.projects.create(project) + try: + created.description = "updated description" + updated = server_admin.projects.update(created) + assert updated.description == "updated description" + + fetched = server_admin.projects.get_by_id(created.id) + assert fetched.description == "updated description" + finally: + server_admin.projects.delete(created.id) + + +def test_create_child_project(server_admin): + """Creating a project with parent_id set results in a child project with matching parent_id.""" + parent = TSC.ProjectItem(name=_name("tsc-e2e-parent")) + created_parent = server_admin.projects.create(parent) + created_child = None + try: + child = TSC.ProjectItem(name=_name("tsc-e2e-child"), parent_id=created_parent.id) + created_child = server_admin.projects.create(child) + assert created_child.parent_id == created_parent.id + + fetched_child = server_admin.projects.get_by_id(created_child.id) + assert fetched_child.parent_id == created_parent.id + finally: + if created_child is not None: + try: + server_admin.projects.delete(created_child.id) + except Exception: + pass + server_admin.projects.delete(created_parent.id) + + +def test_delete_projects(server_admin): + """After deleting child then parent, neither appears in a subsequent filter.""" + parent_name = _name("tsc-e2e-del-parent") + child_name = _name("tsc-e2e-del-child") + parent = server_admin.projects.create(TSC.ProjectItem(name=parent_name)) + child = server_admin.projects.create( + TSC.ProjectItem(name=child_name, parent_id=parent.id) + ) + + parent_id = parent.id + child_id = child.id + + try: + server_admin.projects.delete(child_id) + server_admin.projects.delete(parent_id) + + remaining_parent = list(server_admin.projects.filter(name=parent_name)) + remaining_child = list(server_admin.projects.filter(name=child_name)) + + assert not any(p.id == parent_id for p in remaining_parent), "Parent project was not deleted" + assert not any(p.id == child_id for p in remaining_child), "Child project was not deleted" + finally: + try: + server_admin.projects.delete(child_id) + except Exception: + pass + try: + server_admin.projects.delete(parent_id) + except Exception: + pass + + +def test_populate_workbook_default_permissions(server_admin): + """populate_workbook_default_permissions() runs without error and the result is a list.""" + project = server_admin.projects.create(TSC.ProjectItem(name=_name("tsc-e2e-proj-perms"))) + try: + server_admin.projects.populate_workbook_default_permissions(project) + perms = project.default_workbook_permissions + assert isinstance(perms, list) + finally: + server_admin.projects.delete(project.id) + + +def test_update_workbook_default_permissions(server_admin): + """update_workbook_default_permissions() with a rule for All Users persists and is readable.""" + project = server_admin.projects.create(TSC.ProjectItem(name=_name("tsc-e2e-proj-update-perms"))) + try: + all_users_groups = list(server_admin.groups.filter(name="All Users")) + if not all_users_groups: + pytest.skip("'All Users' group not found on this site") + group = all_users_groups[0] + + grantee = ResourceReference(group.id, "group") + rule = PermissionsRule( + grantee=grantee, + capabilities={Permission.Capability.Read: Permission.Mode.Allow}, + ) + + updated_rules = server_admin.projects.update_workbook_default_permissions(project, [rule]) + assert isinstance(updated_rules, list) + assert any( + r.grantee.id == group.id + and r.capabilities.get(Permission.Capability.Read) == Permission.Mode.Allow + for r in updated_rules + ) + finally: + server_admin.projects.delete(project.id) diff --git a/test_e2e/test_projects_get_by_path.py b/test_e2e/test_projects_get_by_path.py new file mode 100644 index 000000000..74c465e4f --- /dev/null +++ b/test_e2e/test_projects_get_by_path.py @@ -0,0 +1,106 @@ +""" +E2E tests for server.projects.get_by_path() against a real Tableau server. + +Run publisher-level tests with: + TABLEAU_SERVER=https://... TABLEAU_SITE=mysite \ + TABLEAU_TOKEN=... TABLEAU_TOKEN_NAME=... \ + pytest test_e2e/test_projects_get_by_path.py -m e2e -v + +Run admin-level tests additionally with: + TABLEAU_SITEADMIN_TOKEN=... TABLEAU_SITEADMIN_TOKEN_NAME=... \ + pytest test_e2e/test_projects_get_by_path.py -m e2e_admin -v +""" +import pytest +import tableauserverclient as TSC + +# pytestmark is intentionally NOT set at module level because this file mixes +# @pytest.mark.e2e (publisher credentials) and @pytest.mark.e2e_admin (site-admin +# credentials) tests. A module-wide pytestmark would apply the same mark to every +# test, making it impossible to skip one class independently of the other. + + +def test_get_by_path_raises_for_empty_or_root_path(): + """get_by_path must raise ValueError for '' or '/' (both normalise to empty).""" + server = TSC.Server("https://example.com") + with pytest.raises(ValueError): + server.projects.get_by_path("") + with pytest.raises(ValueError): + server.projects.get_by_path("/") + + +@pytest.mark.e2e +def test_get_default_project_by_path(server, default_project): + """get_by_path('Default') returns the Default top-level project.""" + result = server.projects.get_by_path("Default") + assert result is not None, "Expected to find 'Default' project, got None" + assert result.id == default_project.id + assert result.name.lower() == "default" + + +@pytest.mark.e2e +def test_get_by_path_strips_leading_trailing_slashes(server, default_project): + """get_by_path('/Default/') must resolve to the same project as 'Default'.""" + result = server.projects.get_by_path("/Default/") + assert result is not None, "Expected to find project via path '/Default/', got None" + assert result.id == default_project.id, ( + f"Path '/Default/' resolved to project {result.id!r} " + f"but expected {default_project.id!r}" + ) + + +@pytest.mark.e2e +def test_get_by_path_returns_none_for_missing_project(server): + """get_by_path for a non-existent project must return None, not raise.""" + result = server.projects.get_by_path("DoesNotExistAtAll999") + assert result is None, f"Expected None for missing project, got {result!r}" + + +@pytest.mark.e2e_admin +def test_get_child_project_by_path(server_admin): + """get_by_path('Parent/Child') resolves to the correct nested child project.""" + parent_name = "tsc-e2e-gbp-parent" + child_name = "tsc-e2e-gbp-child" + parent = None + child = None + try: + parent = server_admin.projects.create(TSC.ProjectItem(name=parent_name)) + child = server_admin.projects.create( + TSC.ProjectItem(name=child_name, parent_id=parent.id) + ) + + result = server_admin.projects.get_by_path(f"{parent_name}/{child_name}") + + assert result is not None, ( + f"Expected to find child project at path '{parent_name}/{child_name}', got None" + ) + assert result.id == child.id, ( + f"get_by_path returned project {result.id!r} but expected child {child.id!r}" + ) + assert result.parent_id == parent.id, ( + f"Child project parent_id {result.parent_id!r} does not match parent {parent.id!r}" + ) + finally: + if child is not None: + server_admin.projects.delete(child.id) + if parent is not None: + server_admin.projects.delete(parent.id) + + +@pytest.mark.e2e_admin +def test_get_by_path_with_spaces_in_name(server_admin): + """get_by_path works correctly when the project name contains spaces.""" + project_name = "TSC E2E Spaced Project" + project = None + try: + project = server_admin.projects.create(TSC.ProjectItem(name=project_name)) + result = server_admin.projects.get_by_path(project_name) + assert result is not None, ( + f"Expected to find project '{project_name}' by path, got None" + ) + assert result.name == project_name, ( + f"Project name mismatch: expected {project_name!r}, got {result.name!r}" + ) + assert result.id == project.id + finally: + if project is not None: + server_admin.projects.delete(project.id) diff --git a/test_e2e/test_site_admin.py b/test_e2e/test_site_admin.py new file mode 100644 index 000000000..f0b832172 --- /dev/null +++ b/test_e2e/test_site_admin.py @@ -0,0 +1,202 @@ +""" +E2E tests for SiteAdmin-level operations against a real Tableau server. + +Requires SiteAdmin credentials: + TABLEAU_SITEADMIN_TOKEN_NAME=... TABLEAU_SITEADMIN_TOKEN=... + +Run with: + pytest test_e2e/test_site_admin.py -v -m e2e_admin +""" +import uuid +from datetime import time +from pathlib import Path + +import pytest +import tableauserverclient as TSC + +ASSETS_DIR = Path(__file__).parent / "assets" +SAMPLE_WORKBOOK = ASSETS_DIR / "WorkbookWithoutExtract.twbx" +SAMPLE_DATASOURCE = ASSETS_DIR / "WorldIndicators.tdsx" + +pytestmark = pytest.mark.e2e_admin + + +def _name(base): + return f"{base}-{uuid.uuid4().hex[:8]}" + + +# --------------------------------------------------------------------------- +# Schedules +# --------------------------------------------------------------------------- + +def test_schedule_create_and_delete(server_admin): + """An extract schedule can be created and deleted.""" + interval = TSC.HourlyInterval(start_time=time(3, 0), end_time=time(23, 0), interval_value=4) + schedule = TSC.ScheduleItem( + _name("tsc-e2e-schedule"), + 50, + TSC.ScheduleItem.Type.Extract, + TSC.ScheduleItem.ExecutionOrder.Parallel, + interval, + ) + schedule = server_admin.schedules.create(schedule) + try: + assert schedule.id is not None + fetched = server_admin.schedules.get_by_id(schedule.id) + assert fetched.id == schedule.id + finally: + server_admin.schedules.delete(schedule.id) + + +def test_schedule_add_workbook(server_admin, project_id): + """A workbook can be added to an extract refresh schedule.""" + interval = TSC.DailyInterval(start_time=time(4, 0)) + schedule_name = _name("tsc-e2e-sched-wb") + schedule = TSC.ScheduleItem( + schedule_name, + 60, + TSC.ScheduleItem.Type.Extract, + TSC.ScheduleItem.ExecutionOrder.Serial, + interval, + ) + schedule = server_admin.schedules.create(schedule) + wb = TSC.WorkbookItem(name=_name("tsc-e2e-sched-wb"), project_id=project_id) + wb = server_admin.workbooks.publish(wb, SAMPLE_WORKBOOK, TSC.Server.PublishMode.Overwrite) + try: + server_admin.schedules.add_to_schedule(schedule.id, wb) + tasks, _ = server_admin.tasks.get() + assert any(getattr(t, "schedule_id", None) == schedule.id for t in tasks) + finally: + server_admin.workbooks.delete(wb.id) + server_admin.schedules.delete(schedule.id) + + +# --------------------------------------------------------------------------- +# Webhooks +# --------------------------------------------------------------------------- + +def test_webhook_create_and_delete(server_admin): + """A webhook can be created and deleted.""" + webhook = TSC.WebhookItem() + webhook.name = _name("tsc-e2e-webhook") + webhook.url = "https://example.com/tsc-e2e-webhook" + webhook.event = "datasource-created" + webhook = server_admin.webhooks.create(webhook) + try: + assert webhook.id is not None + all_webhooks, _ = server_admin.webhooks.get() + assert any(w.id == webhook.id for w in all_webhooks) + finally: + server_admin.webhooks.delete(webhook.id) + + +# --------------------------------------------------------------------------- +# Connection update +# --------------------------------------------------------------------------- + +def test_datasource_update_connection(server_admin, project_id): + """A datasource connection's embed_password flag can be toggled via update_connection.""" + ds = TSC.DatasourceItem(project_id=project_id, name=_name("tsc-e2e-conn-ds")) + ds = server_admin.datasources.publish(ds, str(SAMPLE_DATASOURCE), TSC.Server.PublishMode.Overwrite) + try: + server_admin.datasources.populate_connections(ds) + if not ds.connections: + pytest.skip("Published datasource has no connections to update") + conn = ds.connections[0] + conn.embed_password = False + updated_conn = server_admin.datasources.update_connection(ds, conn) + assert updated_conn is not None + finally: + server_admin.datasources.delete(ds.id) + + +# --------------------------------------------------------------------------- +# Data freshness policy +# --------------------------------------------------------------------------- + +def test_workbook_data_freshness_policy(server_admin, project_id): + """Workbook data freshness policy can be set to AlwaysLive and back to SiteDefault.""" + wb = TSC.WorkbookItem(name=_name("tsc-e2e-freshness"), project_id=project_id) + wb = server_admin.workbooks.publish(wb, SAMPLE_WORKBOOK, TSC.Server.PublishMode.Overwrite) + try: + wb.data_freshness_policy = TSC.DataFreshnessPolicyItem(TSC.DataFreshnessPolicyItem.Option.AlwaysLive) + updated = server_admin.workbooks.update(wb) + assert updated.data_freshness_policy.option == TSC.DataFreshnessPolicyItem.Option.AlwaysLive + + wb.data_freshness_policy = TSC.DataFreshnessPolicyItem(TSC.DataFreshnessPolicyItem.Option.SiteDefault) + updated = server_admin.workbooks.update(wb) + assert updated.data_freshness_policy.option == TSC.DataFreshnessPolicyItem.Option.SiteDefault + finally: + server_admin.workbooks.delete(wb.id) + + +# --------------------------------------------------------------------------- +# Workbook move between projects +# --------------------------------------------------------------------------- + +def test_workbook_move_project(server_admin, project_id): + """A workbook can be moved from one project to another.""" + dest_name = _name("tsc-e2e-dest-proj") + dest = TSC.ProjectItem(name=dest_name) + dest = server_admin.projects.create(dest) + wb = TSC.WorkbookItem(name=_name("tsc-e2e-move-wb"), project_id=project_id) + wb = server_admin.workbooks.publish(wb, SAMPLE_WORKBOOK, TSC.Server.PublishMode.Overwrite) + try: + wb.project_id = dest.id + updated = server_admin.workbooks.update(wb) + assert updated.project_id == dest.id + finally: + server_admin.workbooks.delete(wb.id) + server_admin.projects.delete(dest.id) + + +# --------------------------------------------------------------------------- +# Users +# --------------------------------------------------------------------------- + +def test_user_add_and_remove(server_admin): + """A user can be added to the site and removed.""" + username = _name("tsc-e2e-user") + user = TSC.UserItem(username, "Unlicensed") + user = server_admin.users.add(user) + try: + assert user.id is not None + fetched = server_admin.users.get_by_id(user.id) + assert fetched.name == username + finally: + server_admin.users.remove(user.id) + + +# --------------------------------------------------------------------------- +# Groups +# --------------------------------------------------------------------------- + +@pytest.fixture(scope="module") +def group(server_admin): + g = TSC.GroupItem(_name("tsc-e2e-admin-grp")) + g = server_admin.groups.create(g) + yield g + server_admin.groups.delete(g.id) + + +def test_group_create_and_get(server_admin, group): + """Created group appears in the group list.""" + results = list(server_admin.groups.filter(name=group.name)) + assert any(g.id == group.id for g in results) + + +def test_group_add_and_remove_user(server_admin, group): + """A user can be added to and removed from a group.""" + username = _name("tsc-e2e-grp-user") + user = TSC.UserItem(username, "Unlicensed") + user = server_admin.users.add(user) + try: + server_admin.groups.add_user(group, user.id) + server_admin.groups.populate_users(group) + assert any(u.id == user.id for u in group.users) + + server_admin.groups.remove_user(group, user.id) + server_admin.groups.populate_users(group) + assert all(u.id != user.id for u in group.users) + finally: + server_admin.users.remove(user.id) diff --git a/test_e2e/test_tagging.py b/test_e2e/test_tagging.py index 4af741255..facbd5079 100644 --- a/test_e2e/test_tagging.py +++ b/test_e2e/test_tagging.py @@ -5,7 +5,6 @@ TABLEAU_SERVER=https://... TABLEAU_SITE=mysite TABLEAU_TOKEN=... TABLEAU_TOKEN_NAME=... \ pytest test_e2e/test_tagging.py -v """ -import os from pathlib import Path import pytest @@ -13,66 +12,134 @@ 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): - """Publish a workbook for tagging tests, clean up after. - - Uses TABLEAU_PROJECT env var if set, otherwise falls back to the first - project named 'Default' or '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 not projects: - pytest.skip(f"Project {project_name!r} not found — set TABLEAU_PROJECT env var") - project = projects[0] - - wb = TSC.WorkbookItem(name="tsc-e2e-tagging-test", project_id=project.id) +def workbook(server, project_id): + """Publish a workbook for tagging tests, clean up after.""" + wb = TSC.WorkbookItem(name="tsc-e2e-tagging-test", project_id=project_id) wb = server.workbooks.publish(wb, SAMPLE_WORKBOOK, TSC.Server.PublishMode.Overwrite) yield wb server.workbooks.delete(wb.id) -def test_tag_with_spaces_stored_as_single_tag(server, workbook): - """A tag containing a space must be stored as one tag, not split on the space.""" - spaced_tag = "Yearly Sales" - server.workbooks.add_tags(workbook, spaced_tag) +@pytest.fixture(scope="module") +def datasource(server, project_id): + """Publish a datasource for tagging tests, clean up after.""" + if not SAMPLE_DATASOURCE.exists(): + pytest.skip(f"Datasource asset not found: {SAMPLE_DATASOURCE}") + + ds = TSC.DatasourceItem(project_id=project_id, name="tsc-e2e-tagging-test-ds") + ds = server.datasources.publish(ds, SAMPLE_DATASOURCE, TSC.Server.PublishMode.Overwrite) + yield ds + server.datasources.delete(ds.id) + + +def test_multiple_tags_including_spaced(server, workbook): + """Adding multiple tags where one has a space should all round-trip correctly.""" + tags = ["simple", "Yearly Sales", "another tag"] + server.workbooks.add_tags(workbook, tags) updated = server.workbooks.get_by_id(workbook.id) try: - assert spaced_tag in updated.tags, ( - f"Tag '{spaced_tag}' not found in {updated.tags!r} — was it split on the space?" - ) - assert "Yearly" not in updated.tags, "Tag was incorrectly split — 'Yearly' should not be a separate tag" - assert "Sales" not in updated.tags, "Tag was incorrectly split — 'Sales' should not be a separate tag" + for tag in tags: + assert tag in updated.tags, f"Tag '{tag}' not found in {updated.tags!r}" finally: - server.workbooks.delete_tags(workbook, spaced_tag) + server.workbooks.delete_tags(workbook, tags) -def test_tag_with_comma_stored_as_single_tag(server, workbook): - """A tag containing a comma must be stored as one tag, not split on the comma.""" - comma_tag = "Sales,Marketing" - server.workbooks.add_tags(workbook, comma_tag) +def test_add_tag_with_space(server, workbook): + """add_tags with a space-containing tag round-trips as a single tag (not split).""" + tag = "Yearly Sales" + server.workbooks.add_tags(workbook, tag) updated = server.workbooks.get_by_id(workbook.id) try: - assert comma_tag in updated.tags, ( - f"Tag '{comma_tag}' not found in {updated.tags!r} — was it split on the comma?" + assert tag in updated.tags, ( + f"Tag {tag!r} not found in {updated.tags!r} — was it split on the space?" + ) + assert '"Yearly Sales"' not in updated.tags, ( + "Tag was stored with literal surrounding quotes — double-quoting leaked into the label" ) + assert "Yearly" not in updated.tags, "Tag was split — 'Yearly' must not be a standalone tag" + assert "Sales" not in updated.tags, "Tag was split — 'Sales' must not be a standalone tag" finally: - server.workbooks.delete_tags(workbook, comma_tag) + server.workbooks.delete_tags(workbook, tag) -def test_multiple_tags_including_spaced(server, workbook): - """Adding multiple tags where one has a space should all round-trip correctly.""" - tags = ["simple", "Yearly Sales", "another tag"] - server.workbooks.add_tags(workbook, tags) +def test_add_tag_with_comma(server, workbook): + """add_tags with a comma-containing tag is stored as exactly one tag, not split on the comma.""" + tag = "Sales, Revenue" + server.workbooks.add_tags(workbook, tag) updated = server.workbooks.get_by_id(workbook.id) try: - for tag in tags: - assert tag in updated.tags, f"Tag '{tag}' not found in {updated.tags!r}" + assert tag in updated.tags, ( + f"Tag {tag!r} not found in {updated.tags!r} — was it split on the comma?" + ) + assert "Sales" not in updated.tags, "Tag was split — 'Sales' must not be a standalone tag" + assert "Revenue" not in updated.tags, "Tag was split — 'Revenue' must not be a standalone tag" finally: - server.workbooks.delete_tags(workbook, tags) + server.workbooks.delete_tags(workbook, tag) + + +def test_delete_tag_with_space(server, workbook): + """A space-containing tag added then deleted must no longer appear on the workbook.""" + tag = "Yearly Sales" + server.workbooks.add_tags(workbook, tag) + try: + after_add = server.workbooks.get_by_id(workbook.id) + assert tag in after_add.tags, ( + f"Precondition failed: tag {tag!r} not in {after_add.tags!r} after add_tags" + ) + server.workbooks.delete_tags(workbook, tag) + after_delete = server.workbooks.get_by_id(workbook.id) + assert tag not in after_delete.tags, ( + f"Tag {tag!r} still present in {after_delete.tags!r} after delete_tags — " + "delete may have failed to match the space-quoting used at write time" + ) + finally: + try: + server.workbooks.delete_tags(workbook, tag) + except Exception: + pass + + +def test_delete_tag_with_comma(server, workbook): + """A comma-containing tag added then deleted must no longer appear on the workbook.""" + tag = "Sales, Revenue" + server.workbooks.add_tags(workbook, tag) + try: + after_add = server.workbooks.get_by_id(workbook.id) + assert tag in after_add.tags, ( + f"Precondition failed: tag {tag!r} not in {after_add.tags!r} after add_tags" + ) + server.workbooks.delete_tags(workbook, tag) + after_delete = server.workbooks.get_by_id(workbook.id) + assert tag not in after_delete.tags, ( + f"Tag {tag!r} still present in {after_delete.tags!r} after delete_tags — " + "delete may have failed to URL-encode the comma in the delete path" + ) + finally: + try: + server.workbooks.delete_tags(workbook, tag) + except Exception: + pass + + +def test_add_space_tag_to_datasource(server, datasource): + """A space-containing tag round-trips correctly when applied to a datasource (not a workbook).""" + tag = "Yearly Sales" + server.datasources.add_tags(datasource, tag) + updated = server.datasources.get_by_id(datasource.id) + try: + assert tag in updated.tags, ( + f"Tag {tag!r} not found in datasource tags {updated.tags!r} — was it split on the space?" + ) + assert '"Yearly Sales"' not in updated.tags, ( + "Tag was stored with literal surrounding quotes on the datasource" + ) + assert "Yearly" not in updated.tags, "Tag was split — 'Yearly' must not be a standalone datasource tag" + assert "Sales" not in updated.tags, "Tag was split — 'Sales' must not be a standalone datasource tag" + finally: + server.datasources.delete_tags(datasource, tag) diff --git a/test_e2e/test_users_groups.py b/test_e2e/test_users_groups.py new file mode 100644 index 000000000..828dee70d --- /dev/null +++ b/test_e2e/test_users_groups.py @@ -0,0 +1,159 @@ +""" +E2E tests for user and group operations against a real Tableau Server. +Requires SiteAdmin credentials. + +Run with: + TABLEAU_SERVER=https://... TABLEAU_SITE=mysite \ + TABLEAU_SITEADMIN_TOKEN_NAME=... TABLEAU_SITEADMIN_TOKEN=... \ + pytest test_e2e/test_users_groups.py -v +""" +import pytest +import tableauserverclient as TSC + +pytestmark = pytest.mark.e2e_admin + + +def test_users_get_returns_nonempty_list(server_admin): + """users.get() returns at least the current (admin) user.""" + users, pagination = server_admin.users.get() + assert len(users) >= 1 + assert pagination.total_available >= 1 + + +def test_groups_get_returns_nonempty_list(server_admin): + """groups.get() returns at least one group.""" + groups, pagination = server_admin.groups.get() + assert len(groups) >= 1 + assert pagination.total_available >= 1 + + +def test_user_add_creates_user(server_admin): + """users.add() creates a new local user and returns it with an assigned id.""" + new_user = TSC.UserItem("tsc-e2e-user", TSC.UserItem.Roles.Unlicensed) + created = None + try: + created = server_admin.users.add(new_user) + assert created.id is not None + assert created.name == "tsc-e2e-user" + finally: + if created and created.id: + server_admin.users.remove(created.id) + + +def test_user_get_by_id_returns_user(server_admin): + """users.get_by_id() returns the correct user for a known id.""" + new_user = TSC.UserItem("tsc-e2e-user-byid", TSC.UserItem.Roles.Unlicensed) + created = None + try: + created = server_admin.users.add(new_user) + fetched = server_admin.users.get_by_id(created.id) + assert fetched.id == created.id + assert fetched.name == "tsc-e2e-user-byid" + finally: + if created and created.id: + server_admin.users.remove(created.id) + + +def test_user_update_changes_role(server_admin): + """users.update() with a new site_role persists the change.""" + new_user = TSC.UserItem("tsc-e2e-user-update", TSC.UserItem.Roles.Unlicensed) + created = None + try: + created = server_admin.users.add(new_user) + created.site_role = TSC.UserItem.Roles.Viewer + updated = server_admin.users.update(created) + assert updated.site_role == TSC.UserItem.Roles.Viewer + + # Verify persisted server-side + fetched = server_admin.users.get_by_id(created.id) + assert fetched.site_role == TSC.UserItem.Roles.Viewer + finally: + if created and created.id: + server_admin.users.remove(created.id) + + +def test_group_create_creates_group(server_admin): + """groups.create() creates a local group and returns it with an assigned id.""" + new_group = TSC.GroupItem("tsc-e2e-group") + created = None + try: + created = server_admin.groups.create(new_group) + assert created.id is not None + assert created.name == "tsc-e2e-group" + finally: + if created and created.id: + server_admin.groups.delete(created.id) + + +def test_group_add_user_appears_in_populate_users(server_admin): + """groups.add_user() results in the user appearing in groups.populate_users().""" + user = None + group = None + try: + user = server_admin.users.add(TSC.UserItem("tsc-e2e-user-grp", TSC.UserItem.Roles.Unlicensed)) + group = server_admin.groups.create(TSC.GroupItem("tsc-e2e-group-add")) + + server_admin.groups.add_user(group, user.id) + + server_admin.groups.populate_users(group) + user_ids = [u.id for u in group.users] + assert user.id in user_ids + finally: + if group and group.id: + server_admin.groups.delete(group.id) + if user and user.id: + server_admin.users.remove(user.id) + + +def test_group_remove_user_no_longer_in_group(server_admin): + """groups.remove_user() removes user from group; user is no longer listed.""" + user = None + group = None + try: + user = server_admin.users.add(TSC.UserItem("tsc-e2e-user-rmv", TSC.UserItem.Roles.Unlicensed)) + group = server_admin.groups.create(TSC.GroupItem("tsc-e2e-group-rmv")) + + server_admin.groups.add_user(group, user.id) + server_admin.groups.remove_user(group, user.id) + + server_admin.groups.populate_users(group) + user_ids = [u.id for u in group.users] + assert user.id not in user_ids + finally: + if group and group.id: + server_admin.groups.delete(group.id) + if user and user.id: + server_admin.users.remove(user.id) + + +def test_group_delete_removes_group(server_admin): + """groups.delete() removes the group; it no longer appears in groups.get().""" + created = None + try: + created = server_admin.groups.create(TSC.GroupItem("tsc-e2e-group-del")) + group_id = created.id + server_admin.groups.delete(group_id) + created = None + + remaining = list(server_admin.groups.filter(name="tsc-e2e-group-del")) + assert len(remaining) == 0 + finally: + if created and created.id: + server_admin.groups.delete(created.id) + + +def test_user_remove_removes_user(server_admin): + """users.remove() removes the user; they no longer appear in users.get().""" + new_user = TSC.UserItem("tsc-e2e-user-del", TSC.UserItem.Roles.Unlicensed) + created = None + try: + created = server_admin.users.add(new_user) + user_id = created.id + server_admin.users.remove(user_id) + created = None + + remaining = list(server_admin.users.filter(name="tsc-e2e-user-del")) + assert len(remaining) == 0 + finally: + if created and created.id: + server_admin.users.remove(created.id) diff --git a/test_e2e/test_views_export.py b/test_e2e/test_views_export.py new file mode 100644 index 000000000..27bb2d4f2 --- /dev/null +++ b/test_e2e/test_views_export.py @@ -0,0 +1,92 @@ +""" +E2E tests for view export operations (PDF, PNG, CSV) against a real Tableau server. + +Run with: + TABLEAU_SERVER=https://... TABLEAU_SITE=mysite TABLEAU_TOKEN=... TABLEAU_TOKEN_NAME=... \ + pytest test_e2e/test_views_export.py -v +""" +from pathlib import Path + +import pytest +import tableauserverclient as TSC + +ASSETS_DIR = Path(__file__).parent / "assets" +SAMPLE_WORKBOOK = ASSETS_DIR / "WorkbookWithoutExtract.twbx" + +pytestmark = pytest.mark.e2e + + +@pytest.fixture(scope="module") +def workbook_with_view(server, project_id): + """Publish a workbook for export tests, clean up after.""" + wb = TSC.WorkbookItem(name="tsc-e2e-views-export-test", 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 a_view(server, workbook_with_view): + """Return the first view from the published workbook.""" + server.workbooks.populate_views(workbook_with_view) + views = workbook_with_view.views + if not views: pytest.fail("Published workbook has no views") + return views[0] + + +def test_views_get_returns_views(server, workbook_with_view): + """server.views.get() returns at least one view after a workbook is published.""" + all_views, pagination = server.views.get() + assert pagination.total_available >= 1 + assert len(all_views) >= 1 + + +def test_populate_pdf_default_options(server, a_view): + """populate_pdf() with no options produces non-empty PDF bytes.""" + server.views.populate_pdf(a_view) + pdf_bytes = a_view.pdf + assert isinstance(pdf_bytes, bytes) + assert len(pdf_bytes) > 0 + assert pdf_bytes[:4] == b"%PDF", "Response does not have PDF magic bytes" + + +def test_populate_pdf_with_page_options(server, a_view): + """populate_pdf() with A4/Landscape options produces non-empty PDF bytes.""" + opts = TSC.PDFRequestOptions( + page_type=TSC.PDFRequestOptions.PageType.A4, + orientation=TSC.PDFRequestOptions.Orientation.Landscape, + ) + server.views.populate_pdf(a_view, opts) + pdf_bytes = a_view.pdf + assert isinstance(pdf_bytes, bytes) + assert len(pdf_bytes) > 0 + assert pdf_bytes[:4] == b"%PDF", "Response does not have PDF magic bytes" + + +def test_populate_image_default_options(server, a_view): + """populate_image() with no options produces non-empty PNG bytes.""" + server.views.populate_image(a_view) + image_bytes = a_view.image + assert isinstance(image_bytes, bytes) + assert len(image_bytes) > 0 + assert image_bytes[:8] == b"\x89PNG\r\n\x1a\n", "Response does not have PNG magic bytes" + + +def test_populate_image_high_resolution(server, a_view): + """populate_image() with high-resolution option produces non-empty PNG bytes.""" + opts = TSC.ImageRequestOptions(imageresolution=TSC.ImageRequestOptions.Resolution.High) + server.views.populate_image(a_view, opts) + image_bytes = a_view.image + assert isinstance(image_bytes, bytes) + assert len(image_bytes) > 0 + assert image_bytes[:8] == b"\x89PNG\r\n\x1a\n", "Response does not have PNG magic bytes" + + +def test_populate_csv(server, a_view): + """populate_csv() produces non-empty CSV bytes.""" + server.views.populate_csv(a_view) + csv_bytes = b"".join(a_view.csv) + assert isinstance(csv_bytes, bytes) + assert len(csv_bytes) > 0 diff --git a/test_e2e/test_workbook_permissions.py b/test_e2e/test_workbook_permissions.py new file mode 100644 index 000000000..f2a320a18 --- /dev/null +++ b/test_e2e/test_workbook_permissions.py @@ -0,0 +1,168 @@ +""" +E2E tests for workbook and view permission operations against a real Tableau server. + +Run with: + TABLEAU_SERVER=https://... TABLEAU_SITE=mysite \ + TABLEAU_SITEADMIN_TOKEN_NAME=... TABLEAU_SITEADMIN_TOKEN=... \ + pytest test_e2e/test_workbook_permissions.py -v +""" +import os +from pathlib import Path + +import pytest +import tableauserverclient as TSC + +ASSETS_DIR = Path(__file__).parent / "assets" +SAMPLE_WORKBOOK = ASSETS_DIR / "WorkbookWithoutExtract.twbx" + +pytestmark = pytest.mark.e2e_admin + + +@pytest.fixture(scope="module") +def workbook_and_user(server_admin): + """Publish a workbook and create a test user for permissions tests; clean up after. + + Uses TABLEAU_PROJECT env var if set, otherwise falls back to 'Default'. + """ + 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_admin.projects.get(opts) + if not projects: + pytest.skip(f"Project {project_name!r} not found — set TABLEAU_PROJECT env var") + project = projects[0] + + wb = TSC.WorkbookItem(name="tsc-e2e-permissions-test", project_id=project.id) + wb = server_admin.workbooks.publish(wb, SAMPLE_WORKBOOK, TSC.Server.PublishMode.Overwrite) + + try: + user = TSC.UserItem("tsc-e2e-perm-testuser", TSC.UserItem.Roles.Viewer) + user = server_admin.users.add(user) + except Exception: + server_admin.workbooks.delete(wb.id) + raise + + yield wb, user + + # Teardown — always attempt both deletes + try: + server_admin.workbooks.delete(wb.id) + finally: + server_admin.users.remove(user.id) + + +def test_populate_permissions_returns_list(server_admin, workbook_and_user): + """populate_permissions populates workbook.permissions as a list of PermissionsRule objects.""" + workbook, _ = workbook_and_user + server_admin.workbooks.populate_permissions(workbook) + assert workbook.permissions is not None + assert isinstance(workbook.permissions, list) + for rule in workbook.permissions: + assert isinstance(rule, TSC.PermissionsRule) + assert isinstance(rule.capabilities, dict) + assert rule.grantee is not None + + +def test_update_permissions_appears_on_populate(server_admin, workbook_and_user): + """A permission granted via update_permissions is visible after populate_permissions.""" + workbook, user = workbook_and_user + + grantee = TSC.UserItem.as_reference(user.id) + new_rule = TSC.PermissionsRule( + grantee, + {TSC.Permission.Capability.Read: TSC.Permission.Mode.Allow}, + ) + + try: + server_admin.workbooks.update_permissions(workbook, [new_rule]) + + # Re-populate to get fresh permissions from the server + server_admin.workbooks.populate_permissions(workbook) + user_rules = [ + r for r in workbook.permissions + if r.grantee.id == user.id and r.grantee.tag_name == "user" + ] + assert len(user_rules) == 1 + assert ( + user_rules[0].capabilities.get(TSC.Permission.Capability.Read) + == TSC.Permission.Mode.Allow + ) + finally: + # Clean up: build the delete rule directly to avoid masking the original exception + delete_rule = TSC.PermissionsRule( + grantee, + {TSC.Permission.Capability.Read: TSC.Permission.Mode.Allow}, + ) + server_admin.workbooks.delete_permission(workbook, delete_rule) + + +def test_view_populate_permissions_returns_list(server_admin, workbook_and_user): + """populate_permissions on a view returns a list of PermissionsRule objects.""" + workbook, _ = workbook_and_user + + server_admin.workbooks.populate_views(workbook) + assert workbook.views, "Workbook has no views — cannot test view permissions" + view = workbook.views[0] + + server_admin.views.populate_permissions(view) + assert view.permissions is not None + assert isinstance(view.permissions, list) + for rule in view.permissions: + assert isinstance(rule, TSC.PermissionsRule) + + +def test_delete_permission_removes_rule(server_admin, workbook_and_user): + """A permission deleted via delete_permission is no longer returned by populate_permissions.""" + workbook, user = workbook_and_user + + grantee = TSC.UserItem.as_reference(user.id) + new_rule = TSC.PermissionsRule( + grantee, + {TSC.Permission.Capability.ExportImage: TSC.Permission.Mode.Allow}, + ) + + try: + # Grant permission + server_admin.workbooks.update_permissions(workbook, [new_rule]) + + # Confirm it was granted + server_admin.workbooks.populate_permissions(workbook) + user_rules = [ + r for r in workbook.permissions + if r.grantee.id == user.id and r.grantee.tag_name == "user" + ] + assert any( + r.capabilities.get(TSC.Permission.Capability.ExportImage) == TSC.Permission.Mode.Allow + for r in user_rules + ), "ExportImage/Allow rule not found after update_permissions" + + # Delete the permission rule + for rule in user_rules: + server_admin.workbooks.delete_permission(workbook, rule) + + # Confirm it is gone + server_admin.workbooks.populate_permissions(workbook) + remaining_user_rules = [ + r for r in workbook.permissions + if r.grantee.id == user.id and r.grantee.tag_name == "user" + ] + assert not any( + r.capabilities.get(TSC.Permission.Capability.ExportImage) == TSC.Permission.Mode.Allow + for r in remaining_user_rules + ), "ExportImage/Allow rule still present after delete_permission" + finally: + # Ensure the granted permission is cleaned up even if assertions fail + delete_rule = TSC.PermissionsRule( + grantee, + {TSC.Permission.Capability.ExportImage: TSC.Permission.Mode.Allow}, + ) + try: + server_admin.workbooks.delete_permission(workbook, delete_rule) + except Exception: + pass # Rule may already be deleted by the test body diff --git a/test_e2e/test_workbooks_crud.py b/test_e2e/test_workbooks_crud.py new file mode 100644 index 000000000..a0f0c0c54 --- /dev/null +++ b/test_e2e/test_workbooks_crud.py @@ -0,0 +1,100 @@ +""" +E2E tests for workbook CRUD operations against a real Tableau server. + +Run with: + TABLEAU_SERVER=https://... TABLEAU_SITE=mysite TABLEAU_TOKEN=... TABLEAU_TOKEN_NAME=... \ + pytest test_e2e/test_workbooks_crud.py -v +""" +import io +from pathlib import Path + +import pytest +import tableauserverclient as TSC +from tableauserverclient.server.endpoint.exceptions import ServerResponseError + +ASSETS_DIR = Path(__file__).parent / "assets" +SAMPLE_WORKBOOK = ASSETS_DIR / "WorkbookWithoutExtract.twbx" + +pytestmark = pytest.mark.e2e + + +@pytest.fixture(scope="module") +def published_workbook(server, project_id): + """Publish WorkbookWithoutExtract.twbx once for all CRUD tests; delete after module.""" + wb = TSC.WorkbookItem(name="tsc-e2e-crud-test", project_id=project_id) + wb = server.workbooks.publish(wb, SAMPLE_WORKBOOK, TSC.Server.PublishMode.Overwrite) + yield wb + server.workbooks.delete(wb.id) + + +def test_workbook_publish_returns_item_with_id(published_workbook): + """publish() returns a WorkbookItem with a server-assigned ID and the expected name.""" + assert published_workbook.id is not None + assert published_workbook.name == "tsc-e2e-crud-test" + + +def test_workbook_get_lists_published_workbook(server, published_workbook): + """workbooks.filter() with a name filter returns the published workbook.""" + results = list(server.workbooks.filter(name="tsc-e2e-crud-test")) + assert any(wb.id == published_workbook.id for wb in results), ( + f"Published workbook {published_workbook.id!r} not found in filter results: " + f"{[wb.id for wb in results]}" + ) + + +def test_workbook_get_by_id_returns_correct_item(server, published_workbook): + """get_by_id() returns the exact workbook that was published.""" + fetched = server.workbooks.get_by_id(published_workbook.id) + assert fetched.id == published_workbook.id + assert fetched.name == published_workbook.name + + +def test_workbook_update_description_persists(server, published_workbook): + """update() persists a changed description on the server.""" + new_description = "tsc-e2e updated description" + fresh = server.workbooks.get_by_id(published_workbook.id) + fresh.description = new_description + server.workbooks.update(fresh) + fetched = server.workbooks.get_by_id(published_workbook.id) + assert fetched.description == new_description + + +def test_workbook_populate_views_returns_at_least_one_view(server, published_workbook): + """populate_views() sets workbook_item.views with at least one ViewItem.""" + server.workbooks.populate_views(published_workbook) + assert published_workbook.views is not None + assert len(published_workbook.views) >= 1 + + +def test_workbook_populate_connections_does_not_raise(server, published_workbook): + """populate_connections() sets the connections attribute without error.""" + server.workbooks.populate_connections(published_workbook) + assert published_workbook.connections is not None + + +def test_workbook_populate_preview_image(server, published_workbook): + """populate_preview_image() sets preview_image without error.""" + server.workbooks.populate_preview_image(published_workbook) + assert published_workbook.preview_image is not None + + +def test_workbook_download_returns_nonempty_bytes(server, published_workbook): + """download() writes a non-empty file into a BytesIO buffer.""" + buffer = io.BytesIO() + server.workbooks.download(published_workbook.id, filepath=buffer) + assert buffer.tell() > 0 + + +def test_workbook_delete_removes_workbook(server, project_id): + """delete() removes a workbook; get_by_id afterwards raises ServerResponseError.""" + wb = TSC.WorkbookItem(name="tsc-e2e-delete-test", project_id=project_id) + wb = server.workbooks.publish(wb, SAMPLE_WORKBOOK, TSC.Server.PublishMode.Overwrite) + try: + server.workbooks.delete(wb.id) + with pytest.raises(ServerResponseError): + server.workbooks.get_by_id(wb.id) + finally: + try: + server.workbooks.delete(wb.id) + except ServerResponseError: + pass