How to use expected_kwargs method in Kiwi

Best Python code snippet using Kiwi_python

test_bgpspeaker.py

Source:test_bgpspeaker.py

1# Copyright (C) 2016 Nippon Telegraph and Telephone Corporation.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain 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,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or12# implied.13# See the License for the specific language governing permissions and14# limitations under the License.15import unittest16import logging17try:18    import mock  # Python 219except ImportError:20    from unittest import mock  # Python 321from nose.tools import raises22from ryu.services.protocols.bgp import bgpspeaker23from ryu.services.protocols.bgp.bgpspeaker import EVPN_MAX_ET24from ryu.services.protocols.bgp.bgpspeaker import ESI_TYPE_LACP25from ryu.services.protocols.bgp.api.prefix import ESI_TYPE_L2_BRIDGE26from ryu.services.protocols.bgp.bgpspeaker import ESI_TYPE_MAC_BASED27from ryu.services.protocols.bgp.api.prefix import REDUNDANCY_MODE_ALL_ACTIVE28from ryu.services.protocols.bgp.api.prefix import REDUNDANCY_MODE_SINGLE_ACTIVE29LOG = logging.getLogger(__name__)30class Test_BGPSpeaker(unittest.TestCase):31    """32    Test case for bgp.bgpspeaker.BGPSpeaker33    """34    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',35                mock.MagicMock(return_value=None))36    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')37    def test_evpn_prefix_add_eth_auto_discovery(self, mock_call):38        # Prepare test data39        route_type = bgpspeaker.EVPN_ETH_AUTO_DISCOVERY40        route_dist = '65000:100'41        esi = {42            'type': ESI_TYPE_LACP,43            'mac_addr': 'aa:bb:cc:dd:ee:ff',44            'port_key': 100,45        }46        ethernet_tag_id = EVPN_MAX_ET47        redundancy_mode = REDUNDANCY_MODE_ALL_ACTIVE48        next_hop = '0.0.0.0'49        expected_kwargs = {50            'route_type': route_type,51            'route_dist': route_dist,52            'esi': esi,53            'ethernet_tag_id': ethernet_tag_id,54            'redundancy_mode': redundancy_mode,55            'next_hop': next_hop,56        }57        # Test58        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')59        speaker.evpn_prefix_add(60            route_type=route_type,61            route_dist=route_dist,62            esi=esi,63            ethernet_tag_id=ethernet_tag_id,64            redundancy_mode=redundancy_mode,65        )66        # Check67        mock_call.assert_called_with(68            'evpn_prefix.add_local', **expected_kwargs)69    @mock.patch(70        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',71        mock.MagicMock(return_value=None))72    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')73    def test_evpn_prefix_add_eth_auto_discovery_vni(self, mock_call):74        # Prepare test data75        route_type = bgpspeaker.EVPN_ETH_AUTO_DISCOVERY76        route_dist = '65000:100'77        esi = {78            'type': ESI_TYPE_L2_BRIDGE,79            'mac_addr': 'aa:bb:cc:dd:ee:ff',80            'priority': 100,81        }82        ethernet_tag_id = EVPN_MAX_ET83        redundancy_mode = REDUNDANCY_MODE_SINGLE_ACTIVE84        vni = 50085        next_hop = '0.0.0.0'86        expected_kwargs = {87            'route_type': route_type,88            'route_dist': route_dist,89            'esi': esi,90            'ethernet_tag_id': ethernet_tag_id,91            'redundancy_mode': redundancy_mode,92            'vni': vni,93            'next_hop': next_hop,94        }95        # Test96        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')97        speaker.evpn_prefix_add(98            route_type=route_type,99            route_dist=route_dist,100            esi=esi,101            ethernet_tag_id=ethernet_tag_id,102            redundancy_mode=redundancy_mode,103            vni=vni104        )105        # Check106        mock_call.assert_called_with(107            'evpn_prefix.add_local', **expected_kwargs)108    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',109                mock.MagicMock(return_value=None))110    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')111    def test_evpn_prefix_add_mac_ip_adv(self, mock_call):112        # Prepare test data113        route_type = bgpspeaker.EVPN_MAC_IP_ADV_ROUTE114        route_dist = '65000:100'115        esi = 0  # denotes single-homed116        ethernet_tag_id = 200117        mac_addr = 'aa:bb:cc:dd:ee:ff'118        ip_addr = '192.168.0.1'119        next_hop = '10.0.0.1'120        expected_kwargs = {121            'route_type': route_type,122            'route_dist': route_dist,123            'esi': esi,124            'ethernet_tag_id': ethernet_tag_id,125            'mac_addr': mac_addr,126            'ip_addr': ip_addr,127            'next_hop': next_hop,128        }129        # Test130        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')131        speaker.evpn_prefix_add(132            route_type=route_type,133            route_dist=route_dist,134            esi=esi,135            ethernet_tag_id=ethernet_tag_id,136            mac_addr=mac_addr,137            ip_addr=ip_addr,138            next_hop=next_hop,139        )140        # Check141        mock_call.assert_called_with(142            'evpn_prefix.add_local', **expected_kwargs)143    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',144                mock.MagicMock(return_value=None))145    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')146    def test_evpn_prefix_add_mac_ip_adv_vni(self, mock_call):147        # Prepare test data148        route_type = bgpspeaker.EVPN_MAC_IP_ADV_ROUTE149        route_dist = '65000:100'150        esi = 0  # denotes single-homed151        ethernet_tag_id = 200152        mac_addr = 'aa:bb:cc:dd:ee:ff'153        ip_addr = '192.168.0.1'154        vni = 500155        next_hop = '10.0.0.1'156        tunnel_type = bgpspeaker.TUNNEL_TYPE_VXLAN157        expected_kwargs = {158            'route_type': route_type,159            'route_dist': route_dist,160            'esi': esi,161            'ethernet_tag_id': ethernet_tag_id,162            'mac_addr': mac_addr,163            'ip_addr': ip_addr,164            'vni': vni,165            'next_hop': next_hop,166            'tunnel_type': tunnel_type,167        }168        # Test169        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')170        speaker.evpn_prefix_add(171            route_type=route_type,172            route_dist=route_dist,173            esi=esi,174            ethernet_tag_id=ethernet_tag_id,175            mac_addr=mac_addr,176            ip_addr=ip_addr,177            vni=vni,178            next_hop=next_hop,179            tunnel_type=tunnel_type,180        )181        # Check182        mock_call.assert_called_with(183            'evpn_prefix.add_local', **expected_kwargs)184    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',185                mock.MagicMock(return_value=None))186    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')187    def test_evpn_prefix_add_multicast_etag(self, mock_call):188        # Prepare test data189        route_type = bgpspeaker.EVPN_MULTICAST_ETAG_ROUTE190        route_dist = '65000:100'191        esi = 0  # denotes single-homed192        ethernet_tag_id = 200193        mac_addr = 'aa:bb:cc:dd:ee:ff'194        ip_addr = '192.168.0.1'195        next_hop = '10.0.0.1'196        expected_kwargs = {197            'route_type': route_type,198            'route_dist': route_dist,199            # 'esi': esi,  # should be ignored200            'ethernet_tag_id': ethernet_tag_id,201            # 'mac_addr': mac_addr,  # should be ignored202            'ip_addr': ip_addr,203            'next_hop': next_hop,204        }205        # Test206        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')207        speaker.evpn_prefix_add(208            route_type=route_type,209            route_dist=route_dist,210            esi=esi,211            ethernet_tag_id=ethernet_tag_id,212            mac_addr=mac_addr,213            ip_addr=ip_addr,214            next_hop=next_hop,215        )216        # Check217        mock_call.assert_called_with(218            'evpn_prefix.add_local', **expected_kwargs)219    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',220                mock.MagicMock(return_value=None))221    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')222    def test_evpn_prefix_add_multicast_etag_no_next_hop(self, mock_call):223        # Prepare test data224        route_type = bgpspeaker.EVPN_MULTICAST_ETAG_ROUTE225        route_dist = '65000:100'226        esi = 0  # denotes single-homed227        ethernet_tag_id = 200228        mac_addr = 'aa:bb:cc:dd:ee:ff'229        ip_addr = '192.168.0.1'230        next_hop = '0.0.0.0'  # the default value231        expected_kwargs = {232            'route_type': route_type,233            'route_dist': route_dist,234            # 'esi': esi,  # should be ignored235            'ethernet_tag_id': ethernet_tag_id,236            # 'mac_addr': mac_addr,  # should be ignored237            'ip_addr': ip_addr,238            'next_hop': next_hop,239        }240        # Test241        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')242        speaker.evpn_prefix_add(243            route_type=route_type,244            route_dist=route_dist,245            esi=esi,246            ethernet_tag_id=ethernet_tag_id,247            mac_addr=mac_addr,248            ip_addr=ip_addr,249            # next_hop=next_hop,  # omitted250        )251        # Check252        mock_call.assert_called_with(253            'evpn_prefix.add_local', **expected_kwargs)254    @mock.patch(255        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',256        mock.MagicMock(return_value=None))257    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')258    def test_evpn_prefix_add_eth_segment(self, mock_call):259        # Prepare test data260        route_type = bgpspeaker.EVPN_ETH_SEGMENT261        route_dist = '65000:100'262        esi = {263            'type': ESI_TYPE_MAC_BASED,264            'mac_addr': 'aa:bb:cc:dd:ee:ff',265            'local_disc': 100,266        }267        ip_addr = '192.168.0.1'268        next_hop = '0.0.0.0'269        expected_kwargs = {270            'route_type': route_type,271            'route_dist': route_dist,272            'esi': esi,273            'ip_addr': ip_addr,274            'next_hop': next_hop,275        }276        # Test277        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')278        speaker.evpn_prefix_add(279            route_type=route_type,280            route_dist=route_dist,281            esi=esi,282            ip_addr=ip_addr,283        )284        # Check285        mock_call.assert_called_with(286            'evpn_prefix.add_local', **expected_kwargs)287    @mock.patch(288        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',289        mock.MagicMock(return_value=None))290    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')291    def test_evpn_prefix_add_ip_prefix_route(self, mock_call):292        # Prepare test data293        route_type = bgpspeaker.EVPN_IP_PREFIX_ROUTE294        route_dist = '65000:100'295        esi = 0  # denotes single-homed296        ethernet_tag_id = 200297        ip_prefix = '192.168.0.0/24'298        gw_ip_addr = '172.16.0.1'299        next_hop = '0.0.0.0'300        expected_kwargs = {301            'route_type': route_type,302            'route_dist': route_dist,303            'esi': esi,304            'ethernet_tag_id': ethernet_tag_id,305            'ip_prefix': ip_prefix,306            'gw_ip_addr': gw_ip_addr,307            'next_hop': next_hop,308        }309        # Test310        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')311        speaker.evpn_prefix_add(312            route_type=route_type,313            route_dist=route_dist,314            esi=esi,315            ethernet_tag_id=ethernet_tag_id,316            ip_prefix=ip_prefix,317            gw_ip_addr=gw_ip_addr,318        )319        # Check320        mock_call.assert_called_with(321            'evpn_prefix.add_local', **expected_kwargs)322    @mock.patch(323        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',324        mock.MagicMock(return_value=None))325    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')326    def test_evpn_prefix_add_ip_prefix_route_vni(self, mock_call):327        # Prepare test data328        route_type = bgpspeaker.EVPN_IP_PREFIX_ROUTE329        route_dist = '65000:100'330        esi = 0  # denotes single-homed331        ethernet_tag_id = 200332        ip_prefix = '192.168.0.0/24'333        gw_ip_addr = '172.16.0.1'334        vni = 500335        tunnel_type = bgpspeaker.TUNNEL_TYPE_VXLAN336        next_hop = '0.0.0.0'337        expected_kwargs = {338            'route_type': route_type,339            'route_dist': route_dist,340            'esi': esi,341            'ethernet_tag_id': ethernet_tag_id,342            'ip_prefix': ip_prefix,343            'gw_ip_addr': gw_ip_addr,344            'tunnel_type': tunnel_type,345            'vni': vni,346            'next_hop': next_hop,347        }348        # Test349        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')350        speaker.evpn_prefix_add(351            route_type=route_type,352            route_dist=route_dist,353            esi=esi,354            ethernet_tag_id=ethernet_tag_id,355            ip_prefix=ip_prefix,356            gw_ip_addr=gw_ip_addr,357            tunnel_type=tunnel_type,358            vni=vni,359        )360        # Check361        mock_call.assert_called_with(362            'evpn_prefix.add_local', **expected_kwargs)363    @raises(ValueError)364    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',365                mock.MagicMock(return_value=None))366    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')367    def test_evpn_prefix_add_invalid_route_type(self, mock_call):368        # Prepare test data369        route_type = 'foobar'  # Invalid EVPN route type370        route_dist = '65000:100'371        esi = 0  # denotes single-homed372        ethernet_tag_id = 200373        mac_addr = 'aa:bb:cc:dd:ee:ff'374        ip_addr = '192.168.0.1'375        next_hop = '10.0.0.1'376        # Test377        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')378        speaker.evpn_prefix_add(379            route_type=route_type,380            route_dist=route_dist,381            esi=esi,382            ethernet_tag_id=ethernet_tag_id,383            mac_addr=mac_addr,384            ip_addr=ip_addr,385            next_hop=next_hop,386        )387        # Check388        mock_call.assert_called_with(389            'evpn_prefix.add_local', 'Invalid arguments detected')390    @mock.patch(391        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',392        mock.MagicMock(return_value=None))393    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')394    def test_evpn_prefix_del_auto_discovery(self, mock_call):395        # Prepare test data396        route_type = bgpspeaker.EVPN_ETH_AUTO_DISCOVERY397        route_dist = '65000:100'398        esi = {399            'type': ESI_TYPE_LACP,400            'mac_addr': 'aa:bb:cc:dd:ee:ff',401            'port_key': 100,402        }403        ethernet_tag_id = EVPN_MAX_ET404        expected_kwargs = {405            'route_type': route_type,406            'route_dist': route_dist,407            'esi': esi,408            'ethernet_tag_id': ethernet_tag_id,409        }410        # Test411        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')412        speaker.evpn_prefix_del(413            route_type=route_type,414            route_dist=route_dist,415            esi=esi,416            ethernet_tag_id=ethernet_tag_id,417        )418        # Check419        mock_call.assert_called_with(420            'evpn_prefix.delete_local', **expected_kwargs)421    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',422                mock.MagicMock(return_value=None))423    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')424    def test_evpn_prefix_del_mac_ip_adv(self, mock_call):425        # Prepare test data426        route_type = bgpspeaker.EVPN_MAC_IP_ADV_ROUTE427        route_dist = '65000:100'428        ethernet_tag_id = 200429        mac_addr = 'aa:bb:cc:dd:ee:ff'430        ip_addr = '192.168.0.1'431        expected_kwargs = {432            'route_type': route_type,433            'route_dist': route_dist,434            'ethernet_tag_id': ethernet_tag_id,435            'mac_addr': mac_addr,436            'ip_addr': ip_addr,437        }438        # Test439        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')440        speaker.evpn_prefix_del(441            route_type=route_type,442            route_dist=route_dist,443            ethernet_tag_id=ethernet_tag_id,444            mac_addr=mac_addr,445            ip_addr=ip_addr,446        )447        # Check448        mock_call.assert_called_with(449            'evpn_prefix.delete_local', **expected_kwargs)450    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',451                mock.MagicMock(return_value=None))452    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')453    def test_evpn_prefix_del_multicast_etag(self, mock_call):454        # Prepare test data455        route_type = bgpspeaker.EVPN_MULTICAST_ETAG_ROUTE456        route_dist = '65000:100'457        esi = 0  # denotes single-homed458        ethernet_tag_id = 200459        mac_addr = 'aa:bb:cc:dd:ee:ff'460        ip_addr = '192.168.0.1'461        expected_kwargs = {462            'route_type': route_type,463            'route_dist': route_dist,464            # 'esi': esi,  # should be ignored465            'ethernet_tag_id': ethernet_tag_id,466            # 'mac_addr': mac_addr,  # should be ignored467            'ip_addr': ip_addr,468        }469        # Test470        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')471        speaker.evpn_prefix_del(472            route_type=route_type,473            route_dist=route_dist,474            esi=esi,475            ethernet_tag_id=ethernet_tag_id,476            mac_addr=mac_addr,477            ip_addr=ip_addr,478        )479        # Check480        mock_call.assert_called_with(481            'evpn_prefix.delete_local', **expected_kwargs)482    @raises(ValueError)483    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',484                mock.MagicMock(return_value=None))485    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')486    def test_evpn_prefix_del_invalid_route_type(self, mock_call):487        # Prepare test data488        route_type = 'foobar'  # Invalid EVPN route type489        route_dist = '65000:100'490        esi = 0  # denotes single-homed491        ethernet_tag_id = 200492        mac_addr = 'aa:bb:cc:dd:ee:ff'493        ip_addr = '192.168.0.1'494        # Test495        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')496        speaker.evpn_prefix_del(497            route_type=route_type,498            route_dist=route_dist,499            esi=esi,500            ethernet_tag_id=ethernet_tag_id,501            mac_addr=mac_addr,502            ip_addr=ip_addr,503        )504        # Check505        mock_call.assert_called_with(506            'evpn_prefix.delete_local', 'Invalid arguments detected')507    @mock.patch(508        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',509        mock.MagicMock(return_value=None))510    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')511    def test_evpn_prefix_del_eth_segment(self, mock_call):512        # Prepare test data513        route_type = bgpspeaker.EVPN_ETH_SEGMENT514        route_dist = '65000:100'515        esi = {516            'esi_type': ESI_TYPE_MAC_BASED,517            'mac_addr': 'aa:bb:cc:dd:ee:ff',518            'local_disc': 100,519        }520        ip_addr = '192.168.0.1'521        expected_kwargs = {522            'route_type': route_type,523            'route_dist': route_dist,524            'esi': esi,525            'ip_addr': ip_addr,526        }527        # Test528        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')529        speaker.evpn_prefix_del(530            route_type=route_type,531            route_dist=route_dist,532            esi=esi,533            ip_addr=ip_addr,534        )535        # Check536        mock_call.assert_called_with(537            'evpn_prefix.delete_local', **expected_kwargs)538    @mock.patch(539        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',540        mock.MagicMock(return_value=None))541    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')542    def test_evpn_prefix_del_ip_prefix_route(self, mock_call):543        # Prepare test data544        route_type = bgpspeaker.EVPN_IP_PREFIX_ROUTE545        route_dist = '65000:100'546        ethernet_tag_id = 200547        ip_prefix = '192.168.0.0/24'548        expected_kwargs = {549            'route_type': route_type,550            'route_dist': route_dist,551            'ethernet_tag_id': ethernet_tag_id,552            'ip_prefix': ip_prefix,553        }554        # Test555        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')556        speaker.evpn_prefix_del(557            route_type=route_type,558            route_dist=route_dist,559            ethernet_tag_id=ethernet_tag_id,560            ip_prefix=ip_prefix,561        )562        # Check563        mock_call.assert_called_with(564            'evpn_prefix.delete_local', **expected_kwargs)565    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',566                mock.MagicMock(return_value=None))567    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')568    def test_evpn_prefix_add_pmsi_no_tunnel_info(self, mock_call):569        # Prepare test data570        route_type = bgpspeaker.EVPN_MULTICAST_ETAG_ROUTE571        route_dist = '65000:100'572        ethernet_tag_id = 200573        next_hop = '0.0.0.0'574        ip_addr = '192.168.0.1'575        pmsi_tunnel_type = bgpspeaker.PMSI_TYPE_NO_TUNNEL_INFO576        expected_kwargs = {577            'route_type': route_type,578            'route_dist': route_dist,579            'ethernet_tag_id': ethernet_tag_id,580            'next_hop': next_hop,581            'ip_addr': ip_addr,582            'pmsi_tunnel_type': pmsi_tunnel_type,583        }584        # Test585        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')586        speaker.evpn_prefix_add(587            route_type=route_type,588            route_dist=route_dist,589            ethernet_tag_id=ethernet_tag_id,590            ip_addr=ip_addr,591            pmsi_tunnel_type=pmsi_tunnel_type,592        )593        # Check594        mock_call.assert_called_with(595            'evpn_prefix.add_local', **expected_kwargs)596    @mock.patch(597        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',598        mock.MagicMock(return_value=None))599    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')600    def test_evpn_prefix_add_pmsi_ingress_rep(self, mock_call):601        # Prepare test data602        route_type = bgpspeaker.EVPN_MULTICAST_ETAG_ROUTE603        route_dist = '65000:100'604        ethernet_tag_id = 200605        next_hop = '0.0.0.0'606        ip_addr = '192.168.0.1'607        pmsi_tunnel_type = bgpspeaker.PMSI_TYPE_INGRESS_REP608        expected_kwargs = {609            'route_type': route_type,610            'route_dist': route_dist,611            'ethernet_tag_id': ethernet_tag_id,612            'next_hop': next_hop,613            'ip_addr': ip_addr,614            'pmsi_tunnel_type': pmsi_tunnel_type,615        }616        # Test617        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')618        speaker.evpn_prefix_add(619            route_type=route_type,620            route_dist=route_dist,621            ethernet_tag_id=ethernet_tag_id,622            ip_addr=ip_addr,623            pmsi_tunnel_type=pmsi_tunnel_type,624        )625        # Check626        mock_call.assert_called_with(627            'evpn_prefix.add_local', **expected_kwargs)628    @raises(ValueError)629    @mock.patch(630        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',631        mock.MagicMock(return_value=None))632    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')633    def test_evpn_prefix_add_invalid_pmsi_tunnel_type(self, mock_call):634        # Prepare test data635        route_type = bgpspeaker.EVPN_MULTICAST_ETAG_ROUTE636        route_dist = '65000:100'637        ethernet_tag_id = 200638        next_hop = '0.0.0.0'639        ip_addr = '192.168.0.1'640        pmsi_tunnel_type = 1641        # Test642        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')643        speaker.evpn_prefix_add(644            route_type=route_type,645            route_dist=route_dist,646            ethernet_tag_id=ethernet_tag_id,647            ip_addr=ip_addr,648            pmsi_tunnel_type=pmsi_tunnel_type,649        )650        # Check651        mock_call.assert_called_with(652            'evpn_prefix.add_local', 'Invalid arguments detected')653    @mock.patch(654        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',655        mock.MagicMock(return_value=None))656    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')657    def test_flowspec_prefix_add_ipv4(self, mock_call):658        # Prepare test data659        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_IPV4660        rules = {661            'dst_prefix': '10.60.1.0/24',662        }663        actions = {664            'traffic_marking': {665                'dscp': 24,666            }667        }668        expected_kwargs = {669            'flowspec_family': flowspec_family,670            'rules': rules,671            'actions': actions,672        }673        # Test674        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')675        speaker.flowspec_prefix_add(676            flowspec_family=flowspec_family,677            rules=rules,678            actions=actions)679        # Check680        mock_call.assert_called_with(681            'flowspec.add', **expected_kwargs)682    @mock.patch(683        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',684        mock.MagicMock(return_value=None))685    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')686    def test_flowspec_prefix_add_ipv4_without_actions(self, mock_call):687        # Prepare test data688        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_IPV4689        rules = {690            'dst_prefix': '10.60.1.0/24',691        }692        expected_kwargs = {693            'flowspec_family': flowspec_family,694            'rules': rules,695            'actions': {},696        }697        # Test698        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')699        speaker.flowspec_prefix_add(700            flowspec_family=flowspec_family,701            rules=rules)702        # Check703        mock_call.assert_called_with(704            'flowspec.add', **expected_kwargs)705    @mock.patch(706        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',707        mock.MagicMock(return_value=None))708    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')709    def test_flowspec_prefix_del_ipv4(self, mock_call):710        # Prepare test data711        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_IPV4712        rules = {713            'dst_prefix': '10.60.1.0/24',714        }715        expected_kwargs = {716            'flowspec_family': flowspec_family,717            'rules': rules,718        }719        # Test720        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')721        speaker.flowspec_prefix_del(722            flowspec_family=flowspec_family,723            rules=rules)724        # Check725        mock_call.assert_called_with(726            'flowspec.del', **expected_kwargs)727    @mock.patch(728        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',729        mock.MagicMock(return_value=None))730    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')731    def test_flowspec_prefix_add_vpnv4(self, mock_call):732        # Prepare test data733        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_VPNV4734        route_dist = '65001:100'735        rules = {736            'dst_prefix': '10.70.1.0/24',737        }738        actions = {739            'traffic_marking': {740                'dscp': 24,741            }742        }743        expected_kwargs = {744            'flowspec_family': flowspec_family,745            'route_dist': route_dist,746            'rules': rules,747            'actions': actions,748        }749        # Test750        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')751        speaker.flowspec_prefix_add(752            flowspec_family=flowspec_family,753            route_dist=route_dist,754            rules=rules,755            actions=actions)756        # Check757        mock_call.assert_called_with(758            'flowspec.add_local', **expected_kwargs)759    @mock.patch(760        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',761        mock.MagicMock(return_value=None))762    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')763    def test_flowspec_prefix_del_vpnv4(self, mock_call):764        # Prepare test data765        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_VPNV4766        route_dist = '65001:100'767        rules = {768            'dst_prefix': '10.70.1.0/24',769        }770        expected_kwargs = {771            'flowspec_family': flowspec_family,772            'route_dist': route_dist,773            'rules': rules,774        }775        # Test776        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')777        speaker.flowspec_prefix_del(778            flowspec_family=flowspec_family,779            route_dist=route_dist,780            rules=rules)781        # Check782        mock_call.assert_called_with(783            'flowspec.del_local', **expected_kwargs)784    @mock.patch(785        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',786        mock.MagicMock(return_value=None))787    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')788    def test_flowspec_prefix_add_ipv6(self, mock_call):789        # Prepare test data790        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_IPV6791        rules = {792            'dst_prefix': '2001::3/128/32',793        }794        actions = {795            'traffic_marking': {796                'dscp': 24,797            }798        }799        expected_kwargs = {800            'flowspec_family': flowspec_family,801            'rules': rules,802            'actions': actions,803        }804        # Test805        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')806        speaker.flowspec_prefix_add(807            flowspec_family=flowspec_family,808            rules=rules,809            actions=actions)810        # Check811        mock_call.assert_called_with(812            'flowspec.add', **expected_kwargs)813    @mock.patch(814        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',815        mock.MagicMock(return_value=None))816    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')817    def test_flowspec_prefix_add_ipv6_without_actions(self, mock_call):818        # Prepare test data819        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_IPV6820        rules = {821            'dst_prefix': '2001::3/128/32',822        }823        expected_kwargs = {824            'flowspec_family': flowspec_family,825            'rules': rules,826            'actions': {},827        }828        # Test829        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')830        speaker.flowspec_prefix_add(831            flowspec_family=flowspec_family,832            rules=rules)833        # Check834        mock_call.assert_called_with(835            'flowspec.add', **expected_kwargs)836    @mock.patch(837        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',838        mock.MagicMock(return_value=None))839    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')840    def test_flowspec_prefix_del_ipv6(self, mock_call):841        # Prepare test data842        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_IPV6843        rules = {844            'dst_prefix': '2001::3/128/32',845        }846        expected_kwargs = {847            'flowspec_family': flowspec_family,848            'rules': rules,849        }850        # Test851        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')852        speaker.flowspec_prefix_del(853            flowspec_family=flowspec_family,854            rules=rules)855        # Check856        mock_call.assert_called_with(857            'flowspec.del', **expected_kwargs)858    @mock.patch(859        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',860        mock.MagicMock(return_value=None))861    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')862    def test_flowspec_prefix_add_vpnv6(self, mock_call):863        # Prepare test data864        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_VPNV6865        route_dist = '65001:100'866        rules = {867            'dst_prefix': '2001::3/128/32',868        }869        actions = {870            'traffic_marking': {871                'dscp': 24,872            }873        }874        expected_kwargs = {875            'flowspec_family': flowspec_family,876            'route_dist': route_dist,877            'rules': rules,878            'actions': actions,879        }880        # Test881        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')882        speaker.flowspec_prefix_add(883            flowspec_family=flowspec_family,884            route_dist=route_dist,885            rules=rules,886            actions=actions)887        # Check888        mock_call.assert_called_with(889            'flowspec.add_local', **expected_kwargs)890    @mock.patch(891        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',892        mock.MagicMock(return_value=None))893    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')894    def test_flowspec_prefix_del_vpnv6(self, mock_call):895        # Prepare test data896        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_VPNV6897        route_dist = '65001:100'898        rules = {899            'dst_prefix': '2001::3/128/32',900        }901        expected_kwargs = {902            'flowspec_family': flowspec_family,903            'route_dist': route_dist,904            'rules': rules,905        }906        # Test907        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')908        speaker.flowspec_prefix_del(909            flowspec_family=flowspec_family,910            route_dist=route_dist,911            rules=rules)912        # Check913        mock_call.assert_called_with(914            'flowspec.del_local', **expected_kwargs)915    @mock.patch(916        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',917        mock.MagicMock(return_value=None))918    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')919    def test_flowspec_prefix_add_l2vpn(self, mock_call):920        # Prepare test data921        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_L2VPN922        route_dist = '65001:100'923        rules = {924            'dst_mac': '12:34:56:78:9a:bc',925        }926        actions = {927            'traffic_marking': {928                'dscp': 24,929            }930        }931        expected_kwargs = {932            'flowspec_family': flowspec_family,933            'route_dist': route_dist,934            'rules': rules,935            'actions': actions,936        }937        # Test938        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')939        speaker.flowspec_prefix_add(940            flowspec_family=flowspec_family,941            route_dist=route_dist,942            rules=rules,943            actions=actions)944        # Check945        mock_call.assert_called_with(946            'flowspec.add_local', **expected_kwargs)947    @mock.patch(948        'ryu.services.protocols.bgp.bgpspeaker.BGPSpeaker.__init__',949        mock.MagicMock(return_value=None))950    @mock.patch('ryu.services.protocols.bgp.bgpspeaker.call')951    def test_flowspec_prefix_del_l2vpn(self, mock_call):952        # Prepare test data953        flowspec_family = bgpspeaker.FLOWSPEC_FAMILY_L2VPN954        route_dist = '65001:100'955        rules = {956            'dst_mac': '12:34:56:78:9a:bc',957        }958        expected_kwargs = {959            'flowspec_family': flowspec_family,960            'route_dist': route_dist,961            'rules': rules,962        }963        # Test964        speaker = bgpspeaker.BGPSpeaker(65000, '10.0.0.1')965        speaker.flowspec_prefix_del(966            flowspec_family=flowspec_family,967            route_dist=route_dist,968            rules=rules)969        # Check970        mock_call.assert_called_with(...

test_encryption_client.py

Source:test_encryption_client.py

1# Copyright 2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"). You4# may not use this file except in compliance with the License. A copy of5# the License is located at6#7# http://aws.amazon.com/apache2.0/8#9# or in the "license" file accompanying this file. This file is10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF11# ANY KIND, either express or implied. See the License for the specific12# language governing permissions and limitations under the License.13"""Unit test suite for aws_encryption_sdk.EncryptionSDKClient"""14import warnings15import pytest16from mock import MagicMock17import aws_encryption_sdk18from aws_encryption_sdk import CommitmentPolicy19from aws_encryption_sdk.internal.utils.signature import SignaturePolicy20from aws_encryption_sdk.materials_managers.base import CryptoMaterialsManager21pytestmark = [pytest.mark.unit, pytest.mark.local]22def test_init_defaults():23    test = aws_encryption_sdk.EncryptionSDKClient()24    assert test.config.commitment_policy == CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT25    assert test.config.max_encrypted_data_keys is None26def test_init_success():27    test = aws_encryption_sdk.EncryptionSDKClient(28        commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT,29        max_encrypted_data_keys=1,30    )31    assert test.config.commitment_policy == CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT32    assert test.config.max_encrypted_data_keys == 133@pytest.mark.parametrize("max_encrypted_data_keys", (1, 10, 2 ** 16 - 1, 2 ** 16))34def test_init_valid_max_encrypted_data_keys(max_encrypted_data_keys):35    test = aws_encryption_sdk.EncryptionSDKClient(max_encrypted_data_keys=max_encrypted_data_keys)36    assert test.config.max_encrypted_data_keys == max_encrypted_data_keys37@pytest.mark.parametrize("max_encrypted_data_keys", (0, -1))38def test_init_invalid_max_encrypted_data_keys(max_encrypted_data_keys):39    with pytest.raises(ValueError) as exc_info:40        aws_encryption_sdk.EncryptionSDKClient(max_encrypted_data_keys=max_encrypted_data_keys)41    exc_info.match("max_encrypted_data_keys cannot be less than 1")42def test_client_encrypt(mocker):43    mocker.patch.object(aws_encryption_sdk, "StreamEncryptor")44    cmm = MagicMock(__class__=CryptoMaterialsManager)45    client = aws_encryption_sdk.EncryptionSDKClient(46        commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT, max_encrypted_data_keys=347    )48    kwargs = {}49    kwargs["source"] = b"plaintext"50    kwargs["materials_manager"] = cmm51    client.encrypt(**kwargs)52    expected_kwargs = kwargs.copy()53    expected_kwargs["commitment_policy"] = CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT54    expected_kwargs["signature_policy"] = SignaturePolicy.ALLOW_ENCRYPT_ALLOW_DECRYPT55    expected_kwargs["max_encrypted_data_keys"] = 356    aws_encryption_sdk.StreamEncryptor.assert_called_once_with(**expected_kwargs)57def test_client_decrypt(mocker):58    mocker.patch.object(aws_encryption_sdk, "StreamDecryptor")59    cmm = MagicMock(__class__=CryptoMaterialsManager)60    client = aws_encryption_sdk.EncryptionSDKClient(61        commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT, max_encrypted_data_keys=362    )63    kwargs = {}64    kwargs["source"] = b"ciphertext"65    kwargs["materials_manager"] = cmm66    client.decrypt(**kwargs)67    expected_kwargs = kwargs.copy()68    expected_kwargs["commitment_policy"] = CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT69    expected_kwargs["signature_policy"] = SignaturePolicy.ALLOW_ENCRYPT_ALLOW_DECRYPT70    expected_kwargs["max_encrypted_data_keys"] = 371    aws_encryption_sdk.StreamDecryptor.assert_called_once_with(**expected_kwargs)72@pytest.mark.parametrize("mode_string", ("e", "encrypt", "ENCRYPT"))73def test_client_stream_encrypt(mocker, mode_string):74    mocker.patch.object(aws_encryption_sdk, "StreamEncryptor")75    cmm = MagicMock(__class__=CryptoMaterialsManager)76    client = aws_encryption_sdk.EncryptionSDKClient(commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT)77    kwargs = {}78    kwargs["mode"] = mode_string79    kwargs["source"] = b"plaintext"80    kwargs["materials_manager"] = cmm81    client.stream(**kwargs)82    expected_kwargs = kwargs.copy()83    expected_kwargs.pop("mode")84    expected_kwargs["commitment_policy"] = CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT85    expected_kwargs["signature_policy"] = SignaturePolicy.ALLOW_ENCRYPT_ALLOW_DECRYPT86    expected_kwargs["max_encrypted_data_keys"] = None87    aws_encryption_sdk.StreamEncryptor.assert_called_once_with(**expected_kwargs)88@pytest.mark.parametrize("mode_string", ("d", "decrypt", "DECRYPT"))89def test_client_stream_decrypt(mocker, mode_string):90    mocker.patch.object(aws_encryption_sdk, "StreamDecryptor")91    cmm = MagicMock(__class__=CryptoMaterialsManager)92    client = aws_encryption_sdk.EncryptionSDKClient(commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT)93    kwargs = {}94    kwargs["mode"] = mode_string95    kwargs["source"] = b"ciphertext"96    kwargs["materials_manager"] = cmm97    client.stream(**kwargs)98    expected_kwargs = kwargs.copy()99    expected_kwargs.pop("mode")100    expected_kwargs["commitment_policy"] = CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT101    expected_kwargs["signature_policy"] = SignaturePolicy.ALLOW_ENCRYPT_ALLOW_DECRYPT102    expected_kwargs["max_encrypted_data_keys"] = None103    aws_encryption_sdk.StreamDecryptor.assert_called_once_with(**expected_kwargs)104@pytest.mark.parametrize("method", ("encrypt", "decrypt", "stream"))105@pytest.mark.parametrize("key", ("commitment_policy", "max_encrypted_data_keys"))106def test_client_bad_kwargs(mocker, method, key):107    mocker.patch.object(aws_encryption_sdk, "StreamEncryptor")108    cmm = MagicMock(__class__=CryptoMaterialsManager)109    kwargs = {}110    kwargs[key] = "foobar"111    kwargs["source"] = b"ciphertext"112    kwargs["materials_manager"] = cmm113    client = aws_encryption_sdk.EncryptionSDKClient(commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT)114    with warnings.catch_warnings(record=True) as w:115        warnings.simplefilter("always")116        client.encrypt(**kwargs)117        assert len(w) == 1118        assert issubclass(w[-1].category, UserWarning)119        message = str(w[-1].message)120        assert "Invalid keyword argument" in message...

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.