forked from wikimedia/pywikibot
-
Notifications
You must be signed in to change notification settings - Fork 3
/
superset_tests.py
executable file
·165 lines (132 loc) · 5.45 KB
/
superset_tests.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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
#!/usr/bin/env python3
"""Tests for superset module.
.. versionadded:: 9.2
"""
#
# (C) Pywikibot team, 2024
#
# Distributed under the terms of the MIT license.
#
from __future__ import annotations
import unittest
from contextlib import suppress
import pywikibot
from pywikibot.data.superset import SupersetQuery
from pywikibot.exceptions import NoUsernameError
from pywikibot.pagegenerators import SupersetPageGenerator
from tests.aspects import TestCase
class TestSupersetWithoutAuth(TestCase):
"""Test Superset without auth."""
family = 'meta'
code = 'meta'
def test_init(self):
"""Test init validation functions."""
# Test initial database_id parameter in wrong format
site = self.get_site()
superset = SupersetQuery(schema_name='fiwiki_p')
self.assertIsInstance(superset, SupersetQuery)
msg = 'Only one of schema_name and site parameters can be defined'
with self.assertRaisesRegex(TypeError, msg):
superset = SupersetQuery(schema_name='enwiki_p',
site=site)
msg = 'database_id should be integer'
with self.assertRaisesRegex(TypeError, msg):
superset = SupersetQuery(schema_name='enwiki_p',
database_id='foo')
class TestSupersetWithAuth(TestCase):
"""Test Superset with auth."""
login = True
family = 'meta'
code = 'meta'
def test_login_and_oauth_permisson(self):
"""Superset login and queries."""
sql = 'SELECT page_id, page_title FROM page LIMIT 2;'
site = self.get_site()
# Test login and initial site parameters
superset = SupersetQuery(site=site)
try:
superset.login()
except NoUsernameError:
self.skipTest('Oauth permission is missing.')
except ConnectionError as e:
self.skipTest(e)
self.assertTrue(site.logged_in())
self.assertTrue(superset.connected)
rows = superset.query(sql)
self.assertLength(rows, 2)
# Test initial schema_name parameter
superset = SupersetQuery(schema_name='fiwiki_p')
rows = superset.query(sql)
self.assertLength(rows, 2)
# Test initial schema_name and database_id parameters
superset = SupersetQuery(schema_name='enwiki_p', database_id=1)
rows = superset.query(sql)
self.assertLength(rows, 2)
# Test get_database_id_by_schema_name()
database_id = superset.get_database_id_by_schema_name('fiwiki_p')
self.assertEqual(database_id, 2)
# Test incorrect initial schema_name parameter
superset = SupersetQuery(schema_name='foowiki_p')
msg = 'Schema "foowiki_p" not found in https://superset.wmcloud.org.'
with self.assertRaisesRegex(KeyError, msg):
rows = superset.query(sql)
# Test incorrect initial database_id parameter
# superset.wmcloud.org fails with 500 server error
# so this is expected to be changed when server side
# is updated
superset = SupersetQuery(schema_name='enwiki_p', database_id=2)
with self.assertRaises(RuntimeError):
rows = superset.query(sql)
# Test overriding database_id in query
rows = superset.query(sql, database_id=1)
self.assertLength(rows, 2)
# Test overriding schema_name in query
rows = superset.query(sql, schema_name='fiwiki_p')
self.assertLength(rows, 2)
# Test overriding schema using site
testsite = pywikibot.Site('fi', 'wikipedia')
rows = superset.query(sql, site=testsite)
self.assertLength(rows, 2)
# Test that overriding both schema_name and site fails
msg = 'Only one of schema_name and site parameters can be defined'
with self.assertRaisesRegex(TypeError, msg):
rows = superset.query(sql, schema_name='fiwiki_p', site=site)
def test_superset_generators(self):
"""Superset generator."""
site = self.get_site()
query = 'SELECT page_id FROM page LIMIT 2'
gen = SupersetPageGenerator(query, site=site)
for page in gen:
t = str(page)
self.assertTrue(t)
query = 'SELECT page_title, page_namespace FROM page LIMIT 2'
gen = SupersetPageGenerator(query, site=site)
for page in gen:
t = str(page)
self.assertTrue(t)
query = 'SELECT page_namespace, page_title FROM page LIMIT 2'
gen = SupersetPageGenerator(query, schema_name='fiwiki_p')
for page in gen:
t = str(page)
self.assertTrue(t)
query = ('SELECT * FROM ('
'SELECT gil_wiki AS page_wikidb, gil_page AS page_id '
'FROM globalimagelinks GROUP BY gil_wiki '
') AS t LIMIT 2')
gen = SupersetPageGenerator(query, schema_name='commonswiki_p')
for page in gen:
t = str(page)
self.assertTrue(t)
query = ('SELECT * FROM ( '
'SELECT gil_wiki AS page_wikidb, '
'gil_page_namespace_id AS page_namespace, '
'gil_page_title AS page_title '
'FROM globalimagelinks GROUP BY gil_wiki '
') AS t LIMIT 2')
gen = SupersetPageGenerator(query, schema_name='commonswiki_p')
for page in gen:
t = str(page)
self.assertTrue(t)
if __name__ == '__main__':
with suppress(SystemExit):
unittest.main()