def test_get_probe_timeout(self):
        """
        CreateContainer probe times-out if get_probe runs too long.
        """
        clock = Clock()

        node_id = uuid4()
        node = Node(uuid=node_id, public_address=IPAddress('10.0.0.1'))
        control_service = FakeFlockerClient([node], node_id)

        cluster = BenchmarkCluster(
            IPAddress('10.0.0.1'),
            lambda reactor: control_service,
            {},
            None,
        )
        operation = CreateContainer(clock, cluster)
        d = operation.get_probe()

        clock.advance(DEFAULT_TIMEOUT.total_seconds())

        # No control_service.synchronize_state() call, so cluster state
        # never shows container is created.

        # The Deferred fails if container not created within 10 minutes.
        self.failureResultOf(d)

 def test_convergence_error_start_new_iteration(self, logger):
     """
     Even if the convergence fails, a new iteration is started anyway.
     """
     local_state = NodeState(hostname=u'192.0.2.123')
     configuration = Deployment(nodes=frozenset([to_node(local_state)]))
     state = DeploymentState(nodes=[local_state])
     action = ControllableAction(result=fail(RuntimeError()))
     # First discovery succeeds, leading to failing action; second
     # discovery will just wait for Deferred to fire. Thus we expect to
     # finish test in discovery state.
     deployer = ControllableDeployer(
         local_state.hostname,
         [succeed(local_state), Deferred()],
         [action])
     client = self.make_amp_client([local_state])
     reactor = Clock()
     loop = build_convergence_loop_fsm(reactor, deployer)
     self.patch(loop, "logger", logger)
     loop.receive(_ClientStatusUpdate(
         client=client, configuration=configuration, state=state))
     reactor.advance(1.0)
     # Calculating actions happened, result was run and caused error...
     # but we started on loop again and are thus in discovery state,
     # which we can tell because all faked local states have been
     # consumed:
     self.assertEqual(len(deployer.local_states), 0)

    def test_tick(self):
        """
        ``/mimic/v1.1/tick`` (handled by :func:`MimicRoot.advance_time`)
        advances the clock associated with the service.
        """
        clock = Clock()

        def do():
            do.done = True

        do.done = False
        clock.callLater(3.5, do)
        core = MimicCore(clock, [])
        root = MimicRoot(core, clock).app.resource()
        self.assertEqual(do.done, False)
        jreq = json_request(
            self, root, "POST", "/mimic/v1.1/tick", body={"amount": 3.6}
        )
        [response, json_content] = self.successResultOf(jreq)
        self.assertEqual(response.code, 200)
        expected = {
            'advanced': 3.6,
            'now': '1970-01-01T00:00:03.600000Z',
        }
        self.assertEqual(json_content, expected)
        self.assertEqual(do.done, True)

    def test_sendPresence(self):
        clock = Clock()
        xmlStream = StubXmlStream()
        settings = { "ServiceAddress" : "pubsub.example.com", "JID" : "jid",
            "NodeConfiguration" : { "pubsub#deliver_payloads" : "1" },
            "Password" : "password", "KeepAliveSeconds" : 5 }
        notifier = XMPPNotifier(settings, reactor=clock, heartbeat=False)
        factory = XMPPNotificationFactory(notifier, settings, reactor=clock)
        factory.connected(xmlStream)
        factory.authenticated(xmlStream)

        self.assertEquals(len(xmlStream.elements), 2)
        presence = xmlStream.elements[0]
        self.assertEquals(presence.name, "presence")
        iq = xmlStream.elements[1]
        self.assertEquals(iq.name, "iq")

        clock.advance(5)

        self.assertEquals(len(xmlStream.elements), 3)
        presence = xmlStream.elements[2]
        self.assertEquals(presence.name, "presence")

        factory.disconnected(xmlStream)
        clock.advance(5)
        self.assertEquals(len(xmlStream.elements), 3)

    def test_convergence_done_unchanged_notify(self):
        """
        An FSM doing convergence that discovers state unchanged from the last
        state acknowledged by the control service does not re-send that state.
        """
        local_state = NodeState(hostname=u'192.0.2.123')
        configuration = Deployment(nodes=[to_node(local_state)])
        state = DeploymentState(nodes=[local_state])
        deployer = ControllableDeployer(
            local_state.hostname,
            [succeed(local_state), succeed(local_state.copy())],
            [no_action(), no_action()]
        )
        client = self.make_amp_client([local_state])
        reactor = Clock()
        loop = build_convergence_loop_fsm(reactor, deployer)
        loop.receive(_ClientStatusUpdate(
            client=client, configuration=configuration, state=state))
        reactor.advance(1.0)

        # Calculating actions happened, result was run... and then we did
        # whole thing again:
        self.assertEqual(
            (deployer.calculate_inputs, client.calls),
            (
                # Check that the loop has run twice
                [(local_state, configuration, state),
                 (local_state, configuration, state)],
                # But that state was only sent once.
                [(NodeStateCommand, dict(state_changes=(local_state,)))],
            )
        )

    def assert_mutate_function_retries_until_success(
            self, mutate_callable, expected_args, success_response,
            expected_result):
        """
        Assert that some CLB function that mutates the CLB will retry on
        pending update until the function succeeds.

        :param mutate_callable: a callable which takes a clb argument and
            a clock argument - this callable should call the CLB's mutate
            function with the required arguments and return the function's
            return value.  For example:
            ``lambda clb, clk: clb.update_node(..., clock=clk)``
        :param expected_args: What are the expected treq arguments?  This
            should be an array of
            [method, url, (expected args, expected kwargs)]
        :param success_response: a tuple of (Response, string response body)
            which should be the successful response back from the API
        :param expected_result: What is the expected successful result of the
            function that is called by ``mutate_callable``
        """
        clock = Clock()
        clb = self.get_clb(*(expected_args + pending_update_response))

        d = mutate_callable(clb, clock)

        self.assertNoResult(d)
        clock.pump([3])
        self.assertNoResult(d)

        clb.treq = get_fake_treq(
            *([self] + expected_args + [success_response]))

        clock.pump([3])
        self.assertEqual(self.successResultOf(d), expected_result)

    def test_timeout(self):
        request_body = (
            "<ENQRequest>"
            "<requestId>0</requestId>"
            "<enqCmd>ENQUIRELINK</enqCmd>"
            "</ENQRequest>")
        expected_request_packet = utils.mk_packet('0', request_body)

        clock = Clock()
        self.client.clock = clock
        self.client.enquire_link_interval = 120
        self.client.timeout_period = 20
        self.client.authenticated = True
        self.client.start_periodic_enquire_link()

        # wait for the first enquire link request
        received_request_packet = yield self.server.wait_for_data()
        self.assertEqual(expected_request_packet, received_request_packet)

        # advance to just before the timeout should occur
        clock.advance(19.9)
        self.assertFalse(self.client.disconnected)

        # advance to just after the timeout should occur
        clock.advance(0.1)
        self.assertTrue(self.client.disconnected)
        self.assert_in_log(
            'msg',
            "No enquire link response received after 20 seconds, "
            "disconnecting")

    def test_limited_exceptions(self):
        """
        By default, ``retry_failure`` retries on any exception. However, if
        it's given an iterable of expected exception types (exactly as one
        might pass to ``Failure.check``), then it will only retry if one of
        *those* exceptions is raised.
        """
        steps = [0.1, 0.2]

        result = object()
        type_error = Failure(TypeError("bad type"))

        results = [
            Failure(ValueError("bad value")),
            type_error,
            succeed(result),
        ]

        def function():
            return results.pop(0)

        clock = Clock()

        d = retry_failure(clock, function, expected=[ValueError], steps=steps)
        self.assertNoResult(d)

        clock.advance(0.1)
        self.assertEqual(self.failureResultOf(d), type_error)

    def test_iterates(self, logger):
        """
        If the predicate returns something falsey followed by something truthy,
        then ``loop_until`` returns it immediately.
        """
        result = object()
        results = [None, result]

        def predicate():
            return results.pop(0)
        clock = Clock()

        d = loop_until(clock, predicate)

        self.assertNoResult(d)

        clock.advance(0.1)
        self.assertEqual(
            self.successResultOf(d),
            result)

        action = LoggedAction.of_type(logger.messages, LOOP_UNTIL_ACTION)[0]
        assertContainsFields(self, action.start_message, {
            'predicate': predicate,
        })
        assertContainsFields(self, action.end_message, {
            'result': result,
        })
        self.assertTrue(action.succeeded)
        message = LoggedMessage.of_type(
            logger.messages, LOOP_UNTIL_ITERATION_MESSAGE)[0]
        self.assertEqual(action.children, [message])
        assertContainsFields(self, message.message, {
            'result': None,
        })

    def test_convergence_done_delays_new_iteration_ack(self):
        """
        A state update isn't sent if the control node hasn't acknowledged the
        last state update.
        """
        self.local_state = local_state = NodeState(hostname=u'192.0.2.123')
        self.configuration = configuration = Deployment()
        self.cluster_state = received_state = DeploymentState(nodes=[])
        self.action = action = ControllableAction(result=succeed(None))
        deployer = ControllableDeployer(
            local_state.hostname, [succeed(local_state)], [action]
        )
        client = self.make_amp_client([local_state])
        reactor = Clock()
        loop = build_convergence_loop_fsm(reactor, deployer)
        loop.receive(_ClientStatusUpdate(
            # We don't want to receive the acknowledgment of the
            # state update.
            client=DelayedAMPClient(client),
            configuration=configuration,
            state=received_state))

        # Wait for the delay in the convergence loop to pass.  This won't do
        # anything, since we are also waiting for state to be acknowledged.
        reactor.advance(1.0)

        # Only one status update was sent.
        self.assertListEqual(
            client.calls,
            [(NodeStateCommand, dict(state_changes=(local_state,)))],
        )

    def test_too_many_iterations(self):
        """
        If ``retry_failure`` fails more times than there are steps provided, it
        errors back with the last failure.
        """
        steps = [0.1]

        result = object()
        failure = Failure(ValueError("really bad value"))

        results = [
            Failure(ValueError("bad value")),
            failure,
            succeed(result),
        ]

        def function():
            return results.pop(0)

        clock = Clock()

        d = retry_failure(clock, function, steps=steps)
        self.assertNoResult(d)

        clock.advance(0.1)
        self.assertEqual(self.failureResultOf(d), failure)

    def test_taskProductionFailed_deferred_doesnt_delay_polling(self):
        # If taskProductionFailed returns a deferred, we don't wait for it to
        # fire before polling again.
        class DeferredFailingConsumer(NoopTaskConsumer):
            def taskProductionFailed(self, reason):
                failures.append(reason)
                return Deferred()

        interval = self.factory.getUniqueInteger()
        clock = Clock()
        produced = []
        failures = []

        def producer():
            exc = RuntimeError()
            produced.append(exc)
            raise exc

        task_source = self.makeTaskSource(task_producer=producer, interval=interval, clock=clock)
        consumer = DeferredFailingConsumer()
        task_source.start(consumer)
        # The call to start polls once and taskProductionFailed is called.
        self.assertEqual((1, 1), (len(produced), len(failures)))
        # Even though taskProductionFailed returned a deferred which has not
        # yet fired, we poll again after 'interval' seconds.
        clock.advance(interval)
        self.assertEqual((2, 2), (len(produced), len(failures)))

    def test_taskStarted_deferred_doesnt_delay_polling(self):
        # If taskStarted returns a deferred, we don't wait for it to fire
        # before polling again.
        class DeferredStartingConsumer(NoopTaskConsumer):
            def taskStarted(self, task):
                started.append(task)
                return Deferred()

        interval = self.factory.getUniqueInteger()
        clock = Clock()
        produced = []
        started = []

        def producer():
            value = self.factory.getUniqueInteger()
            produced.append(value)
            return value

        task_source = self.makeTaskSource(task_producer=producer, interval=interval, clock=clock)
        consumer = DeferredStartingConsumer()
        task_source.start(consumer)
        # The call to start polls once and taskStarted is called.
        self.assertEqual((1, 1), (len(produced), len(started)))
        # Even though taskStarted returned a deferred which has not yet fired,
        # we poll again after 'interval' seconds.
        clock.advance(interval)
        self.assertEqual((2, 2), (len(produced), len(started)))

    def test_only_one_producer_call_at_once(self):
        # If the task producer returns a Deferred, it will not be called again
        # until that deferred has fired, even if takes longer than the
        # interval we're polling at.
        tasks_called = []
        produced_deferreds = []

        def producer():
            deferred = Deferred()
            produced_deferreds.append(deferred)
            return deferred

        clock = Clock()
        interval = self.factory.getUniqueInteger()
        task_source = self.makeTaskSource(task_producer=producer, interval=interval, clock=clock)
        task_source.start(AppendingTaskConsumer(tasks_called))
        # The call to start calls producer.  It returns a deferred which has
        # not been fired.
        self.assertEqual(len(produced_deferreds), 1)
        # If 'interval' seconds passes and the deferred has still not fired
        # the producer is not called again.
        clock.advance(interval)
        self.assertEqual(len(produced_deferreds), 1)
        # If the task-getting deferred is fired and more time passes, we poll
        # again.
        produced_deferreds[0].callback(None)
        clock.advance(interval)
        self.assertEqual(len(produced_deferreds), 2)

    def test_scenario_throws_exception_when_rate_drops(self):
        """
        ReadRequestLoadScenario raises RequestRateTooLow if rate
        drops below the requested rate.

        Establish the requested rate by having the FakeFlockerClient
        respond to all requests, then lower the rate by dropping
        alternate requests. This should result in RequestRateTooLow
        being raised.
        """
        c = Clock()

        cluster = self.make_cluster(RequestDroppingFakeFlockerClient)
        sample_size = 5
        s = ReadRequestLoadScenario(c, cluster, sample_size=sample_size)

        s.start()

        # Advance the clock by `sample_size` seconds to establish the
        # requested rate.
        c.pump(repeat(1, sample_size))

        cluster.get_control_service(c).drop_requests = True

        # Advance the clock by 2 seconds so that a request is dropped
        # and a new rate which is below the target can be established.
        c.advance(2)

        failure = self.failureResultOf(s.maintained())
        self.assertIsInstance(failure.value, RequestRateTooLow)

 def test_convergence_done_start_new_iteration(self):
     """
     After a short delay, an FSM completing the changes from one convergence
     iteration starts another iteration.
     """
     local_state = NodeState(hostname=b'192.0.2.123')
     local_state2 = NodeState(hostname=b'192.0.2.123')
     configuration = Deployment(nodes=frozenset([to_node(local_state)]))
     state = DeploymentState(nodes=[local_state])
     action = ControllableAction(result=succeed(None))
     # Because the second action result is unfired Deferred, the second
     # iteration will never finish; applying its changes waits for this
     # Deferred to fire.
     action2 = ControllableAction(result=Deferred())
     deployer = ControllableDeployer(
         [succeed(local_state), succeed(local_state2)],
         [action, action2])
     client = self.successful_amp_client([local_state, local_state2])
     reactor = Clock()
     loop = build_convergence_loop_fsm(reactor, deployer)
     loop.receive(_ClientStatusUpdate(
         client=client, configuration=configuration, state=state))
     reactor.advance(1.0)
     # Calculating actions happened, result was run... and then we did
     # whole thing again:
     self.assertEqual((deployer.calculate_inputs, client.calls),
                      ([(local_state, configuration, state),
                        (local_state2, configuration, state)],
                       [(NodeStateCommand, dict(node_state=local_state)),
                        (NodeStateCommand, dict(node_state=local_state2))]))

    def assert_mutate_function_retries_until_timeout(
            self, mutate_callable, expected_args, timeout=60):
        """
        Assert that some CLB function that mutates the CLB will retry on
        pending update until the function times out.

        :param mutate_callable: a callable which takes a clb argument and
            a clock argument - this callable should call the CLB's mutate
            function with the required arguments and return the function's
            return value.  For example:
            ``lambda clb, clk: clb.update_node(..., clock=clk)``
        :param expected_args: What are the expected treq arguments?  This
            should be an array of
            [method, url, (expected args, expected kwargs)]
        :param int timeout: When does your function time out retrying?
        """
        clock = Clock()
        clb = self.get_clb(*(expected_args + pending_update_response))

        d = mutate_callable(clb, clock)
        self.assertNoResult(d)

        for _ in range((timeout - 1) / 3):
            clock.pump([3])
            self.assertNoResult(d)

        clock.pump([3])
        self.failureResultOf(d, TimedOutError)

    def test_ignoreAlreadyAccepting(self):
        """
        If the client sees an event change a second time before
        responding to an invitation found on it during the first
        change notification, the second change notification does not
        generate another accept attempt.
        """
        clock = Clock()
        randomDelay = 7
        vevent = Component.fromString(INVITED_EVENT)
        attendees = tuple(vevent.mainComponent().properties('ATTENDEE'))
        userNumber = int(attendees[1].parameterValue('CN').split(None, 1)[1])
        calendarURL = '/some/calendar/'
        calendar = Calendar(
            caldavxml.calendar, set(('VEVENT',)), u'calendar', calendarURL, None)
        client = StubClient(userNumber, self.mktemp())
        client._calendars[calendarURL] = calendar
        event = Event(client.serializeLocation(), calendarURL + u'1234.ics', None, vevent)
        client._events[event.url] = event
        accepter = Accepter(clock, self.sim, client, userNumber)
        accepter.random = Deterministic()

        def _gauss(mu, sigma):
            return randomDelay
        accepter.random.gauss = _gauss
        accepter.eventChanged(event.url)
        accepter.eventChanged(event.url)
        clock.advance(randomDelay)

    def test04_get_only_valid_requests(self):
        """
        - create a lot of requests marked as 'expired'
        - wait some time
        - create another lot of requests marked as 'valid'
        -> check if only 'valid' requests present
        """

        clock = Clock()

        sessions = Sessions(False, 10, clock)
        collector = Collector(sessions)

        dl = []
        for i in xrange(10):
            d = collector.queue_and_process("192.168.45.12", "expired")
            dl.append(d)

        clock.advance(15)

        for i in xrange(10):
            d = collector.queue_and_process("192.168.45.12", "valid")
            dl.append(d)


        dfl = DeferredList(dl)
        @dfl.addCallback
        def get_result(ignored):

            for i in xrange(10):
                uid, ip, request = collector.get()
                self.assertEqual(request, "valid")

    def test_inboxReplyFailedDelete(self):
        """
        When an inbox item that contains a reply is seen by the client, it
        deletes it immediately.  If the delete fails, the appropriate response
        code is returned.
        """
        userNumber = 1
        clock = Clock()
        inboxURL = '/some/inbox/'
        vevent = Component.fromString(INBOX_REPLY)
        inbox = Calendar(
            caldavxml.schedule_inbox, set(), u'the inbox', inboxURL, None)
        client = StubClient(userNumber, self.mktemp())
        client._calendars[inboxURL] = inbox

        inboxEvent = Event(client.serializeLocation(), inboxURL + u'4321.ics', None, vevent)
        client._setEvent(inboxEvent.url, inboxEvent)
        client._failDeleteWithObject(inboxEvent.url, IncorrectResponseCode(
            NO_CONTENT,
            Response(
                ('HTTP', 1, 1), PRECONDITION_FAILED,
                'Precondition Failed', None, None))
        )
        accepter = Accepter(clock, self.sim, client, userNumber)
        accepter.eventChanged(inboxEvent.url)
        clock.advance(3)
        self.assertNotIn(inboxEvent.url, client._events)
        self.assertNotIn('4321.ics', inbox.events)