def _setup_workers(self, bucketters, aggregators, bucket_size):
        broker = FakeAMQPBroker()
        self.broker = BrokerWrapper(broker)

        bucket_config = {
            'buckets': aggregators,
            'bucket_size': bucket_size,
            }
        for _i in range(bucketters):
            worker = get_stubbed_worker(metrics.MetricTimeBucket,
                                        config=bucket_config,
                                        broker=broker)
            yield worker.startWorker()
            self.bucket_workers.append(worker)

        aggregator_config = {
            'bucket_size': bucket_size,
            }
        for i in range(aggregators):
            config = aggregator_config.copy()
            config['bucket'] = i
            worker = get_stubbed_worker(metrics.MetricAggregator,
                                        config=config, broker=broker)
            worker._time = self.fake_time
            yield worker.startWorker()
            self.aggregator_workers.append(worker)

 def setUp(self):
     self.config = {
         'web_port': 9999,
         'web_receive_path': '/t/receive',
         'web_receipt_path': '/t/receipt',
         'url': 'http://localhost:9998/t/send',
     }
     self.worker = get_stubbed_worker(TestFakeVas2NetsWorker, self.config)
     self.test_worker = get_stubbed_worker(TestWorker, self.config)
     self.today = datetime.utcnow().date()

 def setUp(self):
     self.transport_name = 'test'
     self.config = {'transport_name': self.transport_name}
     self.worker = get_stubbed_worker(DummyApplicationWorker,
                                      config=self.config)
     self.broker = self.worker._amqp_client.broker
     yield self.worker.startWorker()

 def get_worker(self):
     worker = get_stubbed_worker(RockPaperScissorsWorkerStub, {
             'transport_name': 'foo',
             'ussd_code': '99999',
             })
     worker.startWorker()
     return worker

 def mk_failure_worker(self, config, broker, redis):
     w = get_stubbed_worker(Vas2NetsFailureWorker, config, broker)
     self.workers.append(w)
     yield w.startWorker()
     w.retry_publisher = yield self.worker.publish_to("foo")
     w.r_server = redis
     returnValue(w)

    def get_application(self, config, cls=None, start=True):
        """
        Get an instance of a worker class.

        :param config: Config dict.
        :param cls: The Application class to instantiate.
                    Defaults to :attr:`application_class`
        :param start: True to start the application (default), False otherwise.

        Some default config values are helpfully provided in the
        interests of reducing boilerplate:

        * ``transport_name`` defaults to :attr:`self.transport_name`
        * ``send_to`` defaults to a dictionary with config for each tag
          defined in worker's SEND_TO_TAGS attribute. Each tag's config
          contains a transport_name set to ``<tag>_outbound``.
        """

        if cls is None:
            cls = self.application_class
        config.setdefault('transport_name', self.transport_name)
        if 'send_to' not in config and cls.SEND_TO_TAGS:
            config['send_to'] = {}
            for tag in cls.SEND_TO_TAGS:
                config['send_to'][tag] = {
                    'transport_name': '%s_outbound' % tag}
        worker = get_stubbed_worker(cls, config, self._amqp)
        self._workers.append(worker)
        if start:
            yield worker.startWorker()
        returnValue(worker)

    def test_bucketing(self):
        config = {'buckets': 4, 'bucket_size': 5}
        worker = get_stubbed_worker(metrics.MetricTimeBucket, config=config)
        broker = BrokerWrapper(worker._amqp_client.broker)
        yield worker.startWorker()

        datapoints = [
            ("vumi.test.foo", ("agg",), [(1230, 1.5), (1235, 2.0)]),
            ("vumi.test.bar", ("sum",), [(1240, 1.0)]),
            ]
        broker.send_datapoints("vumi.metrics", "vumi.metrics", datapoints)
        yield broker.kick_delivery()

        buckets = [broker.recv_datapoints("vumi.metrics.buckets",
                                          "bucket.%d" % i) for i in range(4)]

        expected_buckets = [
            [],
            [[[u'vumi.test.bar', ['sum'], [[1240, 1.0]]]]],
            [[[u'vumi.test.foo', ['agg'], [[1230, 1.5]]]],
             [[u'vumi.test.foo', ['agg'], [[1235, 2.0]]]]],
            [],
            ]

        self.assertEqual(buckets, expected_buckets)

        yield worker.stopWorker()

 def test_send_to_with_no_send_to_tags(self):
     config = {'transport_name': 'notags_app'}
     notags_worker = get_stubbed_worker(ApplicationWorker,
                                        config=config)
     yield notags_worker.startWorker()
     self.assertRaises(ValueError, notags_worker.send_to,
                       '+12345', "Hi!")

 def get_worker(self, **config_extras):
     if getattr(self, 'worker', None) is not None:
         yield self.worker.stopWorker()
     config = {
         "transport_names": [
             "transport1",
             "transport2",
             "transport3",
             ],
         "exposed_names": [
             "app1",
             "app2",
             "app3",
             ],
         "router_class": "vumi.dispatchers.base.SimpleDispatchRouter",
         "route_mappings": {
             "transport1": ["app1"],
             "transport2": ["app2"],
             "transport3": ["app1", "app3"]
             },
         }
     config.update(config_extras)
     self.worker = get_stubbed_worker(BaseDispatchWorker, config)
     self._amqp = self.worker._amqp_client.broker
     yield self.worker.startWorker()

 def get_worker(self):
     worker = get_stubbed_worker(MemoWorker, {
             # 'transport_name': 'foo',
             # 'ussd_code': '99999',
             })
     yield worker.startWorker()
     returnValue(worker)

 def mk_resource_worker(self, body, headers=None, code=http.OK):
     w = get_stubbed_worker(TestResourceWorker, {}, self.broker)
     self.workers.append(w)
     w.set_resources([(self.path, BadVas2NetsResource,
                       (body, headers, code))])
     yield w.startWorker()
     returnValue(w)

 def make_resource_worker(self, code=http.OK, content='',
                             headers={}, callback=None):
     w = get_stubbed_worker(TestResourceWorker, {})
     w.set_resources([
             (self.path, TestResource, (code, content, headers, callback))])
     self._workers.append(w)
     yield w.startWorker()
     returnValue(w)

 def get_worker(self):
     worker = get_stubbed_worker(RockPaperScissorsWorker, {
             'transport_name': 'foo',
             'ussd_code': '99999',
             }, self._amqp)
     self._workers.append(worker)
     worker.startWorker()
     return worker

    def test_start_publisher(self):
        """The publisher should publish"""
        worker = get_stubbed_worker(Worker)
        publisher = yield worker.publish_to('test.routing.key')
        self.assertEquals(publisher.routing_key, 'test.routing.key')
        publisher.publish_message(Message(key="value"))
        [published_msg] = publisher.channel.broker.get_dispatched(
            'vumi', 'test.routing.key')

        self.assertEquals(published_msg.body, '{"key": "value"}')
        self.assertEquals(published_msg.properties, {'delivery mode': 2})

 def get_worker(self, config=None):
     if not config:
         config = {
             'worker_name': 'wikitest',
             'sms_transport': 'sphex',
             }
     config.setdefault('transport_name', self.transport_name)
     worker = get_stubbed_worker(WikipediaWorker, config, self.broker)
     self._workers.append(worker)
     yield worker.startWorker()
     returnValue(worker)

 def setUp(self):
     config = {
         'transport_name': 'test_infobip',
         'transport_type': 'ussd',
         'web_path': "/session/",
         'web_port': 0,
         }
     self.worker = get_stubbed_worker(InfobipTransport, config)
     self.broker = self.worker._amqp_client.broker
     yield self.worker.startWorker()
     self.worker_url = self.worker.get_transport_url()
     self.worker.r_server = FakeRedis()

 def test_send_to_with_bad_config(self):
     config = {'transport_name': 'badconfig_app',
               'send_to': {
                   'default': {},  # missing transport_name
                   'outbound1': {},  # also missing transport_name
                   },
               }
     badcfg_worker = get_stubbed_worker(DummyApplicationWorker,
                                        config=config)
     errors = []
     d = badcfg_worker.startWorker()
     d.addErrback(lambda result: errors.append(result))
     yield d
     self.assertEqual(errors[0].type, ConfigError)

 def make_worker(self, retry_delivery_period=0):
     self.config = {
         "transport_name": "sphex",
         "retry_routing_key": "sms.outbound.%(transport_name)s",
         "failures_routing_key": "sms.failures.%(transport_name)s",
         "retry_delivery_period": retry_delivery_period,
     }
     self.worker = get_stubbed_worker(FailureWorker, self.config)
     self.worker.r_server = FakeRedis()
     self.worker.r_server.flushdb()
     self.redis = self.worker.r_server
     yield self.worker.startWorker()
     self.broker = self.worker._amqp_client.broker
     self.retry_timestamps_key = self.worker._retry_timestamps_key

 def test_subclassing_api(self):
     worker = get_stubbed_worker(ApplicationWorker,
                                 {'transport_name': 'test'})
     ack = TransportEvent(event_type='ack',
                          sent_message_id='remote-id',
                          user_message_id='ack-uuid')
     dr = TransportEvent(event_type='delivery_report',
                         delivery_status='pending',
                         user_message_id='dr-uuid')
     worker.consume_ack(ack)
     worker.consume_delivery_report(dr)
     worker.consume_unknown_event(FakeUserMessage())
     worker.consume_user_message(FakeUserMessage())
     worker.new_session(FakeUserMessage())
     worker.close_session(FakeUserMessage())

 def setUp(self):
     config = {
             "route_mappings": {
                 "test1.inbound": [
                     "test2.outbound",
                     "test3.outbound"],
                 "test2.inbound": [
                     "test1.outbound"],
                 "test3.inbound": [
                     "test1.outbound"]
                 }
         }
     self.worker = get_stubbed_worker(SimpleDispatcher, config)
     self.broker = self.worker._amqp_client.broker
     yield self.worker.startWorker()