Skip to content

chore: pytestify metadata #1688

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: development
Choose a base branch
from
Open

chore: pytestify metadata #1688

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
166 changes: 87 additions & 79 deletions test/test_metadata.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,21 +1,22 @@
import json
import os.path
from pathlib import Path
import unittest

import pytest
import requests_mock

import tableauserverclient as TSC
from tableauserverclient.server.endpoint.exceptions import GraphQLError

TEST_ASSET_DIR = os.path.join(os.path.dirname(__file__), "assets")
TEST_ASSET_DIR = Path(__file__).parent / "assets"

METADATA_QUERY_SUCCESS = os.path.join(TEST_ASSET_DIR, "metadata_query_success.json")
METADATA_QUERY_ERROR = os.path.join(TEST_ASSET_DIR, "metadata_query_error.json")
EXPECTED_PAGED_DICT = os.path.join(TEST_ASSET_DIR, "metadata_query_expected_dict.dict")
METADATA_QUERY_SUCCESS = TEST_ASSET_DIR / "metadata_query_success.json"
METADATA_QUERY_ERROR = TEST_ASSET_DIR / "metadata_query_error.json"
EXPECTED_PAGED_DICT = TEST_ASSET_DIR / "metadata_query_expected_dict.dict"

METADATA_PAGE_1 = os.path.join(TEST_ASSET_DIR, "metadata_paged_1.json")
METADATA_PAGE_2 = os.path.join(TEST_ASSET_DIR, "metadata_paged_2.json")
METADATA_PAGE_3 = os.path.join(TEST_ASSET_DIR, "metadata_paged_3.json")
METADATA_PAGE_1 = TEST_ASSET_DIR / "metadata_paged_1.json"
METADATA_PAGE_2 = TEST_ASSET_DIR / "metadata_paged_2.json"
METADATA_PAGE_3 = TEST_ASSET_DIR / "metadata_paged_3.json"

EXPECTED_DICT = {
"publishedDatasources": [
Expand All @@ -29,74 +30,81 @@
EXPECTED_DICT_ERROR = [{"message": "Reached time limit of PT5S for query execution.", "path": None, "extensions": None}]


class MetadataTests(unittest.TestCase):
def setUp(self):
self.server = TSC.Server("http://test", False)
self.baseurl = self.server.metadata.baseurl
self.server.version = "3.5"

self.server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
self.server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM"

def test_metadata_query(self):
with open(METADATA_QUERY_SUCCESS, "rb") as f:
response_json = json.loads(f.read().decode())
with requests_mock.mock() as m:
m.post(self.baseurl, json=response_json)
actual = self.server.metadata.query("fake query")

datasources = actual["data"]

self.assertDictEqual(EXPECTED_DICT, datasources)

def test_paged_metadata_query(self):
with open(EXPECTED_PAGED_DICT, "rb") as f:
expected = eval(f.read())

# prepare the 3 pages of results
with open(METADATA_PAGE_1, "rb") as f:
result_1 = f.read().decode()
with open(METADATA_PAGE_2, "rb") as f:
result_2 = f.read().decode()
with open(METADATA_PAGE_3, "rb") as f:
result_3 = f.read().decode()

with requests_mock.mock() as m:
m.post(
self.baseurl,
[
{"text": result_1, "status_code": 200},
{"text": result_2, "status_code": 200},
{"text": result_3, "status_code": 200},
],
)

# validation checks for endCursor and hasNextPage,
# but the query text doesn't matter for the test
actual = self.server.metadata.paginated_query(
"fake query endCursor hasNextPage", variables={"first": 1, "afterToken": None}
)

self.assertDictEqual(expected, actual)

def test_metadata_query_ignore_error(self):
with open(METADATA_QUERY_ERROR, "rb") as f:
response_json = json.loads(f.read().decode())
with requests_mock.mock() as m:
m.post(self.baseurl, json=response_json)
actual = self.server.metadata.query("fake query")
datasources = actual["data"]

self.assertNotEqual(actual.get("errors", None), None)
self.assertListEqual(EXPECTED_DICT_ERROR, actual["errors"])
self.assertDictEqual(EXPECTED_DICT, datasources)

def test_metadata_query_abort_on_error(self):
with open(METADATA_QUERY_ERROR, "rb") as f:
response_json = json.loads(f.read().decode())
with requests_mock.mock() as m:
m.post(self.baseurl, json=response_json)

with self.assertRaises(GraphQLError) as e:
self.server.metadata.query("fake query", abort_on_error=True)
self.assertListEqual(e.error, EXPECTED_DICT_ERROR)
@pytest.fixture(scope="function")
def server():
"""Fixture to create a TSC.Server instance for testing."""
server = TSC.Server("http://test", False)

# Fake signin
server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM"
server.version = "3.5"

return server


def test_metadata_query(server: TSC.Server) -> None:
with open(METADATA_QUERY_SUCCESS, "rb") as f:
response_json = json.loads(f.read().decode())
with requests_mock.mock() as m:
m.post(server.metadata.baseurl, json=response_json)
actual = server.metadata.query("fake query")

datasources = actual["data"]

assert EXPECTED_DICT == datasources


def test_paged_metadata_query(server: TSC.Server) -> None:
with open(EXPECTED_PAGED_DICT, "rb") as f:
expected = eval(f.read())

# prepare the 3 pages of results
with open(METADATA_PAGE_1, "rb") as f:
result_1 = f.read().decode()
with open(METADATA_PAGE_2, "rb") as f:
result_2 = f.read().decode()
with open(METADATA_PAGE_3, "rb") as f:
result_3 = f.read().decode()

with requests_mock.mock() as m:
m.post(
server.metadata.baseurl,
[
{"text": result_1, "status_code": 200},
{"text": result_2, "status_code": 200},
{"text": result_3, "status_code": 200},
],
)

# validation checks for endCursor and hasNextPage,
# but the query text doesn't matter for the test
actual = server.metadata.paginated_query(
"fake query endCursor hasNextPage", variables={"first": 1, "afterToken": None}
)

assert expected == actual


def test_metadata_query_ignore_error(server: TSC.Server) -> None:
with open(METADATA_QUERY_ERROR, "rb") as f:
response_json = json.loads(f.read().decode())
with requests_mock.mock() as m:
m.post(server.metadata.baseurl, json=response_json)
actual = server.metadata.query("fake query")
datasources = actual["data"]

assert actual.get("errors", None) is not None
assert EXPECTED_DICT_ERROR == actual["errors"]
assert EXPECTED_DICT == datasources


def test_metadata_query_abort_on_error(server: TSC.Server) -> None:
with open(METADATA_QUERY_ERROR, "rb") as f:
response_json = json.loads(f.read().decode())
with requests_mock.mock() as m:
m.post(server.metadata.baseurl, json=response_json)

with pytest.raises(GraphQLError) as e:
server.metadata.query("fake query", abort_on_error=True)
assert e.error == EXPECTED_DICT_ERROR # type: ignore[attr-defined]
Loading