<commit_before>import shlex
import time
import Queue
from copy import deepcopy
from subprocess import call
import unittest

import sys; sys.path.append('..')
import __meta__; __meta__.load_paths()
from rwtypes.readers.db.DBReader import DBReader

from third.sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
from third.sqlalchemy.orm import sessionmaker


def get_queue(maxsize=1024):
    return Queue.Queue(maxsize=maxsize)


def create_database(db_name):
    call(shlex.split("mysql -e 'create database %s'" % db_name))


def drop_database(db_name):
    call(shlex.split("mysql -e 'drop database %s'" % db_name))


def db_periodic_reading(f):
    def wrapper(self):
        engine = create_engine(self.connection, echo=False)
        metadata = MetaData()
        table = Table('users', metadata, Column(u'id', Integer()),
                                         Column(u'name', String(7)))
        metadata.create_all(engine)
        conn = engine.connect()
        conn.execute(table.insert(), [
            {'id': 0, 'name': 'spam'},
            {'id': 1, 'name': 'egg'},
            {'id': 2, 'name': 'bacon'},
        ])
        f(self)
    return wrapper

class TestDBReader(unittest.TestCase):
    def setUp(self):
        self.db_name = "dbreader_test"
        drop_database(self.db_name)
        self.user = ''
        self.password = ''
        self.host = "localhost"
        self.connection_format = 'mysql+mysqldb://%s:%s@%s/%s'
        self.connection = self.connection_format % (self.user,
                                                    self.password,
                                                    self.host,
                                                    self.db_name)
        self.base_conf = {'database' : self.db_name,
                          'host': self.host,
                          'user': self.user,
                          'passwd': <PASSWORD>,
                          'connection': self.connection_format}
        create_database(self.db_name)

    def tearDown(self):
        drop_database(self.db_name)

    @db_periodic_reading
    def test_happy_periodic_reading(self):
        q = get_queue()
        conf = deepcopy(self.base_conf)
        conf.update({'period': 1})
        conf.update({'columns': ['id', 'name'],
                     'query': 'select id, name from users'})

        myreader = DBReader(q, conf=conf)
        myreader.start()

        # waits for processing messages
        time.sleep(3.1)

        message = q.get()
        self.assertEqual({'id': 0, 'name': 'spam'}, message.content)

        message = q.get()
        self.assertEqual({'id': 1, 'name': 'egg'}, message.content)

        message = q.get()
        self.assertEqual({'id': 2, 'name': 'bacon'}, message.content)

<commit_msg>minor refactoring<commit_after>import shlex
import time
import Queue
from copy import deepcopy
from subprocess import call
import unittest

import sys; sys.path.append('..')
import __meta__; __meta__.load_paths()
from rwtypes.readers.db.DBReader import DBReader

from third.sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
from third.sqlalchemy.orm import sessionmaker


def get_queue(maxsize=1024):
    return Queue.Queue(maxsize=maxsize)


def create_database(db_name):
    call(shlex.split("mysql -e 'create database %s'" % db_name))


def drop_database(db_name):
    call(shlex.split("mysql -e 'drop database %s'" % db_name))


def db_periodic_reading(f):
    def wrapper(self):
        engine = create_engine(self.connection, echo=False)
        metadata = MetaData()
        table = Table('users', metadata, Column(u'id', Integer()),
                                         Column(u'name', String(7)))
        metadata.create_all(engine)
        conn = engine.connect()
        conn.execute(table.insert(), [
            {'id': 0, 'name': 'spam'},
            {'id': 1, 'name': 'egg'},
            {'id': 2, 'name': 'bacon'},
        ])
        f(self)
    return wrapper

class TestDBReader(unittest.TestCase):
    def setUp(self):
        self.db_name = "dbreader_test"
        drop_database(self.db_name)
        self.user = ''
        self.password = ''
        self.host = "localhost"
        self.connection_format = 'mysql+mysqldb://%s:%s@%s/%s'
        self.connection = self.connection_format % (self.user,
                                                    self.password,
                                                    self.host,
                                                    self.db_name)
        self.base_conf = {'database' : self.db_name,
                          'host': self.host,
                          'user': self.user,
                          'passwd': <PASSWORD>,
                          'connection': self.connection_format}
        create_database(self.db_name)

    def tearDown(self):
        drop_database(self.db_name)

    @db_periodic_reading
    def test_happy_periodic_reading(self):
        q = get_queue()
        conf = deepcopy(self.base_conf)
        conf.update({'period': 1,
                     'columns': ['id', 'name'],
                     'query': 'select id, name from users'})

        myreader = DBReader(q, conf=conf)
        myreader.start()

        # waits for processing messages
        time.sleep(3.1)

        message = q.get()
        self.assertEqual({'id': 0, 'name': 'spam'}, message.content)

        message = q.get()
        self.assertEqual({'id': 1, 'name': 'egg'}, message.content)

        message = q.get()
        self.assertEqual({'id': 2, 'name': 'bacon'}, message.content)


{'fraction_non_alphanumeric': 0.1288244766505636, 'fraction_numerical': 0.0050098407586330294, 'mean_word_length': 1.982924226254002, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2945906', 'n_tokens_mistral': 1652, 'n_tokens_neox': 1580, 'n_words': 353}