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)...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!