<reponame>oleksiilytvyn/grailkit
# -*- coding: UTF-8 -*-
"""
Tests for db module

:copyright: (c) 2017-2020 by <NAME> (http://alexlitvin.name).
:license: MIT, see LICENSE for more details.
"""
import unittest

import os
import shutil
import tempfile
import sqlite3 as sqlite

import grailkit.db as db


QUERY_CREATE = """CREATE TABLE `test` (`key` TEXT NOT NULL, `value` TEXT, PRIMARY KEY(key));"""
QUERY_INSERT = """INSERT INTO `test` VALUES('key', 'value')"""
QUERY_GET = """SELECT * FROM `test`"""
QUERY_MULTIPLE = """
    create table person(
        first_name,
        last_name,
        age
    );

    create table book(
        title,
        author,
        published
    );

    insert into book(title, author, published)
    values (
        'Dirk Gently''s Holistic Detective Agency',
        '<NAME>',
        1987
    );
    """
QUERY_INSERT_MULTIPLE = """
    INSERT INTO `test` ('key', 'value') VALUES
        ('one', 'first'),
        ('two', 'second'),
        ('three', 'third');
    """


class TestGrailkitDB(unittest.TestCase):

    def setUp(self):
        """Create a temporary directory"""

        self.test_dir = tempfile.mkdtemp()
        self.res_dir = os.path.abspath(__file__[:-3])

    def tearDown(self):
        """Remove the directory after the test"""

        shutil.rmtree(self.test_dir, ignore_errors=True)

    def test_db_create(self):
        """Test database create"""

        db_path = os.path.join(self.test_dir, 'test.sqlite')
        db_obj = db.DataBase(db_path, query=QUERY_CREATE, create=True)
        db_obj.close()

        self.assertTrue(os.path.isfile(db_path))

    def test_db_open(self):
        """Test database open"""

        db_path = os.path.join(self.res_dir, 'regular.sqlite')
        db_obj = db.DataBase(db_path)
        db_obj.close()

        self.assertTrue(True)

    def test_db_open_not_exists(self):
        """Test database open if file not exists"""

        db_path = os.path.join(self.res_dir, 'file_not_exists.sqlite')

        self.assertRaises(db.DataBaseError, db.DataBase, db_path)

    def test_db_open_corrupt(self):
        """Test database open if file corrupted"""

        db_path = os.path.join(self.res_dir, 'file_not_exists.sqlite')

        self.assertRaises(db.DataBaseError, db.DataBase, db_path)

    def test_db_connection(self):
        """Test connection to database"""

        db_path = os.path.join(self.res_dir, 'regular.sqlite')
        db_obj = db.DataBase(db_path)

        self.assertTrue(isinstance(db_obj.connection, sqlite.Connection))

        db_obj.close()

    def test_db_cursor(self):
        """Test cursor property"""

        db_path = os.path.join(self.res_dir, 'regular.sqlite')
        db_obj = db.DataBase(db_path)

        self.assertTrue(isinstance(db_obj.cursor, sqlite.Cursor))

        db_obj.close()

    def test_db_get(self):
        """Test getting single row"""

        db_path = os.path.join(self.test_dir, 'get.sqlite')
        db_obj = db.DataBase(db_path, create=True)

        db_obj.execute(QUERY_CREATE)
        db_obj.execute(QUERY_INSERT)
        res = db_obj.get(QUERY_GET)

        self.assertEqual(res[0], 'key')
        self.assertEqual(res[1], 'value')

        self.assertEqual(res['key'], 'key')
        self.assertEqual(res['value'], 'value')

        db_obj.close()

    def test_db_all(self):
        """Test getting results by query"""

        db_path = os.path.join(self.test_dir, 'get.sqlite')
        db_obj = db.DataBase(db_path, create=True)

        db_obj.execute(QUERY_CREATE)
        db_obj.execute(QUERY_INSERT_MULTIPLE)

        res = db_obj.all(QUERY_GET)

        self.assertEqual(res[0][0], 'one')
        self.assertEqual(res[0][1], 'first')

        self.assertEqual(res[1][0], 'two')
        self.assertEqual(res[1][1], 'second')

        self.assertEqual(res[2][0], 'three')
        self.assertEqual(res[2][1], 'third')

        db_obj.close()

    def test_db_execute(self):
        """Test query execution"""

        db_path = os.path.join(self.test_dir, 'execute.sqlite')
        db_obj = db.DataBase(db_path, create=True)

        db_obj.execute(QUERY_CREATE)
        db_obj.execute(QUERY_INSERT)

        self.assertTrue(isinstance(db_obj.cursor, sqlite.Cursor))
        self.assertRaises(sqlite.Warning, db_obj.execute, QUERY_MULTIPLE)

        db_obj.close()

    def test_db_host_get(self):
        """Test DataBaseHost get method"""

        db_path = os.path.join(self.res_dir, 'regular.sqlite')
        db_obj = db.DataBaseHost.get(db_path)

        self.assertEqual(db.DataBaseHost.get(db_path), db_obj)

        db_obj.close()


if __name__ == "__main__":
    unittest.main()

# -*- coding: UTF-8 -*-
"""
Tests for db module

:copyright: (c) 2017-2020 by <NAME> (http://alexlitvin.name).
:license: MIT, see LICENSE for more details.
"""
import unittest

import os
import shutil
import tempfile
import sqlite3 as sqlite

import grailkit.db as db


QUERY_CREATE = """CREATE TABLE `test` (`key` TEXT NOT NULL, `value` TEXT, PRIMARY KEY(key));"""
QUERY_INSERT = """INSERT INTO `test` VALUES('key', 'value')"""
QUERY_GET = """SELECT * FROM `test`"""
QUERY_MULTIPLE = """
    create table person(
        first_name,
        last_name,
        age
    );

    create table book(
        title,
        author,
        published
    );

    insert into book(title, author, published)
    values (
        'Dirk Gently''s Holistic Detective Agency',
        '<NAME>',
        1987
    );
    """
QUERY_INSERT_MULTIPLE = """
    INSERT INTO `test` ('key', 'value') VALUES
        ('one', 'first'),
        ('two', 'second'),
        ('three', 'third');
    """


class TestGrailkitDB(unittest.TestCase):

    def setUp(self):
        """Create a temporary directory"""

        self.test_dir = tempfile.mkdtemp()
        self.res_dir = os.path.abspath(__file__[:-3])

    def tearDown(self):
        """Remove the directory after the test"""

        shutil.rmtree(self.test_dir, ignore_errors=True)

    def test_db_create(self):
        """Test database create"""

        db_path = os.path.join(self.test_dir, 'test.sqlite')
        db_obj = db.DataBase(db_path, query=QUERY_CREATE, create=True)
        db_obj.close()

        self.assertTrue(os.path.isfile(db_path))

    def test_db_open(self):
        """Test database open"""

        db_path = os.path.join(self.res_dir, 'regular.sqlite')
        db_obj = db.DataBase(db_path)
        db_obj.close()

        self.assertTrue(True)

    def test_db_open_not_exists(self):
        """Test database open if file not exists"""

        db_path = os.path.join(self.res_dir, 'file_not_exists.sqlite')

        self.assertRaises(db.DataBaseError, db.DataBase, db_path)

    def test_db_open_corrupt(self):
        """Test database open if file corrupted"""

        db_path = os.path.join(self.res_dir, 'file_not_exists.sqlite')

        self.assertRaises(db.DataBaseError, db.DataBase, db_path)

    def test_db_connection(self):
        """Test connection to database"""

        db_path = os.path.join(self.res_dir, 'regular.sqlite')
        db_obj = db.DataBase(db_path)

        self.assertTrue(isinstance(db_obj.connection, sqlite.Connection))

        db_obj.close()

    def test_db_cursor(self):
        """Test cursor property"""

        db_path = os.path.join(self.res_dir, 'regular.sqlite')
        db_obj = db.DataBase(db_path)

        self.assertTrue(isinstance(db_obj.cursor, sqlite.Cursor))

        db_obj.close()

    def test_db_get(self):
        """Test getting single row"""

        db_path = os.path.join(self.test_dir, 'get.sqlite')
        db_obj = db.DataBase(db_path, create=True)

        db_obj.execute(QUERY_CREATE)
        db_obj.execute(QUERY_INSERT)
        res = db_obj.get(QUERY_GET)

        self.assertEqual(res[0], 'key')
        self.assertEqual(res[1], 'value')

        self.assertEqual(res['key'], 'key')
        self.assertEqual(res['value'], 'value')

        db_obj.close()

    def test_db_all(self):
        """Test getting results by query"""

        db_path = os.path.join(self.test_dir, 'get.sqlite')
        db_obj = db.DataBase(db_path, create=True)

        db_obj.execute(QUERY_CREATE)
        db_obj.execute(QUERY_INSERT_MULTIPLE)

        res = db_obj.all(QUERY_GET)

        self.assertEqual(res[0][0], 'one')
        self.assertEqual(res[0][1], 'first')

        self.assertEqual(res[1][0], 'two')
        self.assertEqual(res[1][1], 'second')

        self.assertEqual(res[2][0], 'three')
        self.assertEqual(res[2][1], 'third')

        db_obj.close()

    def test_db_execute(self):
        """Test query execution"""

        db_path = os.path.join(self.test_dir, 'execute.sqlite')
        db_obj = db.DataBase(db_path, create=True)

        db_obj.execute(QUERY_CREATE)
        db_obj.execute(QUERY_INSERT)

        self.assertTrue(isinstance(db_obj.cursor, sqlite.Cursor))
        self.assertRaises(sqlite.Warning, db_obj.execute, QUERY_MULTIPLE)

        db_obj.close()

    def test_db_host_get(self):
        """Test DataBaseHost get method"""

        db_path = os.path.join(self.res_dir, 'regular.sqlite')
        db_obj = db.DataBaseHost.get(db_path)

        self.assertEqual(db.DataBaseHost.get(db_path), db_obj)

        db_obj.close()


if __name__ == "__main__":
    unittest.main()
# -*- coding: UTF-8 -*- Tests for db module
 
 :copyright: (c) 2017-2020 by <NAME> (http://alexlitvin.name).
 :license: MIT, see LICENSE for more details. CREATE TABLE `test` (`key` TEXT NOT NULL, `value` TEXT, PRIMARY KEY(key)); INSERT INTO `test` VALUES('key', 'value') SELECT * FROM `test` create table person(
         first_name,
         last_name,
         age
     );
 
     create table book(
         title,
         author,
         published
     );
 
     insert into book(title, author, published)
     values (
         'Dirk Gently''s Holistic Detective Agency',
         '<NAME>',
         1987
     ); INSERT INTO `test` ('key', 'value') VALUES
         ('one', 'first'),
         ('two', 'second'),
         ('three', 'third'); Create a temporary directory Remove the directory after the test Test database create Test database open Test database open if file not exists Test database open if file corrupted Test connection to database Test cursor property Test getting single row Test getting results by query Test query execution Test DataBaseHost get method