Newer
Older
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Wed Jul 14 13:45:28 2021
@author: geopeter
"""
import unittest
import pickle
import ExportToDatabase as cut
import configparser
import psycopg2
import time
cfg = configparser.ConfigParser()
cfg.read('config.ini')
assert "POSTGRES" in cfg, "missing POSTGRES in config.ini"
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
param_postgres = cfg["POSTGRES"]
stationList = None
testDB = 'weathertestdb'
class TestExportToDatabase(unittest.TestCase):
def testDBExists(self):
print("__Test DB Exists")
#given
cut.drop_db(testDB)
#test
exists = cut.dbexists(testDB)
self.assertFalse(exists)
#finished
print("Test OKAY__")
def testCreateDB(self):
print("__Test CreateDB")
try:
# given
cut.drop_db(testDB)
# test
cut.create_db(testDB)
self.assertTrue(cut.dbexists(testDB))
# finished
cut.drop_db(testDB)
print("Test OKAY__")
except(Exception, psycopg2.DatabaseError) as error:
self.assertRaises(error)
print("Test FAILED__")
def testCreateTable(self):
print("__Test CreateTable")
try:
# given
cut.drop_db(testDB)
cut.create_db(testDB)
# test
with open("./pickle/stationList_with_temperature.pickle", "rb") as pickleFile:
stationList = pickle.load(pickleFile)
stationList = stationList.loc[stationList['country']=="Germany"]
cut.create_table(stationList, testDB)
connection = psycopg2.connect(dbname=testDB, user=param_postgres["user"], password=param_postgres["password"], host=param_postgres["host"], port=param_postgres["port"])
cursor = connection.cursor()
columns = cursor.execute("SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name = 'stations';")
self.assertTrue(cursor.fetchall()[0][0]>=50)
# finished
connection.close()
cut.drop_db(testDB)
print("Test OKAY__")
except(Exception, psycopg2.DatabaseError) as error:
connection.close()
self.assertRaises(error.message)
print("Test FAILED__")
if __name__ == '__main__':
unittest.main()