sql_to_schema_test.py
1.81 KB
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
import pandas as pd
import numpy as np
import unittest
import sql_to_schema as sts
class TestStuff(unittest.TestCase):
def test_filter_whitespace(self):
sql = "SELECT rc.dateCooked, r.name FROM recipeCooked rc INNER JOIN recipe r ON r.recipeID = rc.recipeID"
import sqlparse
tks = sqlparse.parse(sql)[0].tokens
tksf = sts.filter_whitespace(tks)
self.assertEqual(tksf[0].value, 'SELECT')
self.assertEqual(tksf[2].value, 'FROM')
#TODO - fix this to handle tables with or without aliases
# def test_extract_table(self):
# sql = "SELECT rc.dateCooked FROM recipeCooked"
# tables = sts.extract_tables(sql)
# self.assertEqual(len(tables), 1)
# self.assertEqual(tables[0]['name'], 'recipeCooked')
def test_extract_tables(self):
sql = "SELECT rc.dateCooked, r.name FROM recipeCooked rc INNER JOIN recipe r ON r.recipeID = rc.recipeID"
tables = sts.extract_tables(sql)
self.assertEqual(tables[0]['name'], 'recipeCooked')
self.assertEqual(tables[1]['name'], 'recipe')
# def test_extract_schema(self):
# sql = "SELECT rc.dateCooked, r.name FROM recipeCooked rc INNER JOIN recipe r ON r.recipeID = rc.recipeID"
# scm = sts.extract_schema(sql)
# self.assertEqual(scm,None)
if __name__ == '__main__':
# tests_to_run = [
# "test_sqlite"
# # ,"try_a_mock"
# # ,"test_sqalch_insert_read"
# # ,"test_shitty_hang"
# ]
# suite = unittest.TestSuite()
# for test in tests_to_run:
# suite.addTest(TestStuff(test))
# runner = unittest.TextTestRunner()
# runner.run(suite)
unittest.main()