[#47] Unittests Part1

Author: pavijayakuma

tests/test_files/test.csv

@@ -0,0 +1,19 @@
+"Name",     "Sex", "Age", "Height (in)", "Weight (lbs)"
+"Alex",       "M",   41,       74,      170
+"Bert",       "M",   42,       68,      166
+"Carl",       "M",   32,       70,      155
+"Dave",       "M",   39,       72,      167
+"Elly",       "F",   30,       66,      124
+"Fran",       "F",   33,       66,      115
+"Gwen",       "F",   26,       64,      121
+"Hank",       "M",   30,       71,      158
+"Ivan",       "M",   53,       72,      175
+"Jake",       "M",   32,       69,      143
+"Kate",       "F",   47,       69,      139
+"Luke",       "M",   34,       72,      163
+"Myra",       "F",   23,       62,       98
+"Neil",       "M",   36,       75,      160
+"Omar",       "M",   38,       70,      145
+"Page",       "F",   31,       67,      135
+"Quin",       "M",   29,       71,      176
+"Ruth",       "F",   28,       65,      131

tests/test_ppsql_hiveconnection.py

@@ -0,0 +1,12 @@
+from ppextensions.ppsql.connection.hiveconnection import HiveConnection
+from thriftpy.transport import TTransportException
+import unittest
+
+class TestPPSqlHiveConnection(unittest.TestCase):
+    
+    def test_invalid_hive_connection(self):
+        with self.assertRaises(TTransportException):
+            hc = HiveConnection("fake", "fake", 1234, "GSSAPI", "fake")
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_pputils_configuration.py

@@ -0,0 +1,18 @@
+from ppextensions.pputils.utils.configuration import *
+import unittest
+
+class TestPPUtilsConfiguration(unittest.TestCase):
+    
+    def setUp(self):
+        pass
+        
+    def test_load_config(self):
+        conf = load_conf(PATH)
+        self.assertTrue(isinstance(conf, dict))
+
+    def test_conf_info(self):
+        config_info = conf_info("test-engine")
+        self.assertTrue(isinstance(config_info, dict))
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_pputils_csvconnection.py

@@ -0,0 +1,13 @@
+from ppextensions.ppsql.connection.csvconnection import CSVConnection
+import unittest
+
+class TestPPSqlCSVConnection(unittest.TestCase):
+    
+    def setUp(self):
+        self.csvconn = CSVConnection()
+
+    def test_csv_query(self):
+        self.assertTrue(self.csvconn.execute("select * from test_files/test.csv") != None)
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_pputils_fsreaderwriter.py

@@ -0,0 +1,18 @@
+from ppextensions.pputils.utils.filesystemreader import FileSystemReaderWriter
+import unittest
+
+class TestPPUtilsFileSystemReaderWriter(unittest.TestCase):
+    
+    def setUp(self):
+        # In current implementation we will have two RM Objects
+        # One with invalid RM Url & One with Valid one 
+        self.fsreadwriter = FileSystemReaderWriter("~")
+        
+    def test_ensure_path_exists(self):
+        self.assertTrue(self.fsreadwriter.ensure_path_exists() == None)
+
+    def test_ensure_file_exists(self):
+        self.assertTrue(self.fsreadwriter.ensure_file_exists() == None)
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_pputils_log.py

@@ -0,0 +1,22 @@
+from ppextensions.pputils.utils.log import Log
+import unittest
+
+class TestPPUtilsLog(unittest.TestCase):
+    
+    def setUp(self):
+        self.logger = Log("test_logger")
+        
+    def test_log_debug(self):
+        self.assertTrue(self.logger.debug("test_debug") == None)
+
+    def test_log_error(self):
+        self.assertTrue(self.logger.error("test_error") == None)
+
+    def test_log_info(self):
+        self.assertTrue(self.logger.info("test_info") == None)
+
+    def test_log_exception(self):
+        self.assertTrue(self.logger.exception("test_exception") == None)
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_pputils_parameterargs.py

@@ -0,0 +1,29 @@
+from ppextensions.pputils.utils.parameterargs import *
+from IPython.core.magic_arguments import argument, magic_arguments, parse_argstring
+import pandas as pd
+import unittest
+
+class TestPPUtilsParamArgs(unittest.TestCase):
+    
+    @magic_arguments()
+    @argument("-test", "--test", type=str, help="test1")
+    @argument("-test2", "--test2", type=str, help="test2")
+    def setUp(self):
+        args = "--test test --test2 testa:::testb:::testc"
+        self.param_args = ParameterArgs(parse_argstring(self.setUp, args))
+
+    def test_get_args(self):
+        self.assertTrue(self.param_args.get("test") == "test")
+        self.assertTrue(self.param_args.get("test2") == "testa:::testb:::testc")
+        
+    def test_has_args(self):
+        self.assertTrue(self.param_args.hasattr("test"))
+        self.assertTrue(self.param_args.hasattr("test2"))
+        
+    def test_get_list(self):
+        return_vals = self.param_args.get_list("test2")
+        self.assertTrue(isinstance(return_vals, list))
+        self.assertTrue(len(return_vals) == 3)
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_pputils_resultset.py

@@ -0,0 +1,35 @@
+from ppextensions.pputils.utils.resultset import ResultSet, unduplicate_field_names, _nonbreaking_spaces
+import unittest
+import requests
+import pandas as pd
+import matplotlib.pylab as plt
+
+class TestPPUtilsResultSet(unittest.TestCase):
+    
+    def setUp(self):
+        # In current implementation we will have two RM Objects
+        # One with invalid RM Url & One with Valid one 
+        data = {'col1': [1,2,3,4], 'col2': [3,4,5,6], 'col3': [4,5,6,7], 'col4': [5,6,7,8]}
+        cols = ['col1', 'col2', 'col3', 'col4']
+        self.resultset = ResultSet(cols, data)
+        
+    def test_resultset_df(self):
+        df = self.resultset.DataFrame()
+        self.assertTrue(isinstance(df, pd.DataFrame))
+
+    def test_dict(self):
+        dictionary = self.resultset.dict()
+        self.assertTrue(isinstance(dictionary, dict))
+        
+    def test_csv(self):
+        self.assertTrue(self.resultset.csv())
+        
+    def test_unduplicate_field_names(self):
+        cols = ['col1', 'col2', 'col3', 'col4', 'col1']
+        expected_cols = ['col1', 'col2', 'col3', 'col4']
+        updated_cols = unduplicate_field_names(cols)
+        self.assertTrue(isinstance(updated_cols, list))
+        
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_pputils_utils.py

@@ -0,0 +1,33 @@
+from ppextensions.pputils.utils import utils
+import pandas as pd
+import unittest
+
+class TestPPUtilsUtils(unittest.TestCase):
+    
+    def setUp(self):
+        self.test_df = utils.csv_to_df({}, {"csv":"test_files/test.csv"})
+
+    def test_renew_kerberos_ticket_negative(self):
+        self.assertFalse(utils.renew_kerberos_ticket("fake","fake"))
+
+    def test_renew_kerberos_ticket_positive(self):
+        self.assertTrue(utils.renew_kerberos_ticket("pp_batch_opts_admin","~/PPExtensions/tests/test_files/.pp_batch_opts_admin.keytab"))
+        
+    def test_parse_run_str(self):
+        self.assertTrue(utils.parse_run_str("%%info"))
+        self.assertTrue(utils.parse_run_str("test_call_value=123"))
+
+    def test_csv_to_df_and_back(self):
+        df = utils.csv_to_df({}, {"csv":"test_files/test.csv"})
+        self.assertTrue(isinstance(df, pd.DataFrame))
+        args = {"dataframe":"df"}
+        csv_file = utils.df_to_csv({"df":df}, args)
+        self.assertTrue(csv_file != None)
+
+    def test_substitute_params(self):
+        substitute_cell = utils.substitute_params("test_cell_value = {}", {"test":"test"})
+        self.assertTrue(substitute_cell != None)
+
+    
+if __name__ == '__main__':
+    unittest.main()

tests/test_pputils_yarnapi.py

@@ -0,0 +1,21 @@
+from ppextensions.pputils.utils.yarnapi import ResourceManager
+import unittest
+import requests
+
+class TestPPUtilsYarnAPI(unittest.TestCase):
+    
+    def setUp(self):
+        # In current implementation we will have two RM Objects
+        # One with invalid RM Url & One with Valid one 
+        self.invalid_rm = ResourceManager("http://invalidrm.paypalinc.com")
+        
+    def test_invalid_rm_cluster_app(self):
+        with self.assertRaises(requests.exceptions.ConnectionError):
+            self.invalid_rm.cluster_application("fake_app")
+
+    def test_invalid_rm_cluster_metrics(self):
+        with self.assertRaises(requests.exceptions.ConnectionError):
+            self.invalid_rm.cluster_metrics()
+
+if __name__ == '__main__':
+    unittest.main()