<commit_before># Copyright Hybrid Logic Ltd.  See LICENSE file for details.

"""
Tests for ``flocker.node._loop``.
"""

from uuid import uuid4

from eliot.testing import validate_logging, assertHasAction, assertHasMessage
from machinist import LOG_FSM_TRANSITION

from pyrsistent import pset

from twisted.trial.unittest import SynchronousTestCase
from twisted.test.proto_helpers import StringTransport, MemoryReactorClock
from twisted.internet.protocol import Protocol, ReconnectingClientFactory
from twisted.internet.defer import succeed, Deferred, fail
from twisted.internet.task import Clock
from twisted.internet.ssl import ClientContextFactory
from twisted.protocols.tls import TLSMemoryBIOFactory, TLSMemoryBIOProtocol

from ...testtools import FakeAMPClient
from .._loop import (
    build_cluster_status_fsm, ClusterStatusInputs, _ClientStatusUpdate,
    _StatusUpdate, _ConnectedToControlService, ConvergenceLoopInputs,
    ConvergenceLoopStates, build_convergence_loop_fsm, AgentLoopService,
    LOG_SEND_TO_CONTROL_SERVICE,
    LOG_CONVERGE, LOG_CALCULATED_ACTIONS,
    )
from ..testtools import ControllableDeployer, ControllableAction, to_node
from ...control import (
    NodeState, Deployment, Manifestation, Dataset, DeploymentState,
)
from ...control._protocol import NodeStateCommand, AgentAMP
from ...control.test.test_protocol import iconvergence_agent_tests_factory


def build_protocol():
    """
    :return: ``Protocol`` hooked up to transport.
    """
    p = Protocol()
    p.makeConnection(StringTransport())
    return p


class StubFSM(object):
    """
    A finite state machine look-alike that just records inputs.
    """
    def __init__(self):
        self.inputted = []

    def receive(self, symbol):
        self.inputted.append(symbol)


class ClusterStatusFSMTests(SynchronousTestCase):
    """
    Tests for the cluster status FSM.
    """
    def setUp(self):
        self.convergence_loop = StubFSM()
        self.fsm = build_cluster_status_fsm(self.convergence_loop)

    def assertConvergenceLoopInputted(self, expected):
        """
        Assert that that given set of symbols were input to the agent
        operation FSM.
        """
        self.assertEqual(self.convergence_loop.inputted, expected)

    def test_creation_no_side_effects(self):
        """
        Creating the FSM has no side effects.
        """
        self.assertConvergenceLoopInputted([])

    def test_first_status_update(self):
        """
        Once the client has been connected and a status update received it
        notifies the convergence loop FSM of this.
        """
        client = object()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        self.assertConvergenceLoopInputted(
            [_ClientStatusUpdate(client=client, configuration=desired,
                                 state=state)])

    def test_second_status_update(self):
        """
        Further status updates are also passed to the convergence loop FSM.
        """
        client = object()
        desired1 = object()
        state1 = object()
        desired2 = object()
        state2 = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        # Initially some other status:
        self.fsm.receive(_StatusUpdate(configuration=desired1, state=state1))
        self.fsm.receive(_StatusUpdate(configuration=desired2, state=state2))
        self.assertConvergenceLoopInputted(
            [_ClientStatusUpdate(client=client, configuration=desired1,
                                 state=state1),
             _ClientStatusUpdate(client=client, configuration=desired2,
                                 state=state2)])

    def test_status_update_no_disconnect(self):
        """
        Neither new connections nor status updates cause the client to be
        disconnected.
        """
        client = build_protocol()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=object(),
                                       state=object()))
        self.assertFalse(client.transport.disconnecting)

    def test_disconnect_before_status_update(self):
        """
        If the client disconnects before a status update is received then no
        notification is needed for convergence loop FSM.
        """
        self.fsm.receive(_ConnectedToControlService(client=build_protocol()))
        self.fsm.receive(ClusterStatusInputs.DISCONNECTED_FROM_CONTROL_SERVICE)
        self.assertConvergenceLoopInputted([])

    def test_disconnect_after_status_update(self):
        """
        If the client disconnects after a status update is received then the
        convergence loop FSM is notified.
        """
        client = object()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        self.fsm.receive(ClusterStatusInputs.DISCONNECTED_FROM_CONTROL_SERVICE)
        self.assertConvergenceLoopInputted(
            [_ClientStatusUpdate(client=client, configuration=desired,
                                 state=state),
             ConvergenceLoopInputs.STOP])

    def test_status_update_after_reconnect(self):
        """
        If the client disconnects, reconnects, and a new status update is
        received then the convergence loop FSM is notified.
        """
        client = object()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        self.fsm.receive(ClusterStatusInputs.DISCONNECTED_FROM_CONTROL_SERVICE)
        client2 = object()
        desired2 = object()
        state2 = object()
        self.fsm.receive(_ConnectedToControlService(client=client2))
        self.fsm.receive(_StatusUpdate(configuration=desired2, state=state2))
        self.assertConvergenceLoopInputted(
            [_ClientStatusUpdate(client=client, configuration=desired,
                                 state=state),
             ConvergenceLoopInputs.STOP,
             _ClientStatusUpdate(client=client2, configuration=desired2,
                                 state=state2)])

    def test_shutdown_before_connect(self):
        """
        If the FSM is shutdown before a connection is made nothing happens.
        """
        self.fsm.receive(ClusterStatusInputs.SHUTDOWN)
        self.assertConvergenceLoopInputted([])

    def test_shutdown_after_connect(self):
        """
        If the FSM is shutdown after connection but before status update is
        received then it disconnects but does not notify the agent
        operation FSM.
        """
        client = build_protocol()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(ClusterStatusInputs.SHUTDOWN)
        self.assertEqual((client.transport.disconnecting,
                          self.convergence_loop.inputted),
                         (True, []))

    def test_shutdown_after_status_update(self):
        """
        If the FSM is shutdown after connection and status update is received
        then it disconnects and also notifys the convergence loop FSM that
        is should stop.
        """
        client = build_protocol()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        self.fsm.receive(ClusterStatusInputs.SHUTDOWN)
        self.assertEqual((client.transport.disconnecting,
                          self.convergence_loop.inputted[-1]),
                         (True, ConvergenceLoopInputs.STOP))

    def test_shutdown_fsm_ignores_disconnection(self):
        """
        If the FSM has been shutdown it ignores disconnection event.
        """
        client = build_protocol()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        self.fsm.receive(ClusterStatusInputs.SHUTDOWN)
        self.fsm.receive(ClusterStatusInputs.DISCONNECTED_FROM_CONTROL_SERVICE)
        self.assertConvergenceLoopInputted([
            _ClientStatusUpdate(client=client, configuration=desired,
                                state=state),
            # This is caused by the shutdown... and the disconnect results
            # in no further messages:
            ConvergenceLoopInputs.STOP])

    def test_shutdown_fsm_ignores_cluster_status(self):
        """
        If the FSM has been shutdown it ignores cluster status update.
        """
        client = build_protocol()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(ClusterStatusInputs.SHUTDOWN)
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        # We never send anything to convergence loop FSM:
        self.assertConvergenceLoopInputted([])


class ConvergenceLoopFSMTests(SynchronousTestCase):
    """
    Tests for FSM created by ``build_convergence_loop_fsm``.
    """
    def test_new_stopped(self):
        """
        A newly created FSM is stopped.
        """
        loop = build_convergence_loop_fsm(
            Clock(), ControllableDeployer(u"192.168.1.1", [], [])
        )
        self.assertEqual(loop.state, ConvergenceLoopStates.STOPPED)

    def test_new_status_update_starts_discovery(self):
        """
        A stopped FSM that receives a status update starts discovery.
        """
        deployer = ControllableDeployer(u"192.168.1.1", [Deferred()], [])
        loop = build_convergence_loop_fsm(Clock(), deployer)
        loop.receive(_ClientStatusUpdate(client=FakeAMPClient(),
                                         configuration=Deployment(),
                                         state=DeploymentState()))
        self.assertEqual(len(deployer.local_states), 0)  # Discovery started

    def make_amp_client(self, local_states, flaky=False):
        """
        Create AMP client that can respond successfully to a
        ``NodeStateCommand``.

        :param local_states: The node states we expect to be able to send.
        :param Bool health_status: Flag to indicate whether the client should
            succeed or fail.

        :return FakeAMPClient: Fake AMP client appropriately setup.
        """
        client = FakeAMPClient(flaky)
        for local_state in local_states:
            client.register_response(
                NodeStateCommand, dict(state_changes=(local_state,)),
                {"result": None})
        return client

    @validate_logging(assertHasAction, LOG_SEND_TO_CONTROL_SERVICE, True)
    def test_convergence_done_notify(self, logger):
        """
        A FSM doing convergence that gets a discovery result, sends the
        discovered state to the control service using the last received
        client.
        """
        local_state = NodeState(hostname=u"192.0.2.123")
        client = self.make_amp_client([local_state])
        action = ControllableAction(result=Deferred())
        deployer = ControllableDeployer(
            local_state.hostname, [succeed(local_state)], [action]
        )
        loop = build_convergence_loop_fsm(Clock(), deployer)
        self.patch(loop, "logger", logger)
        loop.receive(
            _ClientStatusUpdate(
                client=client,
                configuration=Deployment(
                    nodes=frozenset([to_node(local_state)])
                ),
                state=DeploymentState(
                    nodes=frozenset([local_state])
                )
            )
        )
        self.assertEqual(client.calls, [(NodeStateCommand,
                                         dict(state_changes=(local_state,)))])

    def test_convergence_done_unchanged_notify(self):
        """
        A FSM doing convergence that gets a discovery result that is unchanged
        from the last time it sent data does not send the discovered state to
        the control service.
        """
        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=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(
            local_state.hostname,
            [succeed(local_state), succeed(local_state)],
            [action, action2])
        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 test_convergence_done_changed_notify(self):
        """
        A FSM doing convergence that gets a discovery result that is changed
        from the last time it sent data does send the discoverd state to
        the control service.
        """
        local_state = NodeState(hostname=u'192.0.2.123')
        local_state2 = NodeState(
            hostname=u'192.0.2.123', used_ports=pset([80]),
            applications=pset())
        configuration = Deployment(nodes=frozenset([to_node(local_state)]))
        state = DeploymentState(nodes=[local_state])
        state2 = DeploymentState(nodes=[local_state2])
        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(
            local_state.hostname,
            [succeed(local_state), succeed(local_state2)],
            [action, action2])
        client = self.make_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),
            (
                # Check that the loop has run twice
                [(local_state, configuration, state),
                 (local_state2, configuration, state2)],
                # And the state was sent twice
                [(NodeStateCommand, dict(state_changes=(local_state,))),
                 (NodeStateCommand, dict(state_changes=(local_state2,)))],
            )
        )

    def test_converence_sent_state_fail_resends(self):
        """
        If sending state toe the control node fails, the next iteration will
        send state even if the state hasn't changed.
        """
        local_state = NodeState(hostname=u'192.0.2.123')
        local_state2 = NodeState(hostname=u'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(
            local_state.hostname,
            [succeed(local_state), succeed(local_state2)],
            [action, action2])
        client = self.make_amp_client([local_state, local_state2], True)
        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),<commit_msg>pyrsistent is happy to do the conversion for us<commit_after># Copyright Hybrid Logic Ltd.  See LICENSE file for details.

"""
Tests for ``flocker.node._loop``.
"""

from uuid import uuid4

from eliot.testing import validate_logging, assertHasAction, assertHasMessage
from machinist import LOG_FSM_TRANSITION

from pyrsistent import pset

from twisted.trial.unittest import SynchronousTestCase
from twisted.test.proto_helpers import StringTransport, MemoryReactorClock
from twisted.internet.protocol import Protocol, ReconnectingClientFactory
from twisted.internet.defer import succeed, Deferred, fail
from twisted.internet.task import Clock
from twisted.internet.ssl import ClientContextFactory
from twisted.protocols.tls import TLSMemoryBIOFactory, TLSMemoryBIOProtocol

from ...testtools import FakeAMPClient
from .._loop import (
    build_cluster_status_fsm, ClusterStatusInputs, _ClientStatusUpdate,
    _StatusUpdate, _ConnectedToControlService, ConvergenceLoopInputs,
    ConvergenceLoopStates, build_convergence_loop_fsm, AgentLoopService,
    LOG_SEND_TO_CONTROL_SERVICE,
    LOG_CONVERGE, LOG_CALCULATED_ACTIONS,
    )
from ..testtools import ControllableDeployer, ControllableAction, to_node
from ...control import (
    NodeState, Deployment, Manifestation, Dataset, DeploymentState,
)
from ...control._protocol import NodeStateCommand, AgentAMP
from ...control.test.test_protocol import iconvergence_agent_tests_factory


def build_protocol():
    """
    :return: ``Protocol`` hooked up to transport.
    """
    p = Protocol()
    p.makeConnection(StringTransport())
    return p


class StubFSM(object):
    """
    A finite state machine look-alike that just records inputs.
    """
    def __init__(self):
        self.inputted = []

    def receive(self, symbol):
        self.inputted.append(symbol)


class ClusterStatusFSMTests(SynchronousTestCase):
    """
    Tests for the cluster status FSM.
    """
    def setUp(self):
        self.convergence_loop = StubFSM()
        self.fsm = build_cluster_status_fsm(self.convergence_loop)

    def assertConvergenceLoopInputted(self, expected):
        """
        Assert that that given set of symbols were input to the agent
        operation FSM.
        """
        self.assertEqual(self.convergence_loop.inputted, expected)

    def test_creation_no_side_effects(self):
        """
        Creating the FSM has no side effects.
        """
        self.assertConvergenceLoopInputted([])

    def test_first_status_update(self):
        """
        Once the client has been connected and a status update received it
        notifies the convergence loop FSM of this.
        """
        client = object()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        self.assertConvergenceLoopInputted(
            [_ClientStatusUpdate(client=client, configuration=desired,
                                 state=state)])

    def test_second_status_update(self):
        """
        Further status updates are also passed to the convergence loop FSM.
        """
        client = object()
        desired1 = object()
        state1 = object()
        desired2 = object()
        state2 = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        # Initially some other status:
        self.fsm.receive(_StatusUpdate(configuration=desired1, state=state1))
        self.fsm.receive(_StatusUpdate(configuration=desired2, state=state2))
        self.assertConvergenceLoopInputted(
            [_ClientStatusUpdate(client=client, configuration=desired1,
                                 state=state1),
             _ClientStatusUpdate(client=client, configuration=desired2,
                                 state=state2)])

    def test_status_update_no_disconnect(self):
        """
        Neither new connections nor status updates cause the client to be
        disconnected.
        """
        client = build_protocol()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=object(),
                                       state=object()))
        self.assertFalse(client.transport.disconnecting)

    def test_disconnect_before_status_update(self):
        """
        If the client disconnects before a status update is received then no
        notification is needed for convergence loop FSM.
        """
        self.fsm.receive(_ConnectedToControlService(client=build_protocol()))
        self.fsm.receive(ClusterStatusInputs.DISCONNECTED_FROM_CONTROL_SERVICE)
        self.assertConvergenceLoopInputted([])

    def test_disconnect_after_status_update(self):
        """
        If the client disconnects after a status update is received then the
        convergence loop FSM is notified.
        """
        client = object()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        self.fsm.receive(ClusterStatusInputs.DISCONNECTED_FROM_CONTROL_SERVICE)
        self.assertConvergenceLoopInputted(
            [_ClientStatusUpdate(client=client, configuration=desired,
                                 state=state),
             ConvergenceLoopInputs.STOP])

    def test_status_update_after_reconnect(self):
        """
        If the client disconnects, reconnects, and a new status update is
        received then the convergence loop FSM is notified.
        """
        client = object()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        self.fsm.receive(ClusterStatusInputs.DISCONNECTED_FROM_CONTROL_SERVICE)
        client2 = object()
        desired2 = object()
        state2 = object()
        self.fsm.receive(_ConnectedToControlService(client=client2))
        self.fsm.receive(_StatusUpdate(configuration=desired2, state=state2))
        self.assertConvergenceLoopInputted(
            [_ClientStatusUpdate(client=client, configuration=desired,
                                 state=state),
             ConvergenceLoopInputs.STOP,
             _ClientStatusUpdate(client=client2, configuration=desired2,
                                 state=state2)])

    def test_shutdown_before_connect(self):
        """
        If the FSM is shutdown before a connection is made nothing happens.
        """
        self.fsm.receive(ClusterStatusInputs.SHUTDOWN)
        self.assertConvergenceLoopInputted([])

    def test_shutdown_after_connect(self):
        """
        If the FSM is shutdown after connection but before status update is
        received then it disconnects but does not notify the agent
        operation FSM.
        """
        client = build_protocol()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(ClusterStatusInputs.SHUTDOWN)
        self.assertEqual((client.transport.disconnecting,
                          self.convergence_loop.inputted),
                         (True, []))

    def test_shutdown_after_status_update(self):
        """
        If the FSM is shutdown after connection and status update is received
        then it disconnects and also notifys the convergence loop FSM that
        is should stop.
        """
        client = build_protocol()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        self.fsm.receive(ClusterStatusInputs.SHUTDOWN)
        self.assertEqual((client.transport.disconnecting,
                          self.convergence_loop.inputted[-1]),
                         (True, ConvergenceLoopInputs.STOP))

    def test_shutdown_fsm_ignores_disconnection(self):
        """
        If the FSM has been shutdown it ignores disconnection event.
        """
        client = build_protocol()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        self.fsm.receive(ClusterStatusInputs.SHUTDOWN)
        self.fsm.receive(ClusterStatusInputs.DISCONNECTED_FROM_CONTROL_SERVICE)
        self.assertConvergenceLoopInputted([
            _ClientStatusUpdate(client=client, configuration=desired,
                                state=state),
            # This is caused by the shutdown... and the disconnect results
            # in no further messages:
            ConvergenceLoopInputs.STOP])

    def test_shutdown_fsm_ignores_cluster_status(self):
        """
        If the FSM has been shutdown it ignores cluster status update.
        """
        client = build_protocol()
        desired = object()
        state = object()
        self.fsm.receive(_ConnectedToControlService(client=client))
        self.fsm.receive(ClusterStatusInputs.SHUTDOWN)
        self.fsm.receive(_StatusUpdate(configuration=desired, state=state))
        # We never send anything to convergence loop FSM:
        self.assertConvergenceLoopInputted([])


class ConvergenceLoopFSMTests(SynchronousTestCase):
    """
    Tests for FSM created by ``build_convergence_loop_fsm``.
    """
    def test_new_stopped(self):
        """
        A newly created FSM is stopped.
        """
        loop = build_convergence_loop_fsm(
            Clock(), ControllableDeployer(u"192.168.1.1", [], [])
        )
        self.assertEqual(loop.state, ConvergenceLoopStates.STOPPED)

    def test_new_status_update_starts_discovery(self):
        """
        A stopped FSM that receives a status update starts discovery.
        """
        deployer = ControllableDeployer(u"192.168.1.1", [Deferred()], [])
        loop = build_convergence_loop_fsm(Clock(), deployer)
        loop.receive(_ClientStatusUpdate(client=FakeAMPClient(),
                                         configuration=Deployment(),
                                         state=DeploymentState()))
        self.assertEqual(len(deployer.local_states), 0)  # Discovery started

    def make_amp_client(self, local_states, flaky=False):
        """
        Create AMP client that can respond successfully to a
        ``NodeStateCommand``.

        :param local_states: The node states we expect to be able to send.
        :param Bool health_status: Flag to indicate whether the client should
            succeed or fail.

        :return FakeAMPClient: Fake AMP client appropriately setup.
        """
        client = FakeAMPClient(flaky)
        for local_state in local_states:
            client.register_response(
                NodeStateCommand, dict(state_changes=(local_state,)),
                {"result": None})
        return client

    @validate_logging(assertHasAction, LOG_SEND_TO_CONTROL_SERVICE, True)
    def test_convergence_done_notify(self, logger):
        """
        A FSM doing convergence that gets a discovery result, sends the
        discovered state to the control service using the last received
        client.
        """
        local_state = NodeState(hostname=u"192.0.2.123")
        client = self.make_amp_client([local_state])
        action = ControllableAction(result=Deferred())
        deployer = ControllableDeployer(
            local_state.hostname, [succeed(local_state)], [action]
        )
        loop = build_convergence_loop_fsm(Clock(), deployer)
        self.patch(loop, "logger", logger)
        loop.receive(
            _ClientStatusUpdate(
                client=client,
                configuration=Deployment(
                    nodes=frozenset([to_node(local_state)])
                ),
                state=DeploymentState(
                    nodes=frozenset([local_state])
                )
            )
        )
        self.assertEqual(client.calls, [(NodeStateCommand,
                                         dict(state_changes=(local_state,)))])

    def test_convergence_done_unchanged_notify(self):
        """
        A FSM doing convergence that gets a discovery result that is unchanged
        from the last time it sent data does not send the discovered state to
        the control service.
        """
        local_state = NodeState(hostname=u'192.0.2.123')
        configuration = Deployment(nodes=[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(
            local_state.hostname,
            [succeed(local_state), succeed(local_state)],
            [action, action2])
        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 test_convergence_done_changed_notify(self):
        """
        A FSM doing convergence that gets a discovery result that is changed
        from the last time it sent data does send the discoverd state to
        the control service.
        """
        local_state = NodeState(hostname=u'192.0.2.123')
        local_state2 = NodeState(
            hostname=u'192.0.2.123', used_ports=pset([80]),
            applications=pset())
        configuration = Deployment(nodes=[to_node(local_state)])
        state = DeploymentState(nodes=[local_state])
        state2 = DeploymentState(nodes=[local_state2])
        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(
            local_state.hostname,
            [succeed(local_state), succeed(local_state2)],
            [action, action2])
        client = self.make_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),
            (
                # Check that the loop has run twice
                [(local_state, configuration, state),
                 (local_state2, configuration, state2)],
                # And the state was sent twice
                [(NodeStateCommand, dict(state_changes=(local_state,))),
                 (NodeStateCommand, dict(state_changes=(local_state2,)))],
            )
        )

    def test_converence_sent_state_fail_resends(self):
        """
        If sending state toe the control node fails, the next iteration will
        send state even if the state hasn't changed.
        """
        local_state = NodeState(hostname=u'192.0.2.123')
        local_state2 = NodeState(hostname=u'192.0.2.123')
        configuration = Deployment(nodes=[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(
            local_state.hostname,
            [succeed(local_state), succeed(local_state2)],
            [action, action2])
        client = self.make_amp_client([local_state, local_state2], True)
        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),
{'fraction_non_alphanumeric': 0.08379579858790714, 'fraction_numerical': 0.006566522358136967, 'mean_word_length': 2.5391259640102826, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 16, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3492398', 'n_tokens_mistral': 9282, 'n_tokens_neox': 8932, 'n_words': 2105}