<commit_before># Copyright (c) 2010-2011, Found IT A/S and Piped Project Contributors.
# See LICENSE for details.

import mock
from mock import patch
from pika import frame, exceptions as pika_exceptions
from twisted.application import service
from twisted.python import failure
from twisted.trial import unittest
from twisted.internet import defer, error

from piped import util, processing, dependencies, exceptions, event
from piped.contrib.amqp import providers


class TestConnectionProvider(unittest.TestCase):
    def setUp(self):
        self.runtime_environment = processing.RuntimeEnvironment()
        self.runtime_environment.application = service.MultiService()
        self.runtime_environment.configure()

    def tearDown(self):
        self.runtime_environment.application.stopService()

    def test_provided_connections(self):
        cm = self.runtime_environment.configuration_manager
        cm.set('amqp.connections.test_connection', dict(
            servers = ['tcp:host=localhost:port=5672'],
            max_idle_time = 10,
            reconnect_interval = 5,
            parameters = dict(
                heartbeat = 3
            )
        ))

        provider = providers.AMQPConnectionProvider()
        provider.configure(self.runtime_environment)

        connection_dependency = dependencies.ResourceDependency(provider='amqp.connection.test_connection')
        self.runtime_environment.resource_manager.resolve(connection_dependency)

        with patch.object(providers.endpoints, 'clientFromString') as mocked_client_from_string:
            def connect(factory):
                protocol = factory.buildProtocol(None)
                protocol.transport = mock.Mock()
                # mark the protocol as ready
                protocol.connectionReady(protocol)
                return protocol

            mocked_client_from_string.return_value.connect.side_effect = connect
            provider.startService()

            connection = connection_dependency.get_resource()
            self.assertIsInstance(connection, providers.AMQProtocol)

            self.assertEquals(connection.factory.servers, cm.get('amqp.connections.test_connection.servers'))
            self.assertEquals(connection.factory.reconnect_interval, cm.get('amqp.connections.test_connection.reconnect_interval'))
            self.assertEquals(connection.idle_checker.interval, cm.get('amqp.connections.test_connection.max_idle_time'))
            self.assertEquals(connection.parameters.heartbeat, cm.get('amqp.connections.test_connection.parameters.heartbeat'))

        # if another dependency asks for the same connection, it should receive the same instance
        another_connection_dependency = dependencies.ResourceDependency(provider='amqp.connection.test_connection')
        self.runtime_environment.resource_manager.resolve(another_connection_dependency)

        self.assertIdentical(another_connection_dependency.get_resource(), connection_dependency.get_resource())


class TestMockConnection(unittest.TestCase):

    def setUp(self):
        self.service = service.MultiService()

    @defer.inlineCallbacks
    def tearDown(self):
        yield self.service.stopService()

    def test_servers_config(self):
        # servers are usually specified as a list ...
        rpc_client = providers.AMQPConnection('test_name', servers=['tcp:host=localhost:port=5672'])
        self.assertEquals(rpc_client.get_next_server(), 'tcp:host=localhost:port=5672')

        # ... but it can be a string, which should be wrapped in a list:
        rpc_client = providers.AMQPConnection('test_name', servers='tcp:host=localhost:port=5672')
        self.assertEquals(rpc_client.get_next_server(), 'tcp:host=localhost:port=5672')

    @defer.inlineCallbacks
    def test_get_named_channels(self):
        connection = providers.AMQPConnection('test_name', servers=['tcp:host=server_1:port=5672'])
        connection.setServiceParent(self.service)

        with patch.object(providers.endpoints, 'clientFromString') as mocked_client_from_string:
            def connect(factory):
                protocol = factory.buildProtocol(None)
                protocol.transport = mock.Mock()
                # mark the protocol as ready
                protocol.connectionReady(protocol)
                return protocol

            mocked_client_from_string.return_value.connect.side_effect = connect
            connection.startService()

            # the connection should now be finished connecting:
            self.assertTrue(connection.ready)

            protocol = connection.protocol

            channel_requests = list()
            with patch.object(protocol, 'channel') as mocked_channel:
                def channel():
                    d = defer.Deferred()
                    channel_requests.append(d)
                    return d

                mocked_channel.side_effect = channel

                foo_1_d = protocol.get_named_channel('foo')
                foo_2_d = protocol.get_named_channel('foo')
                foo_3_d = protocol.get_named_channel('foo')

                bar_d = protocol.get_named_channel('bar')

                self.assertEquals(len(channel_requests), 2)

                mocked_foo = mock.Mock()
                mocked_foo_close = event.Event()
                mocked_foo.add_on_close_callback.side_effect = mocked_foo_close.handle
                mocked_bar = mock.Mock()

                channel_requests.pop(0).callback(mocked_foo)
                channel_requests.pop(0).callback(mocked_bar)

                # all the pending requests for the channel should receive the same instance
                foo = yield defer.DeferredList([foo_1_d, foo_2_d, foo_3_d])
                self.assertEquals(set(foo), set([(True, mocked_foo)]))

                bar = yield bar_d
                self.assertEquals(bar, mocked_bar)

                self.assertEquals(len(channel_requests), 0)
                another_foo = yield protocol.get_named_channel('foo')
                self.assertEquals(len(channel_requests), 0)

                # if the channel is closed, the next request to get the channel will recreate it
                mocked_foo_close()
                self.assertEquals(len(channel_requests), 0)
                new_foo = protocol.get_named_channel('foo')
                self.assertEquals(len(channel_requests), 1)

                # if the channel request errbacks, all pending requests for the named channel should errback
                new_foo_2 = protocol.get_named_channel('foo')
                test_exception = Exception()
                channel_requests.pop(0).callback(failure.Failure(test_exception))

                for pending_foo in (new_foo, new_foo_2):
                    try:
                        yield pending_foo
                        self.fail('Expected {0!r} to be raised.'.format(new_foo))
                    except Exception as e:
                        self.assertEquals(e, test_exception)


    def test_connect_disconnect_when_service_starts_stops(self):
        connection = providers.AMQPConnection('test_name', servers=['tcp:host=localhost:port=5672'])

        with patch.object(connection, '_connect') as mocked_connect:
            def verify(endpoint):
                self.assertEquals(endpoint._host, 'localhost')
                self.assertEquals(endpoint._port, 5672)

            mocked_connect.side_effect = verify
            connection.startService()

            mocked_connect.assert_called_once()

        with patch.object(connection, '_disconnect') as mocked_disconnect:
            connection.stopService()

            mocked_disconnect.assert_called_once_with('stopping service')

    @defer.inlineCallbacks
    def test_reconnect_if_connect_fails(self):
        connection = providers.AMQPConnection('test_name', servers=['tcp:host=server_1:port=5672', 'tcp:host=server_2:port=5673'], reconnect_interval=0)
        connection.setServiceParent(self.service)

        # create a list of return values for connect. the last element is a deferred, which enables
        # us to assert that the reconnecting haven't given up.
        endpoints = list()
        connect_return_values = [error.ConnectError('test_error'), error.DNSLookupError('test_error'), error.ConnectionDone('test_error'), Exception('test_exception'), defer.Deferred()]
        expected_errors = list(connect_return_values[:-1])

        disconnects = list()
        connection.on_disconnected += disconnects.append

        with patch.object(connection, '_connect') as mocked_connect:
            def connect(endpoint):
                endpoints.append(dict(host=endpoint._host, port=endpoint._port))
<commit_msg>small typo fix in a comment<commit_after># Copyright (c) 2010-2011, Found IT A/S and Piped Project Contributors.
# See LICENSE for details.

import mock
from mock import patch
from pika import frame, exceptions as pika_exceptions
from twisted.application import service
from twisted.python import failure
from twisted.trial import unittest
from twisted.internet import defer, error

from piped import util, processing, dependencies, exceptions, event
from piped.contrib.amqp import providers


class TestConnectionProvider(unittest.TestCase):
    def setUp(self):
        self.runtime_environment = processing.RuntimeEnvironment()
        self.runtime_environment.application = service.MultiService()
        self.runtime_environment.configure()

    def tearDown(self):
        self.runtime_environment.application.stopService()

    def test_provided_connections(self):
        cm = self.runtime_environment.configuration_manager
        cm.set('amqp.connections.test_connection', dict(
            servers = ['tcp:host=localhost:port=5672'],
            max_idle_time = 10,
            reconnect_interval = 5,
            parameters = dict(
                heartbeat = 3
            )
        ))

        provider = providers.AMQPConnectionProvider()
        provider.configure(self.runtime_environment)

        connection_dependency = dependencies.ResourceDependency(provider='amqp.connection.test_connection')
        self.runtime_environment.resource_manager.resolve(connection_dependency)

        with patch.object(providers.endpoints, 'clientFromString') as mocked_client_from_string:
            def connect(factory):
                protocol = factory.buildProtocol(None)
                protocol.transport = mock.Mock()
                # mark the protocol as ready
                protocol.connectionReady(protocol)
                return protocol

            mocked_client_from_string.return_value.connect.side_effect = connect
            provider.startService()

            connection = connection_dependency.get_resource()
            self.assertIsInstance(connection, providers.AMQProtocol)

            self.assertEquals(connection.factory.servers, cm.get('amqp.connections.test_connection.servers'))
            self.assertEquals(connection.factory.reconnect_interval, cm.get('amqp.connections.test_connection.reconnect_interval'))
            self.assertEquals(connection.idle_checker.interval, cm.get('amqp.connections.test_connection.max_idle_time'))
            self.assertEquals(connection.parameters.heartbeat, cm.get('amqp.connections.test_connection.parameters.heartbeat'))

        # if another dependency asks for the same connection, it should receive the same instance
        another_connection_dependency = dependencies.ResourceDependency(provider='amqp.connection.test_connection')
        self.runtime_environment.resource_manager.resolve(another_connection_dependency)

        self.assertIdentical(another_connection_dependency.get_resource(), connection_dependency.get_resource())


class TestMockConnection(unittest.TestCase):

    def setUp(self):
        self.service = service.MultiService()

    @defer.inlineCallbacks
    def tearDown(self):
        yield self.service.stopService()

    def test_servers_config(self):
        # servers are usually specified as a list ...
        rpc_client = providers.AMQPConnection('test_name', servers=['tcp:host=localhost:port=5672'])
        self.assertEquals(rpc_client.get_next_server(), 'tcp:host=localhost:port=5672')

        # ... but it can be a string, which should be wrapped in a list:
        rpc_client = providers.AMQPConnection('test_name', servers='tcp:host=localhost:port=5672')
        self.assertEquals(rpc_client.get_next_server(), 'tcp:host=localhost:port=5672')

    @defer.inlineCallbacks
    def test_get_named_channels(self):
        connection = providers.AMQPConnection('test_name', servers=['tcp:host=server_1:port=5672'])
        connection.setServiceParent(self.service)

        with patch.object(providers.endpoints, 'clientFromString') as mocked_client_from_string:
            def connect(factory):
                protocol = factory.buildProtocol(None)
                protocol.transport = mock.Mock()
                # mark the protocol as ready
                protocol.connectionReady(protocol)
                return protocol

            mocked_client_from_string.return_value.connect.side_effect = connect
            connection.startService()

            # the connection should now be finished connecting:
            self.assertTrue(connection.ready)

            protocol = connection.protocol

            channel_requests = list()
            with patch.object(protocol, 'channel') as mocked_channel:
                def channel():
                    d = defer.Deferred()
                    channel_requests.append(d)
                    return d

                mocked_channel.side_effect = channel

                foo_1_d = protocol.get_named_channel('foo')
                foo_2_d = protocol.get_named_channel('foo')
                foo_3_d = protocol.get_named_channel('foo')

                bar_d = protocol.get_named_channel('bar')

                self.assertEquals(len(channel_requests), 2)

                mocked_foo = mock.Mock()
                mocked_foo_close = event.Event()
                mocked_foo.add_on_close_callback.side_effect = mocked_foo_close.handle
                mocked_bar = mock.Mock()

                channel_requests.pop(0).callback(mocked_foo)
                channel_requests.pop(0).callback(mocked_bar)

                # all the pending requests for the channel should receive the same instance
                foo = yield defer.DeferredList([foo_1_d, foo_2_d, foo_3_d])
                self.assertEquals(set(foo), set([(True, mocked_foo)]))

                bar = yield bar_d
                self.assertEquals(bar, mocked_bar)

                self.assertEquals(len(channel_requests), 0)
                another_foo = yield protocol.get_named_channel('foo')
                self.assertEquals(len(channel_requests), 0)

                # if the channel is closed, the next request to get the channel will recreate it
                mocked_foo_close()
                self.assertEquals(len(channel_requests), 0)
                new_foo = protocol.get_named_channel('foo')
                self.assertEquals(len(channel_requests), 1)

                # if the channel request errbacks, all pending requests for the named channel should errback
                new_foo_2 = protocol.get_named_channel('foo')
                test_exception = Exception()
                channel_requests.pop(0).callback(failure.Failure(test_exception))

                for pending_foo in (new_foo, new_foo_2):
                    try:
                        yield pending_foo
                        self.fail('Expected {0!r} to be raised.'.format(new_foo))
                    except Exception as e:
                        self.assertEquals(e, test_exception)


    def test_connect_disconnect_when_service_starts_stops(self):
        connection = providers.AMQPConnection('test_name', servers=['tcp:host=localhost:port=5672'])

        with patch.object(connection, '_connect') as mocked_connect:
            def verify(endpoint):
                self.assertEquals(endpoint._host, 'localhost')
                self.assertEquals(endpoint._port, 5672)

            mocked_connect.side_effect = verify
            connection.startService()

            mocked_connect.assert_called_once()

        with patch.object(connection, '_disconnect') as mocked_disconnect:
            connection.stopService()

            mocked_disconnect.assert_called_once_with('stopping service')

    @defer.inlineCallbacks
    def test_reconnect_if_connect_fails(self):
        connection = providers.AMQPConnection('test_name', servers=['tcp:host=server_1:port=5672', 'tcp:host=server_2:port=5673'], reconnect_interval=0)
        connection.setServiceParent(self.service)

        # create a list of return values for connect. the last element is a deferred, which enables
        # us to assert that the reconnecting hasn't given up.
        endpoints = list()
        connect_return_values = [error.ConnectError('test_error'), error.DNSLookupError('test_error'), error.ConnectionDone('test_error'), Exception('test_exception'), defer.Deferred()]
        expected_errors = list(connect_return_values[:-1])

        disconnects = list()
        connection.on_disconnected += disconnects.append

        with patch.object(connection, '_connect') as mocked_connect:
            def connect(endpoint):
                endpoints.append(dict(host=endpoint._host, port=endpoint._port))

{'fraction_non_alphanumeric': 0.08560445234953239, 'fraction_numerical': 0.008491594468988468, 'mean_word_length': 2.710089399744572, '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': '1559094', 'n_tokens_mistral': 4629, 'n_tokens_neox': 4304, 'n_words': 883}