cdplib/tests/testSQLOperations.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Wed Sep 19 14:34:22 2018

@author: tanya
"""

import os
import unittest
import pandas as pd

from libraries.database_operations_library import SQLOperations


class TestSQLOperations(unittest.TestCase):
    '''
    '''
    def __init__(self, test_df = None):
        print('\n', '='*5, 'Testing class : SQLOperations', '='*5)
        self.inst = SQLOperations(db_url = None)
        print('Connected to', self.inst.db_url)
        
        if test_df is None:
            self.test_df = pd.DataFrame({'a' : [1,2,3,4,5], 'b' : ['A', 'B', 'C', 'A', 'V'], 'c' : [0.1, 0.2, 0.3, 0.4, 0.5]})
        else:
            self.test_df = test_df

        
    def _create_test_table(self, test_tablename, create_table_query = None):
        '''
        '''
        self.inst.drop_table_if_exists(test_tablename)        
        
        if create_table_query is None:
            if 'ibm_db' in self.inst.db_url:
                create_table_query = """CREATE TABLE {} (
                                        a INT,
                                        b CHAR,
                                        c DECIMAL(10 , 2 )
                                        );""".format(test_tablename)
            else:
                create_table_query = """CREATE TABLE test (
                                        a INT,
                                        b TEXT,
                                        c DECIMAL(10 , 2 )
                                        );"""
                
        self.inst.execute(create_table_query)


        
class TestExecute(TestSQLOperations):
    '''
    '''
    def __init__(self):
        super(TestExecute, self).__init__()
        print('\n', '-'*2, 'Testing method : execute')
        
    def test_create_table(self, test_tablename, create_table_query = None):
        '''
        '''
        print('-'*4, 'Testing create table operation')
        self._create_test_table(test_tablename = test_tablename, create_table_query = create_table_query)
        self.assertTrue(self.inst.check_if_table_exists(test_tablename))
        self.inst.drop_table_if_exists(test_tablename)
        print('-'*4, 'Test ran successfully!')
        
class TestLoad_csv_to_db(TestSQLOperations):
    '''
    '''
    def __init__(self):
        super(TestLoad_csv_to_db, self).__init__()
        print('\n', '-'*2, 'Testing method : load_csv_to_db')
    
    def test_correct_content(self, test_csv_path, test_tablename, create_table_query = None):
        '''
        '''
        print('-'*4, 'Testig that load operation gives correct result')
        os.makedirs(os.path.dirname(test_csv_path), exist_ok = True)
        if not self.inst.drop_table_if_exists(test_tablename):
            self._create_test_table(test_tablename)
    
        self.test_df.to_csv(test_csv_path, index = False)
        self.inst.load_csv_to_db(csv_path = test_csv_path, tablename = test_tablename)
        try:
            connection = self.inst.engine.connect()
            test_df_from_sql = pd.read_sql(sql = "SELECT * FROM test", con = connection)
            connection.close()
        except Exception as e:
            raise Exception('ERROR: test csv file has not been load to sql at all, \n, exit with {}'.format(e))
        
        print('-'*4, 'Testing data has correct shape')
        self.assertTupleEqual(self.test_df.shape, test_df_from_sql.shape)
        
        print('-'*4,'Testing data has correct columns')
        self.assertSetEqual(set(self.test_df.columns), set(test_df_from_sql.columns))
        
        print('-'*4,'Testing data has correct content')
        for col in self.test_df.columns:
            test_df_from_sql[col] = test_df_from_sql[col].astype(self.test_df[col].dtype)
        pd.testing.assert_frame_equal(self.test_df, test_df_from_sql)
        
        print('-'*4, 'Test ran successfully!')

                        
if __name__ == '__main__':
    
    test_tablename = 'test10'
    test_csv_path = '/home/tanya/acdp/data_samples/test.csv'
    
    TestExecute().test_create_table(test_tablename = test_tablename)
    TestLoad_csv_to_db().test_correct_content(test_csv_path = test_csv_path, test_tablename = test_tablename)
    
    print('Done!', '\n')