added some test for manual connectors and test for DsDbManager

dsdbmanager/dbobject.py

@@ -351,15 +351,15 @@ class DsDbManager(object):
     oracle['some data base']
     """
 
-    def __init__(self, flavor: str):
+    def __init__(self, flavor: str, config_file_manager: ConfigFilesManager = None):
         if flavor.lower() not in FLAVORS_FOR_CONFIG:
             raise NotImplementedFlavor(
                 f"Invalid flavor: expected one of {', '.join(FLAVORS_FOR_CONFIG)}, got {flavor}",
                 None
             )
 
         self._flavor = flavor
-        self._config_file_manager = ConfigFilesManager()
+        self._config_file_manager = ConfigFilesManager() if config_file_manager is None else config_file_manager
         self._host_dict = self._config_file_manager.get_hosts()
 
         if not self._host_dict:

test/test_connectors.py

@@ -0,0 +1,79 @@
+import unittest
+from dsdbmanager.mssql_ import Mssql
+from dsdbmanager.mysql_ import Mysql
+from dsdbmanager.oracle_ import Oracle
+from dsdbmanager.teradata_ import Teradata
+from dsdbmanager.exceptions_ import MissingFlavor, MissingDatabase, MissingPackage
+
+
+class TestConnectors(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.host_dict = {
+            'oracle': {
+                'database1': {
+                    'name': 'database1',
+                    'host': 'somehost',
+                    'port': 0000
+                }
+            },
+            'teradata': {
+                'database1': {
+                    'name': 'database1',
+                    'host': 'somehost',
+                }
+            },
+            'mssql': {
+                'database1': {
+                    'name': 'database1',
+                    'host': 'somehost',
+                    'port': 0000
+                }
+            },
+            'mysql': {
+                'database1': {
+                    'name': 'database1',
+                    'host': 'somehost',
+                    'port': 0000
+                }
+            }
+        }
+
+    @classmethod
+    def tearDownClass(cls):
+        pass
+
+    def test_connectors(self):
+        for name, connector in zip(
+                [
+                    'oracle',
+                    'mysql',
+                    'mssql',
+                    'teradata',
+                ],
+                [
+                    Oracle,
+                    Mysql,
+                    Mssql,
+                    Teradata,
+                ]):
+            with self.subTest(flavor=name):
+                # test with host improper host file. should raise MissingFlavor
+                with self.assertRaises(MissingFlavor):
+                    _ = connector('database1', {'host': 'dummy'})
+
+                # test for database that has not been added
+                with self.assertRaises(MissingDatabase):
+                    _ = connector('database2', self.host_dict)
+
+                # test for database added
+                obj = connector('database1', self.host_dict)
+                self.assertTrue(hasattr(obj, 'create_engine'))
+
+                # this is only useful in an environment where none of the extra packages are available
+                # with self.assertRaises(MissingPackage):
+                #     obj.create_engine()
+
+
+if __name__ == '__main__':
+    unittest.main()

test/test_dbobject.py

@@ -1,4 +1,8 @@
+import json
 import unittest
+import pathlib
+import tempfile
+import contextlib
 import pandas as pd
 import sqlalchemy as sa
 import sqlalchemy.exc as exc
@@ -8,14 +12,19 @@
     update_on_table,
     table_middleware,
     DbMiddleware,
+    DsDbManager,
     TableMeta,
     TableInsert,
     TableUpdate
 )
 from dsdbmanager.exceptions_ import (
     BadArgumentType,
-    NoSuchColumn
+    NoSuchColumn,
+    NotImplementedFlavor,
+    EmptyHostFile,
+    MissingFlavor
 )
+from dsdbmanager.configuring import ConfigFilesManager
 
 
 class TestDbObject(unittest.TestCase):
@@ -66,6 +75,34 @@ def setUpClass(cls):
 
         cls.engine_function = lambda _: sa.create_engine("sqlite:///")
 
+        cls.host = {
+            'oracle': {
+                'mydatabase': {
+                    'name': 'mydatabase',
+                    'host': 'localhost',
+                    'sid': 'pyt'
+                },
+                'newdatabase': {
+                    'name': 'newdatabase',
+                    'host': 'localhost',
+                    'sid': 'pyt',
+                    'schema': 'schemo'
+                }
+            }
+        }
+
+    @classmethod
+    @contextlib.contextmanager
+    def generate_path(cls, suffix=None) -> pathlib.Path:
+        try:
+            temp = tempfile.NamedTemporaryFile(suffix=suffix) if suffix else tempfile.NamedTemporaryFile()
+            yield pathlib.Path(temp.name)
+        except OSError as _:
+            temp = None
+        finally:
+            if temp is not None:
+                temp.close()
+
     @classmethod
     def tearDownClass(cls):
         pass
@@ -290,7 +327,63 @@ def test_dbmiddleware(self):
             self.assertIsInstance(dbm._insert, TableInsert)
             self.assertIsInstance(dbm._update, TableUpdate)
 
-    # TODO - test TableMeta, TableInsert, TableUpdate: check for keyerror etc
+    def test_dsdbmanager(self):
+        with self.assertRaises(NotImplementedFlavor):
+            _ = DsDbManager('somemadeupflavor')
+
+        with self.generate_path(suffix='.json') as hp, self.generate_path(
+                suffix='.json') as cp, self.generate_path() as kp:
+            c = ConfigFilesManager(
+                hp,
+                cp,
+                kp
+            )
+
+            # test empty host file
+            with hp.open('w') as f:
+                json.dump({}, f)
+
+            with self.assertRaises(EmptyHostFile):
+                _ = DsDbManager('oracle', c)
+
+            # test with actual host data
+            with hp.open('w') as f:
+                json.dump(self.host, f)
+
+            dbobject = DsDbManager('oracle', c)
+
+            # test for missing flavor
+            with self.assertRaises(MissingFlavor):
+                _ = DsDbManager('teradata', c)
+
+            # some properties must be there by default
+            for attribute in [
+                '_flavor',
+                '_config_file_manager',
+                '_host_dict',
+                '_available_databases',
+                '_schemas_per_databases',
+                '_connection_object_creator',
+            ]:
+                with self.subTest(attribute=attribute):
+                    self.assertTrue(hasattr(dbobject, attribute))
+
+            # properties based on host file
+            for host_database in self.host.get('oracle'):
+                with self.subTest(host_database=host_database):
+                    self.assertTrue(hasattr(dbobject, host_database))
+
+            # the host_dict should be the same dictionary as host
+            self.assertEqual(self.host, dbobject._host_dict)
+
+            # two available databases and two schemas
+            self.assertEqual(len(dbobject._available_databases), 2)
+            self.assertEqual(len(dbobject._schemas_per_databases), 2)
+            self.assertEqual(dbobject._available_databases, ['mydatabase', 'newdatabase'])
+            self.assertEqual(dbobject._schemas_per_databases, [None, 'schemo'])
+
+
+# TODO - test TableMeta, TableInsert, TableUpdate: check for keyerror etc
 
 
 if __name__ == '__main__':