How to use update_port method in tempest

Best Python code snippet using tempest_python

test_ports_rbac.py

Source:test_ports_rbac.py

...207    @rbac_rule_validation.action(service="neutron",208                                 rules=["get_port", "update_port"],209                                 expected_error_codes=[404, 403])210    @decorators.idempotent_id('afa80981-3c59-42fd-9531-3bcb2cd03711')211    def test_update_port(self):212        with self.override_role():213            self.ports_client.update_port(self.port['id'],214                                          admin_state_up=False)215        self.addCleanup(self.ports_client.update_port, self.port['id'],216                        admin_state_up=True)217    @decorators.idempotent_id('08d70f59-67cb-4fb1-bd6c-a5e59dd5db2b')218    @rbac_rule_validation.action(service="neutron",219                                 rules=["get_port", "update_port",220                                        "update_port:device_owner"],221                                 expected_error_codes=[404, 403, 403])222    def test_update_port_device_owner(self):223        original_device_owner = self.port['device_owner']224        with self.override_role():225            self.ports_client.update_port(226                self.port['id'], device_owner='network:router_interface')227        self.addCleanup(self.ports_client.update_port, self.port['id'],228                        device_owner=original_device_owner)229    @rbac_rule_validation.action(service="neutron",230                                 rules=["get_port", "update_port",231                                        "update_port:mac_address"],232                                 expected_error_codes=[404, 403, 403])233    @decorators.idempotent_id('507140c8-7b14-4d63-b627-2103691d887e')234    def test_update_port_mac_address(self):235        original_mac_address = self.port['mac_address']236        with self.override_role():237            self.ports_client.update_port(238                self.port['id'], mac_address=data_utils.rand_mac_address())239        self.addCleanup(self.ports_client.update_port, self.port['id'],240                        mac_address=original_mac_address)241    @testtools.skipUnless(242        CONF.policy_feature_enabled.update_port_fixed_ips_ip_address_policy,243        '"update_port:fixed_ips:ip_address" must be available in the cloud.')244    @rbac_rule_validation.action(service="neutron",245                                 rules=["get_port", "update_port",246                                        "update_port:fixed_ips:ip_address"],247                                 expected_error_codes=[404, 403, 403])248    @decorators.idempotent_id('c091c825-532b-4c6f-a14f-affd3259c1c3')249    def test_update_port_fixed_ips_ip_address(self):250        # Pick an ip address within the allocation_pools range.251        post_body = {'network': self.network}252        port = self.create_port(**post_body)253        ip_list = self._get_unused_ip_address()254        fixed_ips = [{'ip_address': ip_list[0]}]255        with self.override_role():256            self.ports_client.update_port(port['id'], fixed_ips=fixed_ips)257    @rbac_rule_validation.action(service="neutron",258                                 rules=["get_port", "update_port",259                                        "update_port:port_security_enabled"],260                                 expected_error_codes=[404, 403, 403])261    @decorators.idempotent_id('795541af-6652-4e35-9581-fd58224f7545')262    def test_update_port_security_enabled(self):263        with self.override_role():264            self.ports_client.update_port(self.port['id'],265                                          port_security_enabled=True)266    @utils.requires_ext(extension='binding', service='network')267    @rbac_rule_validation.action(service="neutron",268                                 rules=["get_port", "update_port",269                                        "update_port:binding:host_id"],270                                 expected_error_codes=[404, 403, 403])271    @decorators.idempotent_id('24206a72-0d90-4712-918c-5c9a1ebef64d')272    def test_update_port_binding_host_id(self):273        post_body = {'network': self.network,274                     'binding:host_id': 'rbac_test_host'}275        port = self.create_port(**post_body)276        updated_body = {'port_id': port['id'],277                        'binding:host_id': 'rbac_test_host_updated'}278        with self.override_role():279            self.ports_client.update_port(**updated_body)280    @utils.requires_ext(extension='binding', service='network')281    @rbac_rule_validation.action(service="neutron",282                                 rules=["get_port", "update_port",283                                        "update_port:binding:profile"],284                                 expected_error_codes=[404, 403, 403])285    @decorators.idempotent_id('990ea8d1-9257-4f71-a3bf-d6d0914625c5')286    def test_update_port_binding_profile(self):287        binding_profile = {"foo": "1"}288        post_body = {'network': self.network,289                     'binding:profile': binding_profile}290        port = self.create_port(**post_body)291        new_binding_profile = {"foo": "2"}292        updated_body = {'port_id': port['id'],293                        'binding:profile': new_binding_profile}294        with self.override_role():295            self.ports_client.update_port(**updated_body)296    @rbac_rule_validation.action(service="neutron",297                                 rules=["get_port", "update_port",298                                        "update_port:allowed_address_pairs"],299                                 expected_error_codes=[404, 403, 403])300    @decorators.idempotent_id('729c2151-bb49-4f4f-9d58-3ed8819b7582')301    def test_update_port_allowed_address_pairs(self):302        ip_list = self._get_unused_ip_address()303        # Update allowed address pair attribute of port304        address_pairs = [{'ip_address': ip_list[0],305                          'mac_address': data_utils.rand_mac_address()}]306        post_body = {'network': self.network}307        port = self.create_port(**post_body)308        with self.override_role():309            self.ports_client.update_port(port['id'],310                                          allowed_address_pairs=address_pairs)311    @rbac_rule_validation.action(service="neutron",312                                 rules=["get_port", "delete_port"],313                                 expected_error_codes=[404, 403])314    @decorators.idempotent_id('1cf8e582-bc09-46cb-b32a-82bf991ad56f')315    def test_delete_port(self):316        port = self.create_port(self.network)317        with self.override_role():318            self.ports_client.delete_port(port['id'])319    @rbac_rule_validation.action(service="neutron", rules=["get_port"])320    @decorators.idempotent_id('877ea70d-b000-4af4-9322-0a76b47b7890')321    def test_list_ports(self):322        """List Ports323        RBAC test for the neutron ``list_ports`` function and...

test_nuage_policy_groups_ops.py

Source:test_nuage_policy_groups_ops.py

...85        kwargs = {86            'nuage_policy_groups': [policy_group[0]['ID']],87            'name': 'port-with-vsd-pg'88        }89        self.update_port(port1, **kwargs)90        self.update_port(port2, **kwargs)91        # Then I expext all ports in the show policy group response92        port_present = self._check_port_in_policy_group(port1['id'], policy_group[0]['ID'])93        self.assertTrue(port_present, "Port(%s) assiociated to policy group (%s) is not present" %94                        (port1['id'], policy_group[0]['ID']))95        port_present = self._check_port_in_policy_group(port2['id'], policy_group[0]['ID'])96        self.assertTrue(port_present, "Port(%s) assiociated to policy group (%s) is not present" %97                        (port2['id'], policy_group[0]['ID']))98        # create connectivity VM99        vm_conn = self._create_connectivity_VM(public_network_id=CONF.network.public_network_id,100                                               vsd_l2_subnet=vsd_l2_subnet,101                                               vsd_l2_port=port2)102        floating_ip, the_server = self.floating_ip_tuple103        rslt = self._configure_eth1_server(vm_conn, floating_ip.floating_ip_address)104        # When I spin a VM with this port105        vm1 = self._create_server(name='vm1', network_id=network['id'], port_id=port1['id'])106        vm1_ip_addr = vm1['addresses'][network['name']][0]['addr']107        # These Vm's have connectivity108        # for i in range(5):109        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 10)110        self.assertTrue(connectivity,msg="No ping connectivity in policy group while expected (1)")111        # When I disassociate all ports from the policy group112        kwargs = {113            'nuage_policy_groups': [],114            'name': 'port-without-vsd-pg-1st'115        }116        self.update_port(port1, **kwargs)117        self.update_port(port2, **kwargs)118        # Then these VM's have no more connectivity119        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr,1)120        self.assertFalse(connectivity, msg="Ping connectivity in policy group while NOT expected (1)")121        # When I re-associate all ports with the policy group122        kwargs = {123            'nuage_policy_groups': [policy_group[0]['ID']],124            'name': 'port-with-vsd-pg-2nd'125        }126        self.update_port(port1, **kwargs)127        self.update_port(port2, **kwargs)128        # Then these VM's have again connectivity129        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 10)130        self.assertTrue(connectivity,msg="No ping connectivity in policy group while expected (2)")131        # When I disassociate 1 port from the policy group132        kwargs = {133            'nuage_policy_groups': [],134            'name': 'port-without-vsd-pg-2nd'135        }136        self.update_port(port1, **kwargs)137        # self.update_port(port2, **kwargs)138        # Then these VM's have no more connectivity139        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 1)140        self.assertFalse(connectivity, msg="Ping connectivity in policy group while NOT expected (2)")141        # When I re-associate that port with the policy group142        kwargs = {143            'nuage_policy_groups': [policy_group[0]['ID']],144            'name': 'port-with-vsd-3rd'145        }146        self.update_port(port1, **kwargs)147        # self.update_port(port2, **kwargs)148        # Then these VM's have again connectivity149        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 10)150        self.assertTrue(connectivity,msg="No ping connectivity in policy group while expected (3)")151        # When I disassociate the other port from the policy group152        kwargs = {153            'nuage_policy_groups': [],154            'name': 'port-without-vsd-pg-2nd'155        }156        # self.update_port(port1, **kwargs)157        self.update_port(port2, **kwargs)158        # Then these VM's have no more connectivity159        # time.sleep(5)160        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 1)161        self.assertFalse(connectivity, msg="Ping connectivity in policy group while NOT expected (3)")162        # When I re-associate that port with the policy group163        kwargs = {164            'nuage_policy_groups': [policy_group[0]['ID']],165            'name': 'port-with-vsd-3rd'166        }167        # self.update_port(port1, **kwargs)168        self.update_port(port2, **kwargs)169        # Then these VM's have again connectivity170        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 10)171        self.assertTrue(connectivity,msg="No ping connectivity in policy group while expected (3)")172        #173        #end for loop174        the_floating_ip = self.floating_ips.pop()175        self.floating_ips_client.delete_floatingip(the_floating_ip['id'])176        # self.servers_client.delete_server(vm_conn['id'])177        # self.servers_client.delete_server(vm1['id'])178        self._clear_connectivity_vm_interfaces(self.conn_router_id, self.conn_subnet_id, self.conn_port_id)179    def test_e2e_l3_vm_connectivity_port_to_policygroup(self):180        # Given I have a VSD-L2-Managed-Subnet in openstack with a VSD creeated policy group181        vsd_l3_subnet, vsd_l3_domain = self._create_vsd_l3_managed_subnet()182        network, subnet = self._create_os_l3_vsd_managed_subnet(vsd_l3_subnet)183        policy_group = self.nuage_vsd_client.create_policygroup(constants.DOMAIN,184                                                                vsd_l3_domain[0]['ID'],185                                                                name='myVSD-l3-policygrp',186                                                                type='SOFTWARE',187                                                                extra_params=None)188        # And the policy group has and ingress/egress policy with rules allowing PING189        self._prepare_l3_security_group_entries(policy_group[0]['ID'],190                                                vsd_l3_domain[0]['ID'],191                                                defaultAllowIP=False)192        # When I retrieve the VSD-L2-Managed-Subnet193        policy_group_list = self.nuage_network_client.list_nuage_policy_group_for_subnet(subnet['id'])194        # I expect the policyGroup in my list195        pg_present = self._check_policy_group_in_list(policy_group[0]['ID'], policy_group_list)196        self.assertTrue(pg_present,"Did not find vsd policy group in policy group list")197        # And it has no external ID198        self.assertIsNone(policy_group[0]['externalID'],199                          "Policy Group has an external ID, while it should not")200        show_pg = self.nuage_network_client.show_nuage_policy_group(policy_group[0]['ID'])201        # When I create 2 ports in the subnet202        port1 = self.create_port(network)203        port2 = self.create_port(network)204        # port3 = self.create_port(network)205        # public_network = self.admin_client.show_network(CONF.network.public_network_id)206        # port4 = self.create_port(public_network['network'])207        # And I associate the port with the policy group208        kwargs = {209            'nuage_policy_groups': [policy_group[0]['ID']],210            'name': 'port-with-vsd-pg'211        }212        self.update_port(port1, **kwargs)213        self.update_port(port2, **kwargs)214        # Then I expext the port in the show policy group response215        if CONF.nuage_sut.nuage_plugin_mode != 'ml2' :216            port_present = self._check_port_in_policy_group(port1['id'], policy_group[0]['ID'])217            self.assertTrue(port_present, "Port(%s) assiociated to policy group (%s) is not present" %218                            (port1['id'], policy_group[0]['ID']))219            port_present = self._check_port_in_policy_group(port2['id'], policy_group[0]['ID'])220            self.assertTrue(port_present, "Port(%s) assiociated to policy group (%s) is not present" %221                            (port2['id'], policy_group[0]['ID']))222        # create connectivity VM223        vm_conn = self._create_connectivity_VM(public_network_id=CONF.network.public_network_id,224                                               vsd_l2_subnet=vsd_l3_subnet,225                                               vsd_l2_port=port2)226        floating_ip, the_server = self.floating_ip_tuple227        rslt = self._configure_eth1_server(vm_conn, floating_ip.floating_ip_address)228        # When I spin a VM with this port229        vm1 = self._create_server(name='vm1', network_id=network['id'], port_id=port1['id'])230        # vm2 = self._create_server(name='vm2', network_id=network['id'], port_id=port2['id'])231        # These Vm's have connectivity232        vm1_ip_addr = vm1['addresses'][network['name']][0]['addr']233        # for i in range(5):234        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 10)235        self.assertTrue(connectivity,msg="No ping connectivity in policy group while expected (1)")236        # When I disassociate the port from the policy group237        kwargs = {238            'nuage_policy_groups': [],239            'name': 'port-without-vsd-pg-1st'240        }241        self.update_port(port1, **kwargs)242        self.update_port(port2, **kwargs)243        # Then these VM's have no more connectivity244        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr,1)245        self.assertFalse(connectivity, msg="Ping connectivity in policy group while NOT expected (1)")246        # When I re-associate the port with the policy group247        kwargs = {248            'nuage_policy_groups': [policy_group[0]['ID']],249            'name': 'port-with-vsd-pg-2nd'250        }251        self.update_port(port1, **kwargs)252        self.update_port(port2, **kwargs)253        # Then these VM's have again connectivity254        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 10)255        self.assertTrue(connectivity,msg="No ping connectivity in policy group while expected (2)")256        # When I disassociate the port from the policy group257        kwargs = {258            'nuage_policy_groups': [],259            'name': 'port-without-vsd-pg-2nd'260        }261        self.update_port(port1, **kwargs)262        self.update_port(port2, **kwargs)263        # Then these VM's have no more connectivity264        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 1)265        self.assertFalse(connectivity, msg="Ping connectivity in policy group while NOT expected (3)")266        # When I re-associate the port with the policy group267        kwargs = {268            'nuage_policy_groups': [policy_group[0]['ID']],269            'name': 'port-with-vsd-3rd'270        }271        self.update_port(port1, **kwargs)272        self.update_port(port2, **kwargs)273        # Then these VM's have again connectivity274        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 10)275        self.assertTrue(connectivity,msg="No ping connectivity in policy group while expected (4)")276        # When I disassociate the port from the policy group277        kwargs = {278            'nuage_policy_groups': [],279            'name': 'port-without-vsd-pg-2nd'280        }281        self.update_port(port1, **kwargs)282        self.update_port(port2, **kwargs)283        # Then these VM's have no more connectivity284        connectivity = self._check_vm_policy_group_ping(vm_conn, floating_ip.floating_ip_address, vm1_ip_addr, 1)285        self.assertFalse(connectivity, msg="Ping connectivity in policy group while NOT expected (5)")286        # When I re-associate the port with the policy group287        kwargs = {288            'nuage_policy_groups': [policy_group[0]['ID']],289            'name': 'port-with-vsd-3rd'290        }291        self.update_port(port1, **kwargs)292        self.update_port(port2, **kwargs)293        the_floating_ip = self.floating_ips.pop()294        self.floating_ips_client.delete_floatingip(the_floating_ip['id'])295        # the_server = self.servers.pop()296        self.servers_client.delete_server(vm_conn['id'])297        self.servers_client.delete_server(vm1['id'])...

test_microservice.py

Source:test_microservice.py

1from dl_mqtt_clients.func._general_func import simple_logging_format2from dl_mqtt_clients.microservice import microservice3from dl_mqtt_clients import configurations4from dl_mqtt_clients.payload_serializers.JSONSerializer import JSONSerializer5import mock 6import pytest7from pytest_mock import mocker 8import logging9import random, string, json10logging.basicConfig(format=simple_logging_format, level=logging.INFO)11logger = logging.getLogger()12microservice.__abstractmethods__ = frozenset()13@pytest.fixture(scope='session')14def mqtt_config_d():15	mqtt_config_d = {'host': 'localhost',16					'port': 1883,17					'sub_topics':['model_update'],18					'pub_topic':'global/model',19					'ssl_protocol': 'tls_12',20					'keepalive': 10,21					'sub_qos': 1,22					'pub_qos': 0,23					'mqtt_protocol': 311}24	return mqtt_config_d25def test_init(mqtt_config_d):26	"""tests passing None for serialier on init"""27	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 28					update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 29					serializer=None)30def test_init_serializer(mqtt_config_d):31	"""tests passing a serializer on init"""32	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 33						update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 34						serializer=JSONSerializer())35def test_init_non_configuration_obj_1(mqtt_config_d):36	"""tests passing a non MQTTConnectionConfig object to data_mqtt_config param"""37	with pytest.raises(AssertionError):38		m = microservice(data_mqtt_config=mqtt_config_d, 39						update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 40						serializer=None)41def test_init_non_configuration_obj_2(mqtt_config_d):42	"""tests passing a non MQTTConnectionConfig object to update_mqtt_config param"""43	with pytest.raises(AssertionError):44		m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d),45						update_mqtt_config=mqtt_config_d, 46						serializer=None)47def test_init_logger_name(mqtt_config_d):48	"""tests the init parameter for logger_name"""49	name = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(10))50	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 51					update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 52					serializer=None,53					logger_name=name)54	assert(name == m.logger.name)55def test_update(mqtt_config_d, mocker):56	"""tests the update mqtt method by generating random port to update"""57	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 58					update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 59					serializer=None)60	update_port = random.randint(1000, 9999)61	mocker.patch.object(m.dataClient.client, 'reconnect')62	m._update_mqtt(update_d = {"port": update_port})63	assert(m.dataClient.mqtt_config.port()==update_port)64def test_update_bad_val(mqtt_config_d, mocker):65	"""tests a bad type update"""66	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 67					update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 68					serializer=None)69	prev_port = m.dataClient.mqtt_config.port()70	update_port = random.randint(1000, 9999)71	#mocker.patch.object(m.dataClient.client, 'reconnect')72	m._update_mqtt(update_d = {"port": str(update_port)})73	assert(m.dataClient.mqtt_config.port()== prev_port)74def test_on_msg_update(mqtt_config_d, mocker):75	"""tests the update_mqtt call of the microservice class"""76	update_port = random.randint(1000, 9999)77	class msg:78		def __init__(self):79			self.payload = json.dumps({'port': update_port}).encode('utf-8')80			self.topic = "my/sub/topic"81	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 82					update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 83					serializer=None)84	myMsg = msg()85	mocker.patch.object(m.dataClient.client, 'reconnect')86	m._on_message_topic_upd(client=None, userdata=None, msg=myMsg)87	assert(m.dataClient.mqtt_config.port()==update_port)88def test_on_msg_update_bad_json(mqtt_config_d, mocker):89	"""tests the on_message_upd method of the microservice class, ensuring it can properly decode utf8 encoded strings"""90	class msg:91		def __init__(self):92			self.payload = "{inval: json".encode('utf-8')93			self.topic = "my/sub/topic"94	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 95					update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 96					serializer=None)97	myMsg = msg()98	prev_port = m.dataClient.mqtt_config.port()99	mocker.patch.object(m.dataClient.client, 'reconnect')100	m._on_message_topic_upd(client=None, userdata=None, msg=myMsg)101	assert(prev_port==m.dataClient.mqtt_config.port())102def test_connect_clients(mqtt_config_d, mocker):103	"""tests the logic for connecting clients"""104	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 105					update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 106					serializer=None)107	mocker.patch.object(m.dataClient, 'connect_client')108	mocker.patch.object(m.configClient, 'connect_client')109	m.connect_clients()110def test_connect_clients_valid_error(mqtt_config_d, mocker):111	"""tests to see client connecting with a validation error occuring during _validate"""112	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 113					update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 114					serializer=None)115	mocker.patch.object(m, '_validate', side_effect=AssertionError)116	with pytest.raises(AssertionError):117		m.connect_clients()118def test_on_message_abstract(mqtt_config_d):119	"""simply tests if on_message method is callable"""120	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 121					update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 122					serializer=None)123	m._on_message(client=None, userdata=None, msg=None)124def test_validate(mqtt_config_d):125	"""simply testsw if _validate method is callable"""126	m = microservice(data_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 127					update_mqtt_config=configurations.MQTTConnectionConfig(mqtt_config_d), 128					serializer=None)...

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.