<commit_before>import unittest

from flow.amqp_manager import channel_manager
from flow.amqp_manager import connection_manager
from flow.amqp_manager import exchange_manager
from flow.amqp_manager import queue_manager

import uuid
import os

import logging

if os.getenv('LOG_TESTS'):
    from flow.util import log_formatter

    PIKA_LOG_LEVEL = logging.INFO
    LOG_LEVEL = logging.DEBUG
    LOG_FORMAT = ('%(levelname)-23s %(asctime)s %(name)-60s %(funcName)-45s'
                  ' %(lineno)5d: %(message)s')

    console_handler = logging.StreamHandler()
    console_handler.setFormatter(log_formatter.ColorFormatter(LOG_FORMAT))
    console_handler.setLevel(LOG_LEVEL)

    LOG = logging.getLogger()
    LOG.addHandler(console_handler)

    LOG.setLevel(LOG_LEVEL)
    logging.getLogger('pika').setLevel(PIKA_LOG_LEVEL)
else:
    import flow.amqp_manager

    LOG = logging.getLogger()
    LOG.addHandler(flow.amqp_manager.nh)


class MockProcess(object):
    def __init__(self, em):
        self.em = em

    def message_handler(self, properties, data, ack_callback, reject_callback):
        LOG.info("Got message (properties = %s): '%s'", properties, data)
        ack_callback()
        LOG.info('Going to shut down.')
        self.cm.stop()

    def set_queue_name(self, queue_manager):
        self.queue_name = queue_manager.queue_name

    def start(self, cm):
        self.cm = cm

        self.em.publish(self.queue_name, 'welcome friends')

class SystemTest(unittest.TestCase):
    def setUp(self):
        self.em = exchange_manager.ExchangeManager('', exchange_type='direct')

        self.process = MockProcess(self.em)

        self.amqp_url = os.environ['TESTING_AMQP_URL']
        self.queue_name = 'system_test_%s' % uuid.uuid4().hex

        self.qm = queue_manager.QueueManager(self.queue_name,
                durable=False, auto_delete=True,
                message_handler=self.process.message_handler)
        self.qm.add_ready_callback(self.process.set_queue_name)

    def test_basic_channel_manager(self):
        self.chan_man = channel_manager.ChannelManager(
                delegates=[self.em, self.qm])

        self.con_man = connection_manager.ConnectionManager(
                self.amqp_url, delegates=[self.chan_man])

        self.con_man.add_ready_callback(self.process.start)
<commit_msg>Put in additional enviroment check into test<commit_after>import unittest

from flow.amqp_manager import channel_manager
from flow.amqp_manager import connection_manager
from flow.amqp_manager import exchange_manager
from flow.amqp_manager import queue_manager

import uuid
import os

import logging

if os.getenv('LOG_TESTS'):
    from flow.util import log_formatter

    PIKA_LOG_LEVEL = logging.INFO
    LOG_LEVEL = logging.DEBUG
    LOG_FORMAT = ('%(levelname)-23s %(asctime)s %(name)-60s %(funcName)-45s'
                  ' %(lineno)5d: %(message)s')

    console_handler = logging.StreamHandler()
    console_handler.setFormatter(log_formatter.ColorFormatter(LOG_FORMAT))
    console_handler.setLevel(LOG_LEVEL)

    LOG = logging.getLogger()
    LOG.addHandler(console_handler)

    LOG.setLevel(LOG_LEVEL)
    logging.getLogger('pika').setLevel(PIKA_LOG_LEVEL)
else:
    import flow.amqp_manager

    LOG = logging.getLogger()
    LOG.addHandler(flow.amqp_manager.nh)


class MockProcess(object):
    def __init__(self, em):
        self.em = em

    def message_handler(self, properties, data, ack_callback, reject_callback):
        LOG.info("Got message (properties = %s): '%s'", properties, data)
        ack_callback()
        LOG.info('Going to shut down.')
        self.cm.stop()

    def set_queue_name(self, queue_manager):
        self.queue_name = queue_manager.queue_name

    def start(self, cm):
        self.cm = cm

        self.em.publish(self.queue_name, 'welcome friends')

class SystemTest(unittest.TestCase):
    def setUp(self):
        self.em = exchange_manager.ExchangeManager('', exchange_type='direct')

        self.process = MockProcess(self.em)

        self.amqp_url = os.getenv('TESTING_AMQP_URL')
        if not self.amqp_url:
            raise RuntimeError('Environment variable TESTING_AMQP_URL must be set')
        self.queue_name = 'system_test_%s' % uuid.uuid4().hex

        self.qm = queue_manager.QueueManager(self.queue_name,
                durable=False, auto_delete=True,
                message_handler=self.process.message_handler)
        self.qm.add_ready_callback(self.process.set_queue_name)

    def test_basic_channel_manager(self):
        self.chan_man = channel_manager.ChannelManager(
                delegates=[self.em, self.qm])

        self.con_man = connection_manager.ConnectionManager(
                self.amqp_url, delegates=[self.chan_man])

        self.con_man.add_ready_callback(self.process.start)

{'fraction_non_alphanumeric': 0.10261736601578729, 'fraction_numerical': 0.0033236393851267137, 'mean_word_length': 3.3614130434782608, 'pattern_counts': {'":': 0, '<': 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': '4661383', 'n_tokens_mistral': 1545, 'n_tokens_neox': 1483, 'n_words': 268}