test_ExportToDatabase.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Wed Jul 14 13:45:28 2021

@author: geopeter

execute test from root
"""

import unittest
import pickle
import dataacquisition.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"

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__")
    
    def testCreateInsertStatement(self):
        print("__Test CreateInsertStatement")
        with open("./dataacquisition/pickle/stationList_with_temperature.pickle", "rb") as pickleFile:
            stationList = pickle.load(pickleFile)
            stationList = stationList.loc[stationList['country']=="Germany"]
            stationList["station_id"] = stationList.index
            
            insert_stmt = cut.createInsertStatement(stationList)
            
            
            print(insert_stmt)
            
            self.assertTrue(insert_stmt.startswith("INSERT INTO"))
            self.assertTrue(insert_stmt.endswith("%s,%s)"))
        

if __name__ == '__main__':
    unittest.main()