Merge 71e89b1 into d7b6c4b

.travis.yml

@@ -1,5 +1,6 @@
 sudo: required
 services:
+  - mysql
   - docker
 before_install:
   - docker pull busybox:latest

tanner/tests/test_mysql_db_helper.py

@@ -0,0 +1,123 @@
+import unittest
+import asyncio
+import os
+from unittest import mock
+from tanner.utils.asyncmock import AsyncMock
+from tanner.utils.mysql_db_helper import MySQLDBHelper
+
+
+def mock_config(section, value):
+    if section == 'SQLI' and value == 'host':
+        return '127.0.0.1'
+    if section == 'SQLI' and value == 'user':
+        return 'root'
+    if section == 'SQLI' and value == 'password':
+        return ''
+
+
+class TestMySQLDBHelper(unittest.TestCase):
+    def setUp(self):
+        self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self.loop)
+        self.filename = '/tmp/db/test_db'
+        os.makedirs(os.path.dirname(self.filename), exist_ok=True)
+        open('/tmp/db/test_db', 'a').close()
+
+        self.db_name = 'test_db'
+        self.expected_result = None
+        self.returned_result = None
+        self.query_map = []
+        self.handler = MySQLDBHelper()
+        self.conn = None
+        self.cursor = None
+
+        with mock.patch('tanner.config.TannerConfig.get', side_effect=mock_config) as m:
+            async def connect():
+                self.conn = await self.handler.connect_to_db()
+                self.cursor = await self.conn.cursor()
+
+                # Delete DB if exists
+                self.returned_result = await self.handler.check_db_exists(self.db_name)
+                if self.returned_result == 1:
+                    await self.handler.delete_db(self.db_name)
+
+            self.loop.run_until_complete(connect())
+
+    @mock.patch('tanner.config.TannerConfig.get', side_effect=mock_config)
+    def test_check_db_exists(self, m):
+        self.expected_result = 1
+
+        async def test():
+            await self.cursor.execute('CREATE DATABASE test_db')
+            await self.conn.commit()
+            self.returned_result = await self.handler.check_db_exists(self.db_name)
+
+        self.loop.run_until_complete(test())
+        self.assertEqual(self.expected_result, self.returned_result)
+
+    @mock.patch('tanner.config.TannerConfig.get', side_effect=mock_config)
+    def test_check_no_db_exists(self, m):
+        self.expected_result = 0
+
+        async def test():
+            self.returned_result = await self.handler.check_db_exists(self.db_name)
+
+        self.loop.run_until_complete(test())
+        self.assertEqual(self.expected_result, self.returned_result)
+
+    @mock.patch('tanner.config.TannerConfig.get', side_effect=mock_config)
+    def test_setup_db_from_config(self, m):
+        config = {
+            "name": "test_db",
+            "tables": [
+                {
+                    "schema": "CREATE TABLE TEST (ID INTEGER PRIMARY KEY, USERNAME TEXT)",
+                    "table_name": "TEST",
+                    "data_tokens": "I,L"
+                }
+            ]
+        }
+
+        def mock_read_config():
+            return config
+
+        self.handler.read_config = mock_read_config
+        self.handler.insert_dummy_data = AsyncMock()
+
+        async def test():
+            await self.handler.setup_db_from_config()
+
+        self.loop.run_until_complete(test())
+        assert self.handler.insert_dummy_data.called
+
+    @mock.patch('tanner.config.TannerConfig.get', side_effect=mock_config)
+    def test_create_query_map(self, m):
+
+        self.expected_result_creds = {'COMMON': [{'name': 'NUM', 'type': 'INTEGER'}],
+                                      'CREDS': [{'name': 'ID', 'type': 'INTEGER'}, {'name': 'EMAIL', 'type': 'TEXT'},
+                                                {'name': 'PASSWORD', 'type': 'TEXT'}]}
+
+        self.expected_result_test = {'COMMON': [{'name': 'PARA', 'type': 'TEXT'}],
+                                     'TEST': [{'name': 'ID', 'type': 'INTEGER'},
+                                              {'name': 'USERNAME', 'type': 'TEXT'}]}
+
+        self.query = [
+            ['TEST_DB', "CREATE TABLE TEST (ID INTEGER PRIMARY KEY, USERNAME TEXT)", "CREATE TABLE COMMON (PARA TEXT)"],
+            ['CREDS_DB', "CREATE TABLE CREDS (ID INTEGER PRIMARY KEY, EMAIL VARCHAR(15), PASSWORD VARCHAR(15))",
+             'CREATE TABLE COMMON (NUM INTEGER )']
+        ]
+
+        async def test(data):
+            await self.cursor.execute('CREATE DATABASE {db_name}'.format(db_name=data[0]))
+            await self.cursor.execute('USE {db_name}'.format(db_name=data[0]))
+            await self.cursor.execute(data[1])
+            await self.cursor.execute(data[2])
+            result = await self.handler.create_query_map(data[0])
+            self.query_map.append(result)
+            await self.handler.delete_db(data[0])
+
+        for data in self.query:
+            self.loop.run_until_complete(test(data))
+
+        self.assertEqual(self.query_map[0], self.expected_result_test)
+        self.assertEqual(self.query_map[1], self.expected_result_creds)

tanner/utils/mysql_db_helper.py

@@ -18,7 +18,7 @@ async def connect_to_db(self):
                                       )
         return conn
 
-    async def check_db_exists(self, db_name, ):
+    async def check_db_exists(self, db_name):
         conn = await self.connect_to_db()
         cursor = await conn.cursor()
         check_DB_exists_query = 'SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA '
@@ -128,7 +128,7 @@ async def create_query_map(self, db_name):
                     await cursor.execute(query.format(table_name=table, db_name=db_name))
                     result = await cursor.fetchall()
                     for row in result:
-                        if (row[7] == 'int'):
+                        if row[7] == 'int':
                             columns.append(dict(name=row[3], type='INTEGER'))
                         else:
                             columns.append(dict(name=row[3], type='TEXT'))

tanner/utils/sqlite_db_helper.py

@@ -60,7 +60,7 @@ async def insert_dummy_data(self, table_name, data_tokens, cursor):
         cursor.executemany("INSERT INTO " + table_name + " VALUES(" +
                            inserted_string_patt + ")", inserted_data)
 
-    def create_query_map(self, working_dir, db_name, ):
+    def create_query_map(self, working_dir, db_name):
         query_map = {}
         tables = []