Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use prepare_version method in localstack

Best Python code snippet using localstack_python
test_client.py
Source:test_client.py
1# Copyright 2013 Red Hat, Inc.2#3#    Licensed under the Apache License, Version 2.0 (the "License"); you may4#    not use this file except in compliance with the License. You may obtain5#    a copy of the License at6#7#         http://www.apache.org/licenses/LICENSE-2.08#9#    Unless required by applicable law or agreed to in writing, software10#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT11#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the12#    License for the specific language governing permissions and limitations13#    under the License.14from oslo_config import cfg15import testscenarios16import oslo_messaging17from oslo_messaging import exceptions18from oslo_messaging import serializer as msg_serializer19from oslo_messaging.tests import utils as test_utils20load_tests = testscenarios.load_tests_apply_scenarios21class _FakeTransport(object):22    def __init__(self, conf):23        self.conf = conf24    def _send(self, *args, **kwargs):25        pass26class TestCastCall(test_utils.BaseTestCase):27    scenarios = [28        ('cast_no_ctxt_no_args', dict(call=False, ctxt={}, args={})),29        ('call_no_ctxt_no_args', dict(call=True, ctxt={}, args={})),30        ('cast_ctxt_and_args',31         dict(call=False,32              ctxt=dict(user='testuser', project='testtenant'),33              args=dict(bar='blaa', foobar=11.01))),34        ('call_ctxt_and_args',35         dict(call=True,36              ctxt=dict(user='testuser', project='testtenant'),37              args=dict(bar='blaa', foobar=11.01))),38    ]39    def test_cast_call(self):40        self.config(rpc_response_timeout=None)41        transport = _FakeTransport(self.conf)42        client = oslo_messaging.RPCClient(transport, oslo_messaging.Target())43        self.mox.StubOutWithMock(transport, '_send')44        msg = dict(method='foo', args=self.args)45        kwargs = {'retry': None}46        if self.call:47            kwargs['wait_for_reply'] = True48            kwargs['timeout'] = None49        transport._send(oslo_messaging.Target(), self.ctxt, msg, **kwargs)50        self.mox.ReplayAll()51        method = client.call if self.call else client.cast52        method(self.ctxt, 'foo', **self.args)53class TestCastToTarget(test_utils.BaseTestCase):54    _base = [55        ('all_none', dict(ctor={}, prepare={}, expect={})),56        ('ctor_exchange',57         dict(ctor=dict(exchange='testexchange'),58              prepare={},59              expect=dict(exchange='testexchange'))),60        ('prepare_exchange',61         dict(ctor={},62              prepare=dict(exchange='testexchange'),63              expect=dict(exchange='testexchange'))),64        ('prepare_exchange_none',65         dict(ctor=dict(exchange='testexchange'),66              prepare=dict(exchange=None),67              expect={})),68        ('both_exchange',69         dict(ctor=dict(exchange='ctorexchange'),70              prepare=dict(exchange='testexchange'),71              expect=dict(exchange='testexchange'))),72        ('ctor_topic',73         dict(ctor=dict(topic='testtopic'),74              prepare={},75              expect=dict(topic='testtopic'))),76        ('prepare_topic',77         dict(ctor={},78              prepare=dict(topic='testtopic'),79              expect=dict(topic='testtopic'))),80        ('prepare_topic_none',81         dict(ctor=dict(topic='testtopic'),82              prepare=dict(topic=None),83              expect={})),84        ('both_topic',85         dict(ctor=dict(topic='ctortopic'),86              prepare=dict(topic='testtopic'),87              expect=dict(topic='testtopic'))),88        ('ctor_namespace',89         dict(ctor=dict(namespace='testnamespace'),90              prepare={},91              expect=dict(namespace='testnamespace'))),92        ('prepare_namespace',93         dict(ctor={},94              prepare=dict(namespace='testnamespace'),95              expect=dict(namespace='testnamespace'))),96        ('prepare_namespace_none',97         dict(ctor=dict(namespace='testnamespace'),98              prepare=dict(namespace=None),99              expect={})),100        ('both_namespace',101         dict(ctor=dict(namespace='ctornamespace'),102              prepare=dict(namespace='testnamespace'),103              expect=dict(namespace='testnamespace'))),104        ('ctor_version',105         dict(ctor=dict(version='1.1'),106              prepare={},107              expect=dict(version='1.1'))),108        ('prepare_version',109         dict(ctor={},110              prepare=dict(version='1.1'),111              expect=dict(version='1.1'))),112        ('prepare_version_none',113         dict(ctor=dict(version='1.1'),114              prepare=dict(version=None),115              expect={})),116        ('both_version',117         dict(ctor=dict(version='ctorversion'),118              prepare=dict(version='1.1'),119              expect=dict(version='1.1'))),120        ('ctor_server',121         dict(ctor=dict(server='testserver'),122              prepare={},123              expect=dict(server='testserver'))),124        ('prepare_server',125         dict(ctor={},126              prepare=dict(server='testserver'),127              expect=dict(server='testserver'))),128        ('prepare_server_none',129         dict(ctor=dict(server='testserver'),130              prepare=dict(server=None),131              expect={})),132        ('both_server',133         dict(ctor=dict(server='ctorserver'),134              prepare=dict(server='testserver'),135              expect=dict(server='testserver'))),136        ('ctor_fanout',137         dict(ctor=dict(fanout=True),138              prepare={},139              expect=dict(fanout=True))),140        ('prepare_fanout',141         dict(ctor={},142              prepare=dict(fanout=True),143              expect=dict(fanout=True))),144        ('prepare_fanout_none',145         dict(ctor=dict(fanout=True),146              prepare=dict(fanout=None),147              expect={})),148        ('both_fanout',149         dict(ctor=dict(fanout=True),150              prepare=dict(fanout=False),151              expect=dict(fanout=False))),152    ]153    _prepare = [154        ('single_prepare', dict(double_prepare=False)),155        ('double_prepare', dict(double_prepare=True)),156    ]157    @classmethod158    def generate_scenarios(cls):159        cls.scenarios = testscenarios.multiply_scenarios(cls._base,160                                                         cls._prepare)161    def setUp(self):162        super(TestCastToTarget, self).setUp(conf=cfg.ConfigOpts())163    def test_cast_to_target(self):164        target = oslo_messaging.Target(**self.ctor)165        expect_target = oslo_messaging.Target(**self.expect)166        transport = _FakeTransport(self.conf)167        client = oslo_messaging.RPCClient(transport, target)168        self.mox.StubOutWithMock(transport, '_send')169        msg = dict(method='foo', args={})170        if 'namespace' in self.expect:171            msg['namespace'] = self.expect['namespace']172        if 'version' in self.expect:173            msg['version'] = self.expect['version']174        transport._send(expect_target, {}, msg, retry=None)175        self.mox.ReplayAll()176        if self.prepare:177            client = client.prepare(**self.prepare)178            if self.double_prepare:179                client = client.prepare(**self.prepare)180        client.cast({}, 'foo')181TestCastToTarget.generate_scenarios()182_notset = object()183class TestCallTimeout(test_utils.BaseTestCase):184    scenarios = [185        ('all_none',186         dict(confval=None, ctor=None, prepare=_notset, expect=None)),187        ('confval',188         dict(confval=21, ctor=None, prepare=_notset, expect=21)),189        ('ctor',190         dict(confval=None, ctor=21.1, prepare=_notset, expect=21.1)),191        ('ctor_zero',192         dict(confval=None, ctor=0, prepare=_notset, expect=0)),193        ('prepare',194         dict(confval=None, ctor=None, prepare=21.1, expect=21.1)),195        ('prepare_override',196         dict(confval=None, ctor=10.1, prepare=21.1, expect=21.1)),197        ('prepare_zero',198         dict(confval=None, ctor=None, prepare=0, expect=0)),199    ]200    def test_call_timeout(self):201        self.config(rpc_response_timeout=self.confval)202        transport = _FakeTransport(self.conf)203        client = oslo_messaging.RPCClient(transport, oslo_messaging.Target(),204                                          timeout=self.ctor)205        self.mox.StubOutWithMock(transport, '_send')206        msg = dict(method='foo', args={})207        kwargs = dict(wait_for_reply=True, timeout=self.expect, retry=None)208        transport._send(oslo_messaging.Target(), {}, msg, **kwargs)209        self.mox.ReplayAll()210        if self.prepare is not _notset:211            client = client.prepare(timeout=self.prepare)212        client.call({}, 'foo')213class TestCallRetry(test_utils.BaseTestCase):214    scenarios = [215        ('all_none', dict(ctor=None, prepare=_notset, expect=None)),216        ('ctor', dict(ctor=21, prepare=_notset, expect=21)),217        ('ctor_zero', dict(ctor=0, prepare=_notset, expect=0)),218        ('prepare', dict(ctor=None, prepare=21, expect=21)),219        ('prepare_override', dict(ctor=10, prepare=21, expect=21)),220        ('prepare_zero', dict(ctor=None, prepare=0, expect=0)),221    ]222    def test_call_retry(self):223        transport = _FakeTransport(self.conf)224        client = oslo_messaging.RPCClient(transport, oslo_messaging.Target(),225                                          retry=self.ctor)226        self.mox.StubOutWithMock(transport, '_send')227        msg = dict(method='foo', args={})228        kwargs = dict(wait_for_reply=True, timeout=60,229                      retry=self.expect)230        transport._send(oslo_messaging.Target(), {}, msg, **kwargs)231        self.mox.ReplayAll()232        if self.prepare is not _notset:233            client = client.prepare(retry=self.prepare)234        client.call({}, 'foo')235class TestCallFanout(test_utils.BaseTestCase):236    scenarios = [237        ('target', dict(prepare=_notset, target={'fanout': True})),238        ('prepare', dict(prepare={'fanout': True}, target={})),239        ('both', dict(prepare={'fanout': True}, target={'fanout': True})),240    ]241    def test_call_fanout(self):242        transport = _FakeTransport(self.conf)243        client = oslo_messaging.RPCClient(transport,244                                          oslo_messaging.Target(**self.target))245        if self.prepare is not _notset:246            client = client.prepare(**self.prepare)247        self.assertRaises(exceptions.InvalidTarget,248                          client.call, {}, 'foo')249class TestSerializer(test_utils.BaseTestCase):250    scenarios = [251        ('cast',252         dict(call=False,253              ctxt=dict(user='bob'),254              args=dict(a='a', b='b', c='c'),255              retval=None)),256        ('call',257         dict(call=True,258              ctxt=dict(user='bob'),259              args=dict(a='a', b='b', c='c'),260              retval='d')),261    ]262    def test_call_serializer(self):263        self.config(rpc_response_timeout=None)264        transport = _FakeTransport(self.conf)265        serializer = msg_serializer.NoOpSerializer()266        client = oslo_messaging.RPCClient(transport, oslo_messaging.Target(),267                                          serializer=serializer)268        self.mox.StubOutWithMock(transport, '_send')269        msg = dict(method='foo',270                   args=dict([(k, 's' + v) for k, v in self.args.items()]))271        kwargs = dict(wait_for_reply=True, timeout=None) if self.call else {}272        kwargs['retry'] = None273        transport._send(oslo_messaging.Target(),274                        dict(user='alice'),275                        msg,276                        **kwargs).AndReturn(self.retval)277        self.mox.StubOutWithMock(serializer, 'serialize_entity')278        self.mox.StubOutWithMock(serializer, 'deserialize_entity')279        self.mox.StubOutWithMock(serializer, 'serialize_context')280        for arg in self.args:281            serializer.serialize_entity(self.ctxt, arg).AndReturn('s' + arg)282        if self.call:283            serializer.deserialize_entity(self.ctxt, self.retval).\284                AndReturn('d' + self.retval)285        serializer.serialize_context(self.ctxt).AndReturn(dict(user='alice'))286        self.mox.ReplayAll()287        method = client.call if self.call else client.cast288        retval = method(self.ctxt, 'foo', **self.args)289        if self.retval is not None:290            self.assertEqual('d' + self.retval, retval)291class TestVersionCap(test_utils.BaseTestCase):292    _call_vs_cast = [293        ('call', dict(call=True)),294        ('cast', dict(call=False)),295    ]296    _cap_scenarios = [297        ('all_none',298         dict(cap=None, prepare_cap=_notset,299              version=None, prepare_version=_notset,300              success=True)),301        ('ctor_cap_ok',302         dict(cap='1.1', prepare_cap=_notset,303              version='1.0', prepare_version=_notset,304              success=True)),305        ('ctor_cap_override_ok',306         dict(cap='2.0', prepare_cap='1.1',307              version='1.0', prepare_version='1.0',308              success=True)),309        ('ctor_cap_override_none_ok',310         dict(cap='1.1', prepare_cap=None,311              version='1.0', prepare_version=_notset,312              success=True)),313        ('ctor_cap_minor_fail',314         dict(cap='1.0', prepare_cap=_notset,315              version='1.1', prepare_version=_notset,316              success=False)),317        ('ctor_cap_major_fail',318         dict(cap='2.0', prepare_cap=_notset,319              version=None, prepare_version='1.0',320              success=False)),321    ]322    @classmethod323    def generate_scenarios(cls):324        cls.scenarios = (325            testscenarios.multiply_scenarios(cls._call_vs_cast,326                                             cls._cap_scenarios))327    def test_version_cap(self):328        self.config(rpc_response_timeout=None)329        transport = _FakeTransport(self.conf)330        target = oslo_messaging.Target(version=self.version)331        client = oslo_messaging.RPCClient(transport, target,332                                          version_cap=self.cap)333        if self.success:334            self.mox.StubOutWithMock(transport, '_send')335            if self.prepare_version is not _notset:336                target = target(version=self.prepare_version)337            msg = dict(method='foo', args={})338            if target.version is not None:339                msg['version'] = target.version340            kwargs = {'retry': None}341            if self.call:342                kwargs['wait_for_reply'] = True343                kwargs['timeout'] = None344            transport._send(target, {}, msg, **kwargs)345            self.mox.ReplayAll()346        prep_kwargs = {}347        if self.prepare_cap is not _notset:348            prep_kwargs['version_cap'] = self.prepare_cap349        if self.prepare_version is not _notset:350            prep_kwargs['version'] = self.prepare_version351        if prep_kwargs:352            client = client.prepare(**prep_kwargs)353        method = client.call if self.call else client.cast354        try:355            method({}, 'foo')356        except Exception as ex:357            self.assertIsInstance(ex, oslo_messaging.RPCVersionCapError, ex)358            self.assertFalse(self.success)359        else:360            self.assertTrue(self.success)361TestVersionCap.generate_scenarios()362class TestCanSendVersion(test_utils.BaseTestCase):363    scenarios = [364        ('all_none',365         dict(cap=None, prepare_cap=_notset,366              version=None, prepare_version=_notset,367              can_send_version=_notset,368              can_send=True)),369        ('ctor_cap_ok',370         dict(cap='1.1', prepare_cap=_notset,371              version='1.0', prepare_version=_notset,372              can_send_version=_notset,373              can_send=True)),374        ('ctor_cap_override_ok',375         dict(cap='2.0', prepare_cap='1.1',376              version='1.0', prepare_version='1.0',377              can_send_version=_notset,378              can_send=True)),379        ('ctor_cap_override_none_ok',380         dict(cap='1.1', prepare_cap=None,381              version='1.0', prepare_version=_notset,382              can_send_version=_notset,383              can_send=True)),384        ('ctor_cap_can_send_ok',385         dict(cap='1.1', prepare_cap=None,386              version='1.0', prepare_version=_notset,387              can_send_version='1.1',388              can_send=True)),389        ('ctor_cap_can_send_none_ok',390         dict(cap='1.1', prepare_cap=None,391              version='1.0', prepare_version=_notset,392              can_send_version=None,393              can_send=True)),394        ('ctor_cap_minor_fail',395         dict(cap='1.0', prepare_cap=_notset,396              version='1.1', prepare_version=_notset,397              can_send_version=_notset,398              can_send=False)),399        ('ctor_cap_major_fail',400         dict(cap='2.0', prepare_cap=_notset,401              version=None, prepare_version='1.0',402              can_send_version=_notset,403              can_send=False)),404    ]405    def test_version_cap(self):406        self.config(rpc_response_timeout=None)407        transport = _FakeTransport(self.conf)408        target = oslo_messaging.Target(version=self.version)409        client = oslo_messaging.RPCClient(transport, target,410                                          version_cap=self.cap)411        prep_kwargs = {}412        if self.prepare_cap is not _notset:413            prep_kwargs['version_cap'] = self.prepare_cap414        if self.prepare_version is not _notset:415            prep_kwargs['version'] = self.prepare_version416        if prep_kwargs:417            client = client.prepare(**prep_kwargs)418        if self.can_send_version is not _notset:419            can_send = client.can_send_version(version=self.can_send_version)420        else:421            can_send = client.can_send_version()422        self.assertEqual(self.can_send, can_send)423    def test_invalid_version_type(self):424        target = oslo_messaging.Target(topic='sometopic')425        transport = _FakeTransport(self.conf)426        client = oslo_messaging.RPCClient(transport, target)427        self.assertRaises(exceptions.MessagingException,428                          client.prepare, version='5')429        self.assertRaises(exceptions.MessagingException,...
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.