-
Notifications
You must be signed in to change notification settings - Fork 0
/
tests.py
75 lines (61 loc) · 2.57 KB
/
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
from emr_importer import Importer, open_config
from db_client import Client
import json
import os
import unittest
class BigqueryTest(unittest.TestCase):
def setUp(self):
with open("test_config.json") as f:
self.config = json.loads(f.read())
self.importer = Importer(database_type="bigquery", credentials={
"gcloud_credentials": self.config["gcloud_credentials"],
"gcloud_project": self.config["gcloud_project"]
})
def test_config_file_exists(self):
self.assertTrue(os.path.exists('config.json') or os.path.exists('configs/config.json'))
def test_open_config(self):
open_config()
class TestDataTranformation(BigqueryTest):
'''Given data in the form of a database response, test that it transforms into the expected format'''
# to test: data in valid format? data in readable json after run() completed?
def test_label_extractor_test(self):
importer = Importer(database_type="bigquery", label_extractor="label_extractor", credentials={
"gcloud_credentials": self.config["gcloud_credentials"],
"gcloud_project": self.config["gcloud_project"]
})
class TestRun(BigqueryTest):
def test_run(self):
db = self.config["databases"][0]
table = self.config["tables"][0]
query = f'SELECT * FROM {db}.{table} LIMIT 1;'
print(f"query: '{query}'")
self.importer.run(query)
class TestClientInstantiation(unittest.TestCase):
def test_csv(self):
with self.assertRaises(Exception):
client = Client('csv')
client = Client('csv', filename='test.csv')
def test_sqlite(self):
with self.assertRaises(Exception):
client = Client('sqlite')
client = Client('sqlite', filename='test.sqlite')
def test_dummy(self):
client = Client('dummy')
def test_nonexistant(self):
with self.assertRaises(Exception):
client = Client("dne")
class TestClientQuery(unittest.TestCase):
def setUp(self):
self.dummy_client = Client('dummy', filename='test.csv')
self.csv_client = Client('csv', filename='test.csv')
self.sqlite_client = Client('sqlite', filename='test.sqlite')
def test_csv(self):
self.csv_client.query()
self.csv_client.query("SELECT * FROM test LIMIT 1;")
def test_sqlite(self):
self.sqlite_client.query("SELECT * FROM test LIMIT 1;")
def test_dummy(self):
self.dummy_client.query()
self.dummy_client.query("SELECT * FROM test LIMIT 1;")
if __name__=="__main__":
unittest.main()