KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_dispatcher method in localstack

Best Python code snippet using localstack_python

test_dispatchers.py

Source:test_dispatchers.py

...25        self.listener_port = self.listener.getHost().port26        # NOTE: setting the clock before the setup_dispatcher stuff is called27        PortiaDispatcher.clock = Clock()28        self.disp_helper = self.add_helper(DispatcherHelper(PortiaDispatcher))29    def get_dispatcher(self, **config_extras):30        config = {31            "receive_inbound_connectors": ["transport1", "transport2"],32            "receive_outbound_connectors": ["app1"],33            "portia_endpoint": "tcp:%s:%s" % (self.listener_host,34                                              self.listener_port),35            "mapping": {36                "transport1": {37                    "default": "mno1"38                },39                "transport2": {40                    "default": "mno2"41                },42            }43        }44        config.update(config_extras)45        return self.disp_helper.get_dispatcher(config)46    def ch(self, connector_name):47        return self.disp_helper.get_connector_helper(connector_name)48    def assert_rkeys_used(self, *rkeys):49        broker = self.disp_helper.worker_helper.broker50        self.assertEqual(set(rkeys), set(broker.dispatched['vumi'].keys()))51    def assert_dispatched_endpoint(self, msg, endpoint, dispatched_msgs):52        msg.set_routing_endpoint(endpoint)53        self.assertEqual([msg], dispatched_msgs)54    def test_unique_mno_config(self):55        failure = self.assertRaises(56            DispatcherError,57            self.get_dispatcher,58            mapping={59                'transport1': {60                    "default": 'mno1',61                },62                'transport2': {63                    "default": 'mno1',64                    "ep1": 'mno2',65                },66            })67        self.assertEqual(68            str(failure),69            'PortiaDispatcher mappings are not unique.')70    def test_single_receive_outbound_connector(self):71        failure = self.assertRaises(72            DispatcherError, self.get_dispatcher,73            receive_outbound_connectors=['app1', 'app2'])74        self.assertEqual(75            str(failure),76            ('PortiaRouter is only able to work with 1 receive outbound '77             'connector, there are 2 configured.'))78    def test_single_receive_inbound_connector_mapping(self):79        failure = self.assertRaises(80            DispatcherError, self.get_dispatcher,81            receive_inbound_connectors=[82                'transport1', 'transport2', 'transport3'])83        self.assertEqual(84            str(failure),85            ('Not all receive_inbound_connectors mapped to MNOs.'))86    @inlineCallbacks87    def test_inbound_message_routing(self):88        from_addr = '+27123456789'89        resolve_response = yield self.portia.resolve(90            portia_normalize_msisdn(from_addr))91        self.assertEqual(resolve_response['network'], None)92        yield self.get_dispatcher()93        msg = yield self.ch("transport1").make_dispatch_inbound(94            "inbound", from_addr=from_addr)95        self.assert_rkeys_used('transport1.inbound', 'app1.inbound')96        self.assert_dispatched_endpoint(97            msg, 'default', self.ch('app1').get_dispatched_inbound())98        resolve_response = yield self.portia.resolve(99            portia_normalize_msisdn(from_addr))100        self.assertEqual(resolve_response['network'], 'mno1')101    @inlineCallbacks102    def test_inbound_event_routing(self):103        yield self.get_dispatcher()104        msg = yield self.ch('transport1').make_dispatch_ack()105        self.assert_rkeys_used('transport1.event', 'app1.event')106        self.assert_dispatched_endpoint(107            msg, 'default', self.ch('app1').get_dispatched_events())108    @inlineCallbacks109    def test_outbound_message_routing(self):110        to_addr = '+27123456789'111        yield self.portia.annotate(112            portia_normalize_msisdn(to_addr),113            key='observed-network', value='mno1',114            timestamp=self.portia.now())115        yield self.get_dispatcher()116        msg = yield self.ch('app1').make_dispatch_outbound(117            "outbound", to_addr=to_addr)118        self.assert_rkeys_used('app1.outbound', 'transport1.outbound')119        self.assert_dispatched_endpoint(120            msg, 'default', self.ch('transport1').get_dispatched_outbound())121    @inlineCallbacks122    def test_outbound_message_unresolvable(self):123        to_addr = '+27123456789'124        yield self.get_dispatcher()125        yield self.ch('app1').make_dispatch_outbound(126            "outbound", to_addr=to_addr)127        [failure] = self.flushLoggedErrors()128        self.assertTrue(129            "Portia was unable to resolve:" in failure.getErrorMessage())130    @inlineCallbacks131    def test_outbound_message_unroutable(self):132        to_addr = '+27123456789'133        yield self.portia.annotate(134            portia_normalize_msisdn(to_addr),135            key='observed-network',136            # NOTE: this is an MNO that is not configured and137            #       vxportia can not route138            value='XXX',139            timestamp=self.portia.now())140        yield self.get_dispatcher()141        yield self.ch('app1').make_dispatch_outbound(142            "outbound", to_addr=to_addr)143        [failure] = self.flushLoggedErrors()144        self.assertTrue(145            "Unable to route outbound message to:"...

__init__.py

Source:__init__.py

...13#  limitations under the License.14# Import all backends so that they register themselves to the Dispatcher15from .backends import *16from .backend import get_dispatcher, set_data_dir, get_data_dir17save = get_dispatcher().save18load = get_dispatcher().load19get_backend = get_dispatcher().get_backend20get_backends = get_dispatcher().get_backends21get_backend_by_name = get_dispatcher().get_backend_by_name22from .decorator import Marshaller23# External code shouldn't care about the Dispatcher instance24del get_dispatcher25from kale.common import logutils26logutils.get_or_create_logger(module=__name__, name="marshalling")...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.