<commit_before>import unittest
from loremdb.database.core import DataBaseCreator
from loremdb.database import mysql
from loremdb.util import ContentGen
from random import Random
from abc import ABCMeta, abstractmethod
from datetime import date, datetime


class DataBaseTestCase(unittest.TestCase):
    def setUp(self):
        self.database = self.get_database()
        self.database.show_errors = False
        self.clean_tables()

    def tearDown(self):
        self.database.commit()

    def get_database(self):
        return DataBaseCreator().create_mysql(
            ContentGen(),
            user="root",
            database="loremdb_test"
        )

    def clean_tables(self):
        c = self.database.get_cursor()
        c.execute("DELETE FROM users")
        c.execute("DELETE FROM sections")
        self.database.commit()
        c.close()


class TestDataBase(DataBaseTestCase):
    num = 10

    def setUp(self):
        super(TestDataBase, self).setUp()
        self.signal_calls = []
        self.signal_calls_counter = 0

    def test_fill(self):
        c = self.database.get_cursor()

        c.execute("SELECT * from users")
        self.assertEquals(0, len(c.fetchall()))

        c.execute("SELECT * from sections")
        self.assertEquals(0, len(c.fetchall()))

        self.database.fill(n=10)

        c.execute("SELECT * from users")
        self.assertEquals(10, len(c.fetchall()))

        c.execute("SELECT * from sections")
        self.assertEquals(10, len(c.fetchall()))

        c.close()

    def test_get_tables(self):
        tables = self.database.get_tables()

        self.assertEquals(2, len(tables))

        self.assertTrue("users" in tables)

    def test_filter(self):
        c = self.database.get_cursor()

        c.execute("SELECT * from users")
        self.assertEquals(0, len(c.fetchall()))

        c.execute("SELECT * from sections")
        self.assertEquals(0, len(c.fetchall()))

        self.database.filter("sections")
        self.database.fill(n=10)

        c.execute("SELECT * from users")
        self.assertEquals(0, len(c.fetchall()))

        c.execute("SELECT * from sections")
        self.assertEquals(10, len(c.fetchall()))

    def test_filter_with_nonexistent_table(self):
        self.database.filter("flunfla")
        self.assertRaises(Exception, self.database.fill, 10)

    def test_signal_on_change_table(self):
        self.database.on_change_table.register(self.signal_callback)

        self.database.fill(10)

        result = [
            {"name": "sections"},
            {"name": "users"},
        ]

        self.assertEquals(result, self.signal_calls)

    def test_signal_on_insert(self):
        self.database.on_insert.register(self.signal_callback_on_insert)
        self.database.filter("sections")
        self.database.fill(10)
        self.assertEquals(10, self.signal_calls_counter)

    def signal_callback(self, table_info):
        self.signal_calls.append(table_info)

    def signal_callback_on_insert(self):
        self.signal_calls_counter += 1


class TestTable(DataBaseTestCase):

    def setUp(self):
        super(TestTable, self).setUp()
        self.table = mysql.Table(self.database, "users", ContentGen())
        self.table.show_errors = False
        self.signal_calls_counter = 0

    def test_fill(self):
        c = self.table.get_cursor()

        c.execute("SELECT name, age from users")
        self.assertEquals(0, len(c.fetchall()))

        self.table.fill(n=10)
        self.database.commit()

        c.execute("SELECT name, age from users")
        self.assertEquals(10, len(c.fetchall()))

    def test_signal(self):
        self.table.on_insert.register(self.signal_callback)

        self.table.fill(n=10)

        self.assertEquals(10, self.signal_calls_counter)

    def signal_callback(self):
        self.signal_calls_counter += 1

    def test_get_table_info(self):
        self.assertEquals(
            {"name": "users"},
            self.table.table_info
        )


class TestFieldCreatorFromMysql(unittest.TestCase):
    def setUp(self):
        self.creator = mysql.FieldCreatorFromMysql()<commit_msg>Fixed unit/integration test for mysql<commit_after>import unittest
from loremdb.database.core import DataBaseCreator
from loremdb.database import mysql
from loremdb.util import ContentGen
from random import Random
from abc import ABCMeta, abstractmethod
from datetime import date, datetime


class DataBaseTestCase(unittest.TestCase):
    def setUp(self):
        self.database = self.get_database()
        self.database.show_errors = False
        self.clean_tables()

    def tearDown(self):
        self.database.commit()

    def get_database(self):
        return DataBaseCreator().create_mysql(
            ContentGen(),
            user="root",
            database="loremdb_test"
        )

    def clean_tables(self):
        c = self.database.get_cursor()
        c.execute("DELETE FROM users")
        c.execute("DELETE FROM sections")
        self.database.commit()
        c.close()


class TestDataBase(DataBaseTestCase):
    num = 10

    def setUp(self):
        super(TestDataBase, self).setUp()
        self.signal_calls = []
        self.signal_calls_counter = 0

    def test_fill(self):
        c = self.database.get_cursor()

        c.execute("SELECT * from users")
        self.assertEquals(0, len(c.fetchall()))

        c.execute("SELECT * from sections")
        self.assertEquals(0, len(c.fetchall()))

        self.database.fill(n=10)

        c.execute("SELECT * from users")
        self.assertEquals(10, len(c.fetchall()))

        c.execute("SELECT * from sections")
        self.assertEquals(10, len(c.fetchall()))

        c.close()

    def test_get_tables(self):
        tables = self.database.get_tables()

        self.assertEquals(2, len(tables))

        self.assertTrue("users" in tables)

    def test_filter(self):
        c = self.database.get_cursor()

        c.execute("SELECT * from users")
        self.assertEquals(0, len(c.fetchall()))

        c.execute("SELECT * from sections")
        self.assertEquals(0, len(c.fetchall()))

        self.database.filter("sections")
        self.database.fill(n=10)

        c.execute("SELECT * from users")
        self.assertEquals(0, len(c.fetchall()))

        c.execute("SELECT * from sections")
        self.assertEquals(10, len(c.fetchall()))

    def test_filter_with_nonexistent_table(self):
        self.database.filter("flunfla")
        self.assertRaises(Exception, self.database.fill, 10)

    def test_signal_on_change_table(self):
        self.database.on_change_table.register(self.signal_callback)

        self.database.fill(10)

        result = [
            {"name": "sections"},
            {"name": "users"},
        ]

        self.assertEquals(result, self.signal_calls)

    def test_signal_on_insert(self):
        self.database.on_insert.register(self.signal_callback_on_insert)
        self.database.filter("sections")
        self.database.fill(10)
        self.assertEquals(10, self.signal_calls_counter)

    def signal_callback(self, table_info):
        self.signal_calls.append(table_info)

    def signal_callback_on_insert(self):
        self.signal_calls_counter += 1


class TestTable(DataBaseTestCase):

    def setUp(self):
        super(TestTable, self).setUp()
        self.table = mysql.Table(self.database, "users", ContentGen())
        self.table.show_errors = False
        self.signal_calls_counter = 0

    def test_fill(self):
        c = self.table.get_cursor()

        c.execute("SELECT * from users")
        self.assertEquals(0, len(c.fetchall()))

        self.table.fill(n=10)
        self.database.commit()

        c.execute("SELECT * from users")
        self.assertEquals(10, len(c.fetchall()))

    def test_signal(self):
        self.table.on_insert.register(self.signal_callback)

        self.table.fill(n=10)

        self.assertEquals(10, self.signal_calls_counter)

    def signal_callback(self):
        self.signal_calls_counter += 1

    def test_get_table_info(self):
        self.assertEquals(
            {"name": "users"},
            self.table.table_info
        )


class TestFieldCreatorFromMysql(unittest.TestCase):
    def setUp(self):
        self.creator = mysql.FieldCreatorFromMysql()
{'fraction_non_alphanumeric': 0.11799374548953573, 'fraction_numerical': 0.009381765696415684, 'mean_word_length': 2.6405429071803854, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4526774', 'n_tokens_mistral': 2542, 'n_tokens_neox': 2430, 'n_words': 437}