-
Notifications
You must be signed in to change notification settings - Fork 209
/
test_tableau_dashboard_extractor.py
77 lines (59 loc) · 2.97 KB
/
test_tableau_dashboard_extractor.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
# Copyright Contributors to the Amundsen project.
# SPDX-License-Identifier: Apache-2.0
import logging
import unittest
from typing import Any, Dict
from mock import patch
from pyhocon import ConfigFactory
from databuilder import Scoped
from databuilder.extractor.dashboard.tableau.tableau_dashboard_extractor import TableauDashboardExtractor
from databuilder.extractor.dashboard.tableau.tableau_dashboard_utils \
import TableauDashboardAuth, TableauGraphQLApiExtractor
logging.basicConfig(level=logging.INFO)
def mock_query(*_args: Any, **_kwargs: Any) -> Dict[str, Any]:
return {
'workbooks': [
{
'id': 'fake-id',
'name': 'Test Workbook',
'createdAt': '2020-04-08T05:32:01Z',
'description': '',
'projectName': 'Test Project',
'projectVizportalUrlId': 123,
'vizportalUrlId': 456
}
]
}
def mock_token(*_args: Any, **_kwargs: Any) -> str:
return '123-abc'
class TestTableauDashboardExtractor(unittest.TestCase):
@patch.object(TableauDashboardAuth, '_authenticate', mock_token)
@patch.object(TableauGraphQLApiExtractor, 'execute_query', mock_query)
def test_dashboard_metadata_extractor(self) -> None:
config = ConfigFactory.from_dict({
'extractor.tableau_dashboard_metadata.api_base_url': 'api_base_url',
'extractor.tableau_dashboard_metadata.tableau_base_url': 'tableau_base_url',
'extractor.tableau_dashboard_metadata.api_version': 'tableau_api_version',
'extractor.tableau_dashboard_metadata.site_name': 'tableau_site_name',
'extractor.tableau_dashboard_metadata.tableau_personal_access_token_name':
'tableau_personal_access_token_name',
'extractor.tableau_dashboard_metadata.tableau_personal_access_token_secret':
'tableau_personal_access_token_secret',
'extractor.tableau_dashboard_metadata.excluded_projects': [],
'extractor.tableau_dashboard_metadata.cluster': 'tableau_dashboard_cluster',
'extractor.tableau_dashboard_metadata.database': 'tableau_dashboard_database',
'extractor.tableau_dashboard_metadata.transformer.timestamp_str_to_epoch.timestamp_format':
'%Y-%m-%dT%H:%M:%SZ',
})
extractor = TableauDashboardExtractor()
extractor.init(Scoped.get_scoped_conf(conf=config, scope=extractor.get_scope()))
record = extractor.extract()
self.assertEqual(record.dashboard_id, 'Test Workbook')
self.assertEqual(record.dashboard_name, 'Test Workbook')
self.assertEqual(record.dashboard_group_id, 'Test Project')
self.assertEqual(record.dashboard_group, 'Test Project')
self.assertEqual(record.product, 'tableau')
self.assertEqual(record.cluster, 'tableau_dashboard_cluster')
self.assertEqual(record.created_timestamp, 1586323921)
if __name__ == '__main__':
unittest.main()