<commit_before>import unittest
from Queue import Queue, Empty
from datetime import datetime, timedelta
from multiprocessing import get_logger

from carrot.connection import BrokerConnection
from carrot.backends.base import BaseMessage
from billiard.serialization import pickle

from celery import registry
from celery.utils import gen_unique_id
from celery.worker import CarrotListener, WorkController
from celery.worker.job import TaskWrapper
from celery.worker.scheduler import Scheduler
from celery.decorators import task as task_dec


@task_dec()
def foo_task(x, y, z, **kwargs):
    return x * y * z


class MockLogger(object):

    def critical(self, *args, **kwargs):
        pass

    def info(self, *args, **kwargs):
        pass

    def error(self, *args, **kwargs):
        pass

    def debug(self, *args, **kwargs):
        pass


class MockBackend(object):
    _acked = False

    def ack(self, delivery_tag):
        self._acked = True


class MockPool(object):

    def __init__(self, *args, **kwargs):
        self.raise_regular = kwargs.get("raise_regular", False)
        self.raise_base = kwargs.get("raise_base", False)

    def apply_async(self, *args, **kwargs):
        if self.raise_regular:
            raise KeyError("some exception")
        if self.raise_base:
            raise KeyboardInterrupt("Ctrl+c")

    def start(self):
        pass

    def stop(self):
        pass
        return True


class MockController(object):

    def __init__(self, w, *args, **kwargs):
        self._w = w
        self._stopped = False

    def start(self):
        self._w["started"] = True
        self._stopped = False

    def stop(self):
        self._stopped = True


def create_message(backend, **data):
    data["id"] = gen_unique_id()
    return BaseMessage(backend, body=pickle.dumps(dict(**data)),
                       content_type="application/x-python-serialize",
                       content_encoding="binary")


class TestCarrotListener(unittest.TestCase):

    def setUp(self):
        self.ready_queue = Queue()
        self.eta_scheduler = Scheduler(self.ready_queue)
        self.logger = get_logger()
        self.logger.setLevel(0)

    def test_connection(self):
        l = CarrotListener(self.ready_queue, self.eta_scheduler, self.logger)

        c = l.reset_connection()
        self.assertTrue(isinstance(l.amqp_connection, BrokerConnection))

        l.close_connection()
        self.assertTrue(l.amqp_connection is None)
        self.assertTrue(l.task_consumer is None)

        c = l.reset_connection()
        self.assertTrue(isinstance(l.amqp_connection, BrokerConnection))

        l.stop()
        self.assertTrue(l.amqp_connection is None)
        self.assertTrue(l.task_consumer is None)

    def test_receieve_message(self):
        l = CarrotListener(self.ready_queue, self.eta_scheduler, self.logger)
        backend = MockBackend()
        m = create_message(backend, task=foo_task.name,
                           args=[2, 4, 8], kwargs={})

        l.receive_message(m.decode(), m)

        in_bucket = self.ready_queue.get_nowait()
        self.assertTrue(isinstance(in_bucket, TaskWrapper))
        self.assertEquals(in_bucket.task_name, foo_task.name)
        self.assertEquals(in_bucket.execute(), 2 * 4 * 8)
        self.assertTrue(self.eta_scheduler.empty())

    def test_receieve_message_not_registered(self):
        l = CarrotListener(self.ready_queue, self.eta_scheduler, self.logger)
        backend = MockBackend()
        m = create_message(backend, task="x.X.31x", args=[2, 4, 8], kwargs={})

        self.assertFalse(l.receive_message(m.decode(), m))
        self.assertRaises(Empty, self.ready_queue.get_nowait)
        self.assertTrue(self.eta_scheduler.empty())

    def test_receieve_message_eta(self):
        l = CarrotListener(self.ready_queue, self.eta_scheduler, self.logger)
        backend = MockBackend()
        m = create_message(backend, task=foo_task.name,
                           args=[2, 4, 8], kwargs={},
                           eta=(datetime.now() +
                               timedelta(days=1)).isoformat())

        l.receive_message(m.decode(), m)

        in_hold = self.eta_scheduler.queue[0]
        self.assertEquals(len(in_hold), 4)
        eta, priority, task, on_accept = in_hold
        self.assertTrue(isinstance(task, TaskWrapper))
        self.assertTrue(callable(on_accept))
        self.assertEquals(task.task_name, foo_task.name)
        self.assertEquals(task.execute(), 2 * 4 * 8)
        self.assertRaises(Empty, self.ready_queue.get_nowait)


class TestWorkController(unittest.TestCase):

    def setUp(self):
        self.worker = WorkController(concurrency=1, loglevel=0,
                                     is_detached=False)
        self.worker.logger = MockLogger()

    def test_attrs(self):
        worker = self.worker
        self.assertTrue(isinstance(worker.eta_scheduler, Scheduler))
        self.assertTrue(worker.schedule_controller)
        self.assertTrue(worker.pool)
        self.assertTrue(worker.broker_listener)
        self.assertTrue(worker.mediator)
        self.assertTrue(worker.components)

    def test_safe_process_task(self):
        worker = self.worker
        worker.pool = MockPool()
        backend = MockBackend()
        m = create_message(backend, task=foo_task.name, args=[4, 8, 10],
                           kwargs={})
        task = TaskWrapper.from_message(m, m.decode())
        worker.safe_process_task(task)
        worker.pool.stop()

    def test_safe_process_task_raise_base(self):
        worker = self.worker
        worker.pool = MockPool(raise_base=True)
        backend = MockBackend()
        m = create_message(backend, task=foo_task.name, args=[4, 8, 10],
                           kwargs={})
        task = TaskWrapper.from_message(m, m.decode())
        worker.safe_process_task(task)
        worker.pool.stop()

    def test_safe_process_task_raise_regular(self):
        worker = self.worker
        worker.pool = MockPool(raise_regular=True)
        backend = MockBackend()
        m = create_message(backend, task=foo_task.name, args=[4, 8, 10],
                           kwargs={})
        task = TaskWrapper.from_message(m, m.decode())
        worker.safe_process_task(task)
        worker.pool.stop()

    def test_start_stop(self):
        worker = self.worker
        w1 = {"started": False}
        w2 = {"started": False}
        w3 = {"started": False}
        w4 = {"started": False}
        worker.components = [MockController(w1), MockController(w2),
                             MockController(w3), MockController(w4)]

        worker.start()
        for w in (w1, w2, w3, w4):
            self.assertTrue(w["started"])
        for component in worker.components:
            self.assertTrue(component._stopped)
<commit_msg>Fix broken tests.<commit_after>import unittest
from Queue import Queue, Empty
from datetime import datetime, timedelta
from multiprocessing import get_logger

from carrot.connection import BrokerConnection
from carrot.backends.base import BaseMessage
from billiard.serialization import pickle

from celery import registry
from celery.utils import gen_unique_id
from celery.worker import CarrotListener, WorkController
from celery.worker.job import TaskWrapper
from celery.worker.scheduler import Scheduler
from celery.decorators import task as task_dec


class MockEventDispatcher(object):

    def send(self, *args, **kwargs):
        pass

    def close(self):
        pass


@task_dec()
def foo_task(x, y, z, **kwargs):
    return x * y * z


class MockLogger(object):

    def critical(self, *args, **kwargs):
        pass

    def info(self, *args, **kwargs):
        pass

    def error(self, *args, **kwargs):
        pass

    def debug(self, *args, **kwargs):
        pass


class MockBackend(object):
    _acked = False

    def ack(self, delivery_tag):
        self._acked = True


class MockPool(object):

    def __init__(self, *args, **kwargs):
        self.raise_regular = kwargs.get("raise_regular", False)
        self.raise_base = kwargs.get("raise_base", False)

    def apply_async(self, *args, **kwargs):
        if self.raise_regular:
            raise KeyError("some exception")
        if self.raise_base:
            raise KeyboardInterrupt("Ctrl+c")

    def start(self):
        pass

    def stop(self):
        pass
        return True


class MockController(object):

    def __init__(self, w, *args, **kwargs):
        self._w = w
        self._stopped = False

    def start(self):
        self._w["started"] = True
        self._stopped = False

    def stop(self):
        self._stopped = True


def create_message(backend, **data):
    data["id"] = gen_unique_id()
    return BaseMessage(backend, body=pickle.dumps(dict(**data)),
                       content_type="application/x-python-serialize",
                       content_encoding="binary")


class TestCarrotListener(unittest.TestCase):

    def setUp(self):
        self.ready_queue = Queue()
        self.eta_scheduler = Scheduler(self.ready_queue)
        self.logger = get_logger()
        self.logger.setLevel(0)

    def test_connection(self):
        l = CarrotListener(self.ready_queue, self.eta_scheduler, self.logger,
                           send_events=False)

        c = l.reset_connection()
        self.assertTrue(isinstance(l.amqp_connection, BrokerConnection))

        l.close_connection()
        self.assertTrue(l.amqp_connection is None)
        self.assertTrue(l.task_consumer is None)

        c = l.reset_connection()
        self.assertTrue(isinstance(l.amqp_connection, BrokerConnection))

        l.stop()
        self.assertTrue(l.amqp_connection is None)
        self.assertTrue(l.task_consumer is None)

    def test_receieve_message(self):
        l = CarrotListener(self.ready_queue, self.eta_scheduler, self.logger,
                           send_events=False)
        backend = MockBackend()
        m = create_message(backend, task=foo_task.name,
                           args=[2, 4, 8], kwargs={})

        l.event_dispatcher = MockEventDispatcher()
        l.receive_message(m.decode(), m)

        in_bucket = self.ready_queue.get_nowait()
        self.assertTrue(isinstance(in_bucket, TaskWrapper))
        self.assertEquals(in_bucket.task_name, foo_task.name)
        self.assertEquals(in_bucket.execute(), 2 * 4 * 8)
        self.assertTrue(self.eta_scheduler.empty())

    def test_receieve_message_not_registered(self):
        l = CarrotListener(self.ready_queue, self.eta_scheduler, self.logger,
                          send_events=False)
        backend = MockBackend()
        m = create_message(backend, task="x.X.31x", args=[2, 4, 8], kwargs={})

        l.event_dispatcher = MockEventDispatcher()
        self.assertFalse(l.receive_message(m.decode(), m))
        self.assertRaises(Empty, self.ready_queue.get_nowait)
        self.assertTrue(self.eta_scheduler.empty())

    def test_receieve_message_eta(self):
        l = CarrotListener(self.ready_queue, self.eta_scheduler, self.logger,
                          send_events=False)
        backend = MockBackend()
        m = create_message(backend, task=foo_task.name,
                           args=[2, 4, 8], kwargs={},
                           eta=(datetime.now() +
                               timedelta(days=1)).isoformat())

        l.reset_connection()
        l.receive_message(m.decode(), m)

        in_hold = self.eta_scheduler.queue[0]
        self.assertEquals(len(in_hold), 4)
        eta, priority, task, on_accept = in_hold
        self.assertTrue(isinstance(task, TaskWrapper))
        self.assertTrue(callable(on_accept))
        self.assertEquals(task.task_name, foo_task.name)
        self.assertEquals(task.execute(), 2 * 4 * 8)
        self.assertRaises(Empty, self.ready_queue.get_nowait)


class TestWorkController(unittest.TestCase):

    def setUp(self):
        self.worker = WorkController(concurrency=1, loglevel=0,
                                     is_detached=False)
        self.worker.logger = MockLogger()

    def test_attrs(self):
        worker = self.worker
        self.assertTrue(isinstance(worker.eta_scheduler, Scheduler))
        self.assertTrue(worker.schedule_controller)
        self.assertTrue(worker.pool)
        self.assertTrue(worker.broker_listener)
        self.assertTrue(worker.mediator)
        self.assertTrue(worker.components)

    def test_safe_process_task(self):
        worker = self.worker
        worker.pool = MockPool()
        backend = MockBackend()
        m = create_message(backend, task=foo_task.name, args=[4, 8, 10],
                           kwargs={})
        task = TaskWrapper.from_message(m, m.decode())
        worker.safe_process_task(task)
        worker.pool.stop()

    def test_safe_process_task_raise_base(self):
        worker = self.worker
        worker.pool = MockPool(raise_base=True)
        backend = MockBackend()
        m = create_message(backend, task=foo_task.name, args=[4, 8, 10],
                           kwargs={})
        task = TaskWrapper.from_message(m, m.decode())
        worker.safe_process_task(task)
        worker.pool.stop()

    def test_safe_process_task_raise_regular(self):
        worker = self.worker
        worker.pool = MockPool(raise_regular=True)
        backend = MockBackend()
        m = create_message(backend, task=foo_task.name, args=[4, 8, 10],
                           kwargs={})
        task = TaskWrapper.from_message(m, m.decode())
        worker.safe_process_task(task)
        worker.pool.stop()

    def test_start_stop(self):
        worker = self.worker
        w1 = {"started": False}
        w2 = {"started": False}
        w3 = {"started": False}
        w4 = {"started": False}
        worker.components = [MockController(w1), MockController(w2),
                             MockController(w3), MockController(w4)]

        worker.start()
        for w in (w1, w2, w3, w4):
            self.assertTrue(w["started"])
        for component in worker.components:
            self.assertTrue(component._stopped)

{'fraction_non_alphanumeric': 0.10859087042812589, 'fraction_numerical': 0.0066628863056421885, 'mean_word_length': 2.5219670494258613, 'pattern_counts': {'":': 8, '<': 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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1975710', 'n_tokens_mistral': 4248, 'n_tokens_neox': 4119, 'n_words': 825}