UNITTESTS Add the sub package for unittests

rivus/tests/test_db.py

@@ -0,0 +1,96 @@
+import unittest
+import pyomo.environ  # although is not used direktly, is needee by pyomo
+from pyomo.opt.base import SolverFactory
+from rivus.main.rivus import read_excel, create_model
+from rivus.main.rivus import get_timeseries, get_constants
+from rivus.utils.prerun import setup_solver
+from rivus.gridder.create_grid import create_square_grid as square_grid
+from rivus.gridder.extend_grid import vert_init_commodities
+from rivus.gridder.extend_grid import extend_edge_data
+from rivus.io import db as rdb
+from sqlalchemy import create_engine
+import json
+import os
+pdir = os.path.dirname
+
+
+class RivusDBTest(unittest.TestCase):
+
+    def test_df_insert_query(self):
+        """Are the stored dataframes and the retrieved ones identical?
+
+        - Comparison form of frames is *after* create_model. (index is set)
+        - Comparison form expects that input dataframes only have meaningful
+          columns. (See pull request #23)
+        - Only implemented dataframes are tested.
+
+        Note
+        ----
+        Requires a ``config.json`` file in the root of rivus-repo with the
+        database credentials. For Example:
+        ::
+
+            {
+                "db" : {
+                    "user" : "postgres",
+                    "pass" : "postgres",
+                    "host" : "localhost",
+                    "base" : "rivus"
+                }
+            }
+        """
+        conf_path = os.path.join(pdir(pdir(pdir(__file__))), 'config.json')
+        config = []
+        with open(conf_path) as conf:
+            config = json.load(conf)
+        # DB connection
+        _user = config['db']['user']
+        _pass = config['db']['pass']
+        _host = config['db']['host']
+        _base = config['db']['base']
+        engine_string = ('postgresql://{}:{}@{}/{}'
+                         .format(_user, _pass, _host, _base))
+        engine = create_engine(engine_string)
+
+        proj_name = 'mnl'
+        base_directory = os.path.join('data', proj_name)
+        data_spreadsheet = os.path.join(base_directory, 'data.xlsx')
+        data = read_excel(data_spreadsheet)
+        # data_bup = data.copy()
+        vertex, edge = square_grid()
+        vert_init_commodities(vertex, ['Elec', 'Gas'], [('Elec', 0, 100000)])
+        extend_edge_data(edge)
+        prob = create_model(data, vertex, edge)
+        solver = SolverFactory(config['solver'])
+        solver = setup_solver(solver, log_to_console=False)
+        solver.solve(prob, tee=True)
+
+        test_id = rdb.init_run(engine, runner='Unittest')
+        rdb.store(engine, prob, run_id=test_id)
+
+        this_df = None
+        dfs = data.keys()
+        for df in dfs:
+            if df == 'hub':
+                continue  # is not implemented yet
+            this_df = data[df]
+            print(df)
+            re_df = rdb.df_from_table(engine, df, test_id)
+            self.assertTrue(all(this_df.fillna(0) ==
+                                re_df.reindex(this_df.index).fillna(0)),
+                            msg=('{}: Original and retrieved frames'
+                                 ' are not identical'.format(df)))
+        # Add implemented result dataframes
+        cost, pmax, kappa_hub, kappa_process = get_constants(prob)
+        source, _, _, _, _, = get_timeseries(prob)
+        results = dict(source=source, cost=cost, pmax=pmax,
+                       kappa_hub=kappa_hub, kappa_process=kappa_process)
+        dfs = ['source', 'cost', 'pmax', 'kappa_hub', 'kappa_process']
+        for df in dfs:
+            this_df = results[df]
+            print(df)
+            re_df = rdb.df_from_table(engine, df, test_id)
+            self.assertTrue(all(this_df.fillna(0) ==
+                                re_df.reindex(this_df.index).fillna(0)),
+                            msg=('{}: Original and retrieved frames'
+                                 ' are not identical'.format(df)))

rivus/tests/test_gridder.py

@@ -0,0 +1,17 @@
+import unittest
+import os
+from rivus.gridder.create_grid import create_square_grid
+from rivus.gridder.create_grid import get_source_candidates
+from rivus.main.rivus import read_excel
+
+
+class RivusGridTest(unittest.TestCase):
+
+    # TODO There is plenty of more functions (if not all)
+    # which reside in the code-base untested. Bellow some placeholders.
+
+    def test_source_calculation(self):
+        pass
+
+    def test_create_sq_grid(self):
+        pass

rivus/tests/test_main.py

@@ -0,0 +1,23 @@
+import unittest
+from rivus.main.rivus import read_excel
+
+
+class RivusMainTest(unittest.TestCase):
+
+    # TODO There is plenty of more functions (if not all)
+    # which reside in the code-base untested. Bellow some placeholders.
+
+    def test_line_length(self):
+        pass
+
+    def test_source_calculation(self):
+        pass
+
+    def test_pair_vertex_to_edge(self):
+        pass
+
+    def test_save_load(self):
+        pass
+
+    def test_read_excel(self):
+        pass

rivus/tests/test_utils.py

@@ -0,0 +1,97 @@
+import unittest
+import os
+pdir = os.path.dirname
+from rivus.utils.notify import email_me
+from rivus.utils.runmany import parameter_range
+from rivus.main.rivus import read_excel
+import json
+
+
+class RivusUtilsTest(unittest.TestCase):
+
+    def test_parameter_range(self):
+        """Load minimal example data and test with one of the parameters
+
+        + Tests for side-effects on the original.
+        + Tests for the awaited numer of parameters.
+        + Tests for range of the parameters
+        """
+        proj_name = 'mnl'
+        base_directory = os.path.join('data', proj_name)
+        data_spreadsheet = os.path.join(base_directory, 'data.xlsx')
+        data = read_excel(data_spreadsheet)
+        data_bup = data.copy()
+        which_sheet = 'commodity'
+        selected_df = data[which_sheet]
+        index = 'Heat',
+        column = 'cost-inv-fix'
+        lims = dict(lim_lo=0.5, lim_up=1.5, step=0.25)
+        the_param = selected_df.loc[index][column]
+        awaited_number = (lims['lim_up'] - lims['lim_lo']) // lims['step']
+        if the_param == 0:
+            awaited_number = 0
+        got_parameters = []
+        for df in parameter_range(selected_df, index, column, **lims):
+            got_parameters.append(df.loc[index][column])
+
+        self.assertTrue(len(got_parameters) == awaited_number,
+                        msg='got {} instead of awaited {}'
+                            .format(len(got_parameters), awaited_number))
+
+        self.assertTrue(all(data_bup[which_sheet].loc[index].fillna(0) ==
+                            data[which_sheet].loc[index].fillna(0)),
+                        msg='Func changed original row.')
+
+        if the_param != 0:
+            self.assertTrue(max(got_parameters) < lims['lim_up'] * the_param,
+                            msg='Got parameter bigger than awaited.')
+
+            self.assertTrue(min(got_parameters) >= lims['lim_lo'] * the_param,
+                            msg='Got parameter smaller than awaited.')
+
+    def test_email_notification(self):
+        """It only can test, whether the notification function run trhrough
+        successfully.
+
+        Can be useful for quick testing the email parameters in the config file
+
+        Note
+        -----
+        Requires a ``config.json`` file in the root of rivus-repo with the
+        database credentials. For Example:
+        ::
+
+            {
+                "email" : {
+                    "s_user" : "[email protected]",
+                    "s_pass" : "TheAnswerIs42!",
+                    "r_user" : "[email protected]",
+                    "smpt_addr" : "smtp.gmail.com",
+                    "smpt_port" : "587"
+                }
+            }
+        """
+        # Concatenate the absolute path to the config file.
+        # conf_path = __file__[:-len('rivus/tests/utils.py')] + 'config.json'
+        conf_path = os.path.join(pdir(pdir(pdir(__file__))), 'config.json')
+        config = []
+        with open(conf_path) as conf:
+            config = json.load(conf)
+        email_setup = {
+            'sender': config['email']['s_user'],
+            'send_pass': config['email']['s_pass'],
+            'recipient': config['email']['r_user'],
+            'smtp_addr': config['email']['smtp_addr'],
+            'smtp_port': config['email']['smtp_port']
+        }
+
+        sub = 'Testing from unittest [rivus][test]'
+        msg = ('rivus is a mixed integer linear programming model '
+               'for multi-commodity energy infrastructure networks systems '
+               'with a focus on high spatial resolution.\n'
+               'It finds the minimum cost energy infrastructure networks to '
+               'satisfy a given energy distribution for '
+               'possibly multiple commodities '
+               '(e.g. electricity, heating, cooling, ...).')
+        self.assertEqual(email_me(msg, subject=sub, **email_setup), 0,
+                         msg='Something went wrong during email notification.')