KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use deploy_resource method in localstack
Best Python code snippet using localstack_python
test_neutron.py
Source:test_neutron.py 
...28 project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)29 n = openstack::Network(provider=p, name="test_net", project=project, external=true)30 """31 )32 n1 = project.deploy_resource("openstack::Network", name="test_net")33 networks = neutron.list_networks(name=n1.name)["networks"]34 assert len(networks) == 135 assert networks[0]["router:external"]36 ctx = project.deploy(n1)37 assert ctx.status == inmanta.const.ResourceState.deployed38 project.compile(39 """40 import unittest41 import openstack42 tenant = std::get_env("OS_PROJECT_NAME")43 p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),44 password=std::get_env("OS_PASSWORD"), tenant=tenant)45 project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)46 n = openstack::Network(provider=p, name="test_net", project=project, purged=true)47 """48 )49 n1 = project.get_resource("openstack::Network", name="test_net")50 ctx = project.deploy(n1)51 assert ctx.status == inmanta.const.ResourceState.deployed52 networks = neutron.list_networks(name=n1.name)["networks"]53 assert len(networks) == 054 finally:55 # cleanup56 networks = neutron.list_networks(name=n1.name)["networks"]57 if len(networks) > 0:58 for network in networks:59 neutron.delete_network(network["id"])60def test_subnet(project, neutron):61 name = "inmanta_unit_test"62 try:63 project.compile(64 """65 import unittest66 import openstack67 tenant = std::get_env("OS_PROJECT_NAME")68 p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),69 password=std::get_env("OS_PASSWORD"), tenant=tenant)70 project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)71 n = openstack::Network(provider=p, name="%(name)s", project=project)72 subnet = openstack::Subnet(provider=p, project=project, network=n, dhcp=true, name="%(name)s",73 network_address="10.255.255.0/24", dns_servers=["8.8.8.8", "8.8.4.4"])74 """75 % {"name": name}76 )77 net = project.deploy_resource("openstack::Network")78 subnet = project.deploy_resource("openstack::Subnet")79 subnets = neutron.list_subnets(name=subnet.name)["subnets"]80 assert len(subnets) == 181 assert len(neutron.list_networks(name=net.name)["networks"]) == 182 os_subnet = subnets[0]83 assert len(os_subnet["dns_nameservers"]) == 284 assert os_subnet["gateway_ip"] == "10.255.255.1"85 project.compile(86 """87 import unittest88 import openstack89 tenant = std::get_env("OS_PROJECT_NAME")90 p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),91 password=std::get_env("OS_PASSWORD"), tenant=tenant)92 project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)93 n = openstack::Network(provider=p, name="%(name)s", project=project, purged=true)94 subnet = openstack::Subnet(provider=p, project=project, network=n, dhcp=true, name="%(name)s",95 network_address="10.255.255.0/24", purged=true)96 """97 % {"name": name}98 )99 net = project.deploy_resource("openstack::Network")100 subnet = project.deploy_resource("openstack::Subnet")101 assert len(neutron.list_subnets(name=subnet.name)["subnets"]) == 0102 assert len(neutron.list_networks(name=net.name)["networks"]) == 0103 finally:104 # cleanup105 networks = neutron.list_subnets(name=name)["subnets"]106 if len(networks) > 0:107 for network in networks:108 neutron.delete_subnet(network["id"])109 networks = neutron.list_networks(name=name)["networks"]110 if len(networks) > 0:111 for network in networks:112 neutron.delete_network(network["id"])113def test_router(project, neutron):114 name = "inmanta_unit_test"115 external = None116 for network in neutron.list_networks()["networks"]:117 if network["router:external"]:118 external = network119 assert external is not None, "This test requires an external network to be defined."120 project.compile(121 """122import unittest123import openstack124tenant = std::get_env("OS_PROJECT_NAME")125p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),126 password=std::get_env("OS_PASSWORD"), tenant=tenant)127project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)128ext = openstack::Network(provider=p, name="%(external)s", project=project, managed=false)129router = openstack::Router(provider=p, project=project, name="%(name)s", ext_gateway=ext, admin_state="up")130n = openstack::Network(provider=p, name="%(name)s", project=project)131subnet = openstack::Subnet(provider=p, project=project, network=n, dhcp=true, name="%(name)s",132 network_address="10.255.255.0/24")133router.subnets = subnet134 """135 % {"external": external["name"], "name": name}136 )137 net = project.deploy_resource("openstack::Network")138 subnet = project.deploy_resource("openstack::Subnet")139 project.deploy_resource("openstack::Router")140 routers = neutron.list_routers(name=name)["routers"]141 assert len(routers) == 1142 assert len(neutron.list_networks(name=net.name)["networks"]) == 1143 assert len(neutron.list_subnets(name=subnet.name)["subnets"]) == 1144 ports = neutron.list_ports(device_id=routers[0]["id"])["ports"]145 assert len(ports) == 2146 project.compile(147 """148import unittest149import openstack150tenant = std::get_env("OS_PROJECT_NAME")151p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),152 password=std::get_env("OS_PASSWORD"), tenant=tenant)153project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)154ext = openstack::Network(provider=p, name="%(external)s", project=project, managed=false)155router = openstack::Router(provider=p, project=project, name="%(name)s", ext_gateway=ext, admin_state="up", purged=true)156n = openstack::Network(provider=p, name="%(name)s", project=project, purged=true)157subnet = openstack::Subnet(provider=p, project=project, network=n, dhcp=true, name="%(name)s",158 network_address="10.255.255.0/24", purged=true)159router.subnets = subnet160 """161 % {"external": external["name"], "name": name}162 )163 project.deploy_resource("openstack::Router")164 routers = neutron.list_routers(name=name)["routers"]165 assert len(routers) == 0166 net = project.deploy_resource("openstack::Network")167 subnet = project.deploy_resource("openstack::Subnet")168 assert len(neutron.list_subnets(name=subnet.name)["subnets"]) == 0169 assert len(neutron.list_networks(name=net.name)["networks"]) == 0170 routers = neutron.list_routers(name=name)["routers"]171 if len(routers) > 0:172 for router in routers:173 neutron.delete_router(router["id"])174 networks = neutron.list_subnets(name=name)["subnets"]175 if len(networks) > 0:176 for network in networks:177 neutron.delete_subnet(network["id"])178 networks = neutron.list_networks(name=name)["networks"]179 if len(networks) > 0:180 for network in networks:181 neutron.delete_network(network["id"])182def test_router_port(project, neutron):183 name = "inmanta_unit_test"184 external = None185 for network in neutron.list_networks()["networks"]:186 if network["router:external"]:187 external = network188 assert external is not None, "This test requires an external network to be defined."189 project.compile(190 """191import unittest192import openstack193tenant = std::get_env("OS_PROJECT_NAME")194p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),195 password=std::get_env("OS_PASSWORD"), tenant=tenant)196project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)197ext = openstack::Network(provider=p, name="%(external)s", project=project, managed=false)198router = openstack::Router(provider=p, project=project, name="%(name)s", ext_gateway=ext, admin_state="up")199n = openstack::Network(provider=p, name="%(name)s", project=project)200subnet = openstack::Subnet(provider=p, project=project, network=n, dhcp=true, name="%(name)s",201 network_address="10.255.255.0/24")202openstack::RouterPort(provider=p, project=project, name="%(name)s", router=router, subnet=subnet, address="10.255.255.200")203 """204 % {"external": external["name"], "name": name}205 )206 net = project.deploy_resource("openstack::Network")207 subnet = project.deploy_resource("openstack::Subnet")208 project.deploy_resource("openstack::Router")209 project.deploy_resource("openstack::RouterPort")210 routers = neutron.list_routers(name=name)["routers"]211 assert len(routers) == 1212 assert len(neutron.list_networks(name=net.name)["networks"]) == 1213 assert len(neutron.list_subnets(name=subnet.name)["subnets"]) == 1214 ports = neutron.list_ports(device_id=routers[0]["id"])["ports"]215 assert len(ports) == 2216 # remove217 project.compile(218 """219import unittest220import openstack221tenant = std::get_env("OS_PROJECT_NAME")222p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),223 password=std::get_env("OS_PASSWORD"), tenant=tenant)224project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)225ext = openstack::Network(provider=p, name="%(external)s", project=project, managed=false)226router = openstack::Router(provider=p, project=project, name="%(name)s", ext_gateway=ext, admin_state="up", purged=true)227n = openstack::Network(provider=p, name="%(name)s", project=project, purged=true)228subnet = openstack::Subnet(provider=p, project=project, network=n, dhcp=true, name="%(name)s", purged=true,229 network_address="10.255.255.0/24")230openstack::RouterPort(provider=p, project=project, name="%(name)s", router=router, subnet=subnet, address="10.255.255.200",231 purged=true)232 """233 % {"external": external["name"], "name": name}234 )235 project.deploy_resource("openstack::RouterPort")236 project.deploy_resource("openstack::Router")237 subnet = project.deploy_resource("openstack::Subnet")238 net = project.deploy_resource("openstack::Network")239 routers = neutron.list_routers(name=name)["routers"]240 assert len(routers) == 0241 assert len(neutron.list_networks(name=net.name)["networks"]) == 0242 assert len(neutron.list_subnets(name=subnet.name)["subnets"]) == 0243def test_update_rules_in_security_group(project, openstack, neutron):244 os_project = openstack.get_project("test")245 resource_name_sec_group1 = os_project.get_resource_name("sec_group1")246 resource_name_sec_group2 = os_project.get_resource_name("sec_group2")247 resource_name_sec_group3 = os_project.get_resource_name("sec_group3")248 project.compile(249 f"""250import unittest251import openstack252tenant = "{os_project._tenant}"253p = openstack::Provider(254 name="test",255 connection_url="{os_project._auth_url}",256 username="{os_project._username}",257 password="{os_project._password}",258 tenant=tenant)259project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)260sg_base = openstack::SecurityGroup(provider=p, project=project, name="{resource_name_sec_group1}", description="test")261openstack::IPrule(262 group=sg_base,263 direction="ingress",264 ip_protocol="tcp",265 port_min=200,266 port_max=201,267 remote_prefix="0.0.0.0/0"268)269openstack::IPrule(270 group=sg_base,271 direction="ingress",272 ip_protocol="tcp",273 port_min=300,274 port_max=301,275 remote_prefix="0.0.0.0/0"276)277openstack::IPrule(278 group=sg_base,279 direction="ingress",280 ip_protocol="tcp",281 port_min=400,282 port_max=401,283 remote_prefix="0.0.0.0/0"284)285openstack::IPrule(286 group=sg_base,287 direction="ingress",288 ip_protocol="tcp",289 port_min=500,290 port_max=501,291 remote_prefix="0.0.0.0/0"292)293openstack::GroupRule(294 group=sg_base,295 direction="ingress",296 ip_protocol="tcp",297 port=222,298 remote_group=sg_base2299)300sg_base2 = openstack::SecurityGroup(provider=p, project=project, name="{resource_name_sec_group2}", description="test")301openstack::IPrule(302 group=sg_base2,303 direction="ingress",304 ip_protocol="tcp",305 port=8080,306 remote_prefix="0.0.0.0/0"307)308 """309 )310 @dataclass(frozen=True)311 class Rule:312 direction: str313 protocol: str314 port_range_min: int315 port_range_max: int316 remote_ip_prefix: Optional[str]317 remote_group_id: Optional[uuid.UUID]318 @classmethod319 def from_dict(cls, dct: Dict):320 return Rule(321 direction=dct["direction"],322 protocol=dct["protocol"],323 port_range_min=dct["port_range_min"],324 port_range_max=dct["port_range_max"],325 remote_ip_prefix=dct["remote_ip_prefix"],326 remote_group_id=dct["remote_group_id"],327 )328 project.deploy_resource("openstack::SecurityGroup", name=resource_name_sec_group2)329 project.deploy_resource("openstack::SecurityGroup", name=resource_name_sec_group1)330 sgs = neutron.list_security_groups(331 project_id=os_project.project_object.id, name=resource_name_sec_group1332 )["security_groups"]333 assert len(sgs) == 1334 actual_rules = set(335 Rule.from_dict(dct)336 for dct in sgs[0]["security_group_rules"]337 if dct["ethertype"] == "IPv4"338 )339 assert len(actual_rules) == 5340 assert Rule("ingress", "tcp", 200, 201, "0.0.0.0/0", None) in actual_rules341 assert Rule("ingress", "tcp", 300, 301, "0.0.0.0/0", None) in actual_rules342 assert Rule("ingress", "tcp", 400, 401, "0.0.0.0/0", None) in actual_rules343 assert Rule("ingress", "tcp", 500, 501, "0.0.0.0/0", None) in actual_rules344 remote_group_id1 = [r.remote_group_id for r in actual_rules if r.remote_group_id][0]345 assert Rule("ingress", "tcp", 222, 222, None, remote_group_id1) in actual_rules346 project.compile(347 f"""348import unittest349import openstack350tenant = "{os_project._tenant}"351p = openstack::Provider(352 name="test",353 connection_url="{os_project._auth_url}",354 username="{os_project._username}",355 password="{os_project._password}",356 tenant=tenant)357project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)358sg_base = openstack::SecurityGroup(provider=p, project=project, name="{resource_name_sec_group1}", description="test")359# Update direction360openstack::IPrule(361 group=sg_base,362 direction="egress",363 ip_protocol="tcp",364 port_min=200,365 port_max=201,366 remote_prefix="0.0.0.0/0"367)368# Update protocol369openstack::IPrule(370 group=sg_base,371 direction="ingress",372 ip_protocol="udp",373 port_min=300,374 port_max=301,375 remote_prefix="0.0.0.0/0"376)377# Update port378openstack::IPrule(379 group=sg_base,380 direction="ingress",381 ip_protocol="tcp",382 port_min=900,383 port_max=901,384 remote_prefix="0.0.0.0/0"385)386# Update remote prefix387openstack::IPrule(388 group=sg_base,389 direction="ingress",390 ip_protocol="tcp",391 port_min=500,392 port_max=501,393 remote_prefix="192.168.0.0/24"394)395# Attach new remote_group396openstack::GroupRule(397 group=sg_base,398 direction="ingress",399 ip_protocol="tcp",400 port=222,401 remote_group=sg_base3402)403sg_base3 = openstack::SecurityGroup(provider=p, project=project, name="{resource_name_sec_group3}", description="test")404openstack::IPrule(405 group=sg_base3,406 direction="ingress",407 ip_protocol="tcp",408 port=80,409 remote_prefix="0.0.0.0/0"410)411 """412 )413 project.deploy_resource("openstack::SecurityGroup", name=resource_name_sec_group3)414 project.deploy_resource("openstack::SecurityGroup", name=resource_name_sec_group1)415 sgs = neutron.list_security_groups(416 project_id=os_project.project_object.id, name=resource_name_sec_group1417 )["security_groups"]418 assert len(sgs) == 1419 actual_rules = set(420 Rule.from_dict(dct)421 for dct in sgs[0]["security_group_rules"]422 if dct["ethertype"] == "IPv4"423 )424 assert len(actual_rules) == 5425 assert Rule("egress", "tcp", 200, 201, "0.0.0.0/0", None) in actual_rules426 assert Rule("ingress", "udp", 300, 301, "0.0.0.0/0", None) in actual_rules427 assert Rule("ingress", "tcp", 900, 901, "0.0.0.0/0", None) in actual_rules428 assert Rule("ingress", "tcp", 500, 501, "192.168.0.0/24", None) in actual_rules429 remote_group_id2 = [r.remote_group_id for r in actual_rules if r.remote_group_id][0]430 assert remote_group_id1 != remote_group_id2431 assert Rule("ingress", "tcp", 222, 222, None, remote_group_id2) in actual_rules432def test_security_group(project, neutron):433 name = "inmanta_unit_test"434 sgs = neutron.list_security_groups(name=name)435 if len(sgs["security_groups"]) > 0:436 neutron.delete_security_group(sgs["security_groups"][0]["id"])437 project.compile(438 """439import unittest440import openstack441tenant = std::get_env("OS_PROJECT_NAME")442p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),443 password=std::get_env("OS_PASSWORD"), tenant=tenant)444project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)445sg_base = openstack::SecurityGroup(provider=p, project=project, name="%(name)s", description="Clearwater base")446openstack::IPrule(group=sg_base, direction="egress", ip_protocol="all", remote_prefix="0.0.0.0/0")447openstack::IPrule(group=sg_base, direction="ingress", ip_protocol="udp", port_min=161, port_max=162,448 remote_prefix="0.0.0.0/0")449openstack::IPrule(group=sg_base, direction="ingress", ip_protocol="tcp", port_min=161, port_max=162,450 remote_prefix="0.0.0.0/0")451 """452 % {"name": name}453 )454 project.deploy_resource("openstack::SecurityGroup")455 sgs = neutron.list_security_groups(name=name)456 assert len(sgs["security_groups"]) == 1457 assert (458 len(459 [460 x461 for x in sgs["security_groups"][0]["security_group_rules"]462 if x["ethertype"] == "IPv4"463 ]464 )465 == 3466 )467 project.compile(468 """469import unittest470import openstack471tenant = std::get_env("OS_PROJECT_NAME")472p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),473 password=std::get_env("OS_PASSWORD"), tenant=tenant)474project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)475sg_base = openstack::SecurityGroup(provider=p, project=project, name="%(name)s", description="Clearwater base")476openstack::IPrule(group=sg_base, direction="egress", ip_protocol="all", remote_prefix="0.0.0.0/0")477openstack::IPrule(group=sg_base, direction="ingress", ip_protocol="udp", port_min=161, port_max=162,478 remote_prefix="0.0.0.0/0")479openstack::IPrule(group=sg_base, direction="ingress", ip_protocol="tcp", port_min=161, port_max=162,480 remote_prefix="0.0.0.0/0")481 """482 % {"name": name}483 )484 # deploy a second time485 project.deploy_resource("openstack::SecurityGroup")486 sgs = neutron.list_security_groups(name=name)487 assert len(sgs["security_groups"]) == 1488 assert (489 len(490 [491 x492 for x in sgs["security_groups"][0]["security_group_rules"]493 if x["ethertype"] == "IPv4"494 ]495 )496 == 3497 )498 # purge it499 project.compile(500 """501import unittest502import openstack503tenant = std::get_env("OS_PROJECT_NAME")504p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),505 password=std::get_env("OS_PASSWORD"), tenant=tenant)506project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)507sg_base = openstack::SecurityGroup(provider=p, project=project, name="%(name)s", description="Clearwater base", purged=true)508openstack::IPrule(group=sg_base, direction="egress", ip_protocol="all", remote_prefix="0.0.0.0/0")509openstack::IPrule(group=sg_base, direction="ingress", ip_protocol="udp", port_min=161, port_max=162,510 remote_prefix="0.0.0.0/0")511 """512 % {"name": name}513 )514 project.deploy_resource("openstack::SecurityGroup")515 sgs = neutron.list_security_groups(name=name)516 assert len(sgs["security_groups"]) == 0517@pytest.mark.skip(reason="This tsest is currently broken and needs to be fixed")518def test_security_group_vm(project, neutron, nova):519 name = "inmanta-unit-test"520 key = (521 "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQCsiYV4Cr2lD56bkVabAs2i0WyGSjJbuNHP6IDf8Ru3Pg7DJkz0JaBmETHNjIs+yQ98DNkwH9gZX0"522 "gfrSgX0YfA/PwTatdPf44dwuwWy+cjS2FAqGKdLzNVwLfO5gf74nit4NwATyzakoojHn7YVGnd9ScWfwFNd5jQ6kcLZDq/1w== "523 "bart@wolf.inmanta.com"524 )525 project.compile(526 """527import unittest528import openstack529import ssh530tenant = std::get_env("OS_PROJECT_NAME")531p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),532 password=std::get_env("OS_PASSWORD"), tenant=tenant)533project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)534sg_mgmt = openstack::SecurityGroup(provider=p, project=project, name="%(name)s", description="Test Mgmt SG")535openstack::IPrule(group=sg_mgmt, direction="egress", ip_protocol="all", remote_prefix="0.0.0.0/0")536openstack::IPrule(group=sg_mgmt, direction="ingress", ip_protocol="icmp", remote_prefix="0.0.0.0/0")537openstack::IPrule(group=sg_mgmt, direction="ingress", ip_protocol="tcp", port=22, remote_prefix="0.0.0.0/0")538openstack::IPrule(group=sg_mgmt, direction="ingress", ip_protocol="all", remote_prefix="0.0.0.0/0")539os = std::OS(name="cirros", version=0.4, family=std::linux)540key = ssh::Key(name="%(name)s", public_key="%(key)s")541net = openstack::Network(provider=p, project=project, name="%(name)s")542subnet = openstack::Subnet(provider=p, project=project, network=net, dhcp=true, name="%(name)s",543 network_address="10.255.255.0/24")544vm = openstack::Host(provider=p, project=project, key_pair=key, name="%(name)s", os=os,545 image=openstack::find_image(p, os), flavor=openstack::find_flavor(p, 1, 0.5), user_data="", subnet=subnet)546vm.vm.security_groups=[sg_mgmt]547vm2 = openstack::Host(provider=p, project=project, key_pair=key, name="%(name)s-2", os=os,548 image=openstack::find_image(p, os), flavor=openstack::find_flavor(p, 1, 0.5), user_data="", subnet=subnet)549vm2.vm.security_groups=[sg_mgmt]550 """551 % {"name": name, "key": key}552 )553 sg1 = project.get_resource("openstack::SecurityGroup", name=name)554 ctx = project.deploy(sg1)555 assert ctx.status == inmanta.const.ResourceState.deployed556 assert neutron.list_security_groups(name=name)557 n1 = project.get_resource("openstack::Network", name=name)558 ctx = project.deploy(n1)559 assert ctx.status == inmanta.const.ResourceState.deployed560 s1 = project.get_resource("openstack::Subnet", name=name)561 ctx = project.deploy(s1)562 assert ctx.status == inmanta.const.ResourceState.deployed563 h1 = project.get_resource("openstack::VirtualMachine", name=name)564 ctx = project.deploy(h1)565 assert ctx.status == inmanta.const.ResourceState.deployed566 hp1 = project.get_resource("openstack::HostPort", name=name + "_eth0")567 ctx = project.deploy(hp1)568 assert ctx.status == inmanta.const.ResourceState.deployed569 h1 = project.get_resource("openstack::VirtualMachine", name=name + "-2")570 ctx = project.deploy(h1)571 assert ctx.status == inmanta.const.ResourceState.deployed572 hp1 = project.get_resource("openstack::HostPort", name=name + "-2_eth0")573 ctx = project.deploy(hp1)574 assert ctx.status == inmanta.const.ResourceState.deployed575def test_scoping_on_security_group_resource(project, openstack, neutron):576 """577 Verify that the SecurityGroup resource respects the scope of the project.578 """579 project1 = openstack.get_project("inmanta_unit_test1")580 project2 = openstack.get_project("inmanta_unit_test2")581 description_sec_group1 = "test"582 description_sec_group2 = "other description"583 resource_name = project1.get_resource_name("sec_group")584 # Make sure resource exists in project1585 project1.neutron.create_security_group(586 {587 "security_group": {588 "name": resource_name,589 "description": description_sec_group1,590 }591 }592 )593 # Deploy resource with same name in other project594 project.compile(595 f"""596import unittest597import openstack598tenant = "{project2._tenant}"599p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),600 password=std::get_env("OS_PASSWORD"), tenant=tenant)601project = openstack::Project(provider=p, name=tenant, description="test", enabled=true, managed=false)602rule1 = openstack::IPrule(ip_protocol="all", port=0, direction="egress", remote_prefix="0.0.0.0/0")603rule2 = openstack::IPrule(ip_protocol="tcp", port=8888, direction="ingress", remote_prefix="0.0.0.0/0")604rule3 = openstack::IPrule(ip_protocol="tcp", port=22, direction="ingress", remote_prefix="0.0.0.0/0")605rule4 = openstack::IPrule(ip_protocol="tcp", port=5432, direction="ingress", remote_prefix="0.0.0.0/0")606security_group = openstack::SecurityGroup(607 name="{resource_name}",608 description="{description_sec_group2}",609 provider=p,610 project=project,611 rules=[rule1, rule2, rule3, rule4]612)613 """614 )615 project.deploy_resource("openstack::SecurityGroup")616 sgs = project1.neutron.list_security_groups(617 name=resource_name, project_id=project1.project_object.id618 )619 assert len(sgs["security_groups"]) == 1620 assert sgs["security_groups"][0]["project_id"] == project1.project_object.id621 assert sgs["security_groups"][0]["description"] == description_sec_group1622 ipv4_rules = [623 r624 for r in sgs["security_groups"][0]["security_group_rules"]625 if r["ethertype"] == "IPv4"626 ]627 assert len(ipv4_rules) == 1 # Egress is allowed by default628 sgs = project2.neutron.list_security_groups(629 name=resource_name, project_id=project2.project_object.id630 )631 assert len(sgs["security_groups"]) == 1632 assert sgs["security_groups"][0]["project_id"] == project2.project_object.id633 assert sgs["security_groups"][0]["description"] == description_sec_group2634 ipv4_rules = [635 r636 for r in sgs["security_groups"][0]["security_group_rules"]637 if r["ethertype"] == "IPv4"638 ]639 assert len(ipv4_rules) == 4640def test_shared_network(project, openstack):641 """642 Create a shared network as one tenant and add ports to it as another tenant643 """644 tenant1 = openstack.get_project("tenant1")645 tenant2 = openstack.get_project("tenant2")646 net_name = tenant1.get_resource_name("net")647 key_name = tenant2.get_resource_name("key")648 server_name = tenant2.get_resource_name("server").replace("_", "-")649 # create a shared network in tenant1650 project.compile(651 """652 import unittest653 import openstack654 tenant = "%(project)s"655 p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),656 password=std::get_env("OS_PASSWORD"), tenant=tenant)657 project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)658 n = openstack::Network(provider=p, name="%(name)s", project=project, shared=true)659 subnet = openstack::Subnet(provider=p, project=project, network=n, dhcp=true, name="%(name)s",660 network_address="10.255.255.0/24", dns_servers=["8.8.8.8", "8.8.4.4"])661 """662 % {"name": net_name, "project": tenant1._tenant}663 )664 project.deploy_resource("openstack::Network", name=net_name)665 project.deploy_resource("openstack::Subnet", name=net_name)666 # create a hostport on the shared network in tenant2667 project.compile(668 """669 import unittest670 import openstack671 import ssh672 tenant = "%(project)s"673 p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),674 password=std::get_env("OS_PASSWORD"), tenant=tenant)675 project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)676 n1 = openstack::Network(provider=p, name="%(net_name)s", project=project, managed=false)677 s1 = openstack::Subnet(provider=p, project=project, network=n1, dhcp=true, name="%(net_name)s",678 network_address="10.255.255.0/24", dns_servers=["8.8.8.8", "8.8.4.4"], managed=false)679 n2 = openstack::Network(provider=p, name="%(net_name)s2", project=project, managed=true)680 s2 = openstack::Subnet(provider=p, project=project, network=n2, dhcp=true, name="%(net_name)s2",681 network_address="10.255.254.0/24", dns_servers=["8.8.8.8", "8.8.4.4"], managed=true)682 os = std::OS(name="cirros", version=0.4, family=std::linux)683 key = ssh::Key(name="%(key_name)s", public_key="")684 vm = openstack::Host(provider=p, project=project, key_pair=key, name="%(server_name)s", os=os,685 image=openstack::find_image(p, os), flavor=openstack::find_flavor(p, 1, 0.5), user_data="",686 subnet=s2)687 port = openstack::HostPort(provider=p, vm=vm.vm, subnet=s1, name="%(server_name)s_eth1", address="10.255.255.123",688 project=project, port_index=2, purged=false, dhcp=false)689 """690 % {691 "net_name": net_name,692 "project": tenant2._tenant,693 "key_name": key_name,694 "server_name": server_name,695 }696 )697 project.deploy_resource("openstack::Network", name=net_name + "2")698 project.deploy_resource("openstack::Subnet", name=net_name + "2")699 project.deploy_resource("openstack::VirtualMachine")700 project.deploy_resource("openstack::HostPort")701 project.compile(702 """703 import unittest704 import openstack705 import ssh706 tenant = "%(project)s"707 p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),708 password=std::get_env("OS_PASSWORD"), tenant=tenant)709 project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)710 n1 = openstack::Network(provider=p, name="%(net_name)s", project=project, managed=false)711 s1 = openstack::Subnet(provider=p, project=project, network=n1, dhcp=true, name="%(net_name)s",712 network_address="10.255.255.0/24", dns_servers=["8.8.8.8", "8.8.4.4"], managed=false)713 n2 = openstack::Network(provider=p, name="%(net_name)s2", project=project, managed=true)714 s2 = openstack::Subnet(provider=p, project=project, network=n2, dhcp=true, name="%(net_name)s2",715 network_address="10.255.254.0/24", dns_servers=["8.8.8.8", "8.8.4.4"], managed=true)716 n3 = openstack::Network(provider=p, name="%(net_name)s3", project=project, managed=true)717 s3 = openstack::Subnet(provider=p, project=project, network=n3, dhcp=true, name="%(net_name)s3",718 network_address="10.255.253.0/24", dns_servers=["8.8.8.8", "8.8.4.4"], managed=true)719 os = std::OS(name="cirros", version=0.4, family=std::linux)720 key = ssh::Key(name="%(key_name)s", public_key="")721 vm = openstack::Host(provider=p, project=project, key_pair=key, name="%(server_name)s", os=os,722 image=openstack::find_image(p, os), flavor=openstack::find_flavor(p, 1, 0.5), user_data="",723 subnet=s2)724 openstack::HostPort(provider=p, vm=vm.vm, subnet=s1, name="%(server_name)s_eth1", address="10.255.255.123",725 project=project, port_index=2, purged=false, dhcp=false)726 openstack::HostPort(provider=p, vm=vm.vm, subnet=s3, name="%(server_name)s_eth2", address="10.255.253.12",727 project=project, port_index=3, purged=false, dhcp=false)728 """729 % {730 "net_name": net_name,731 "project": tenant2._tenant,732 "key_name": key_name,733 "server_name": server_name,734 }735 )736 project.deploy_resource("openstack::Network", name=net_name + "3")737 project.deploy_resource("openstack::Subnet", name=net_name + "3")738 project.deploy_resource("openstack::HostPort", name=server_name + "_eth2")739def test_allowed_addr_port(project, openstack):740 """741 Test creating a port with allowed address pairs742 """743 tenant1 = openstack.get_project("tenant1")744 net_name = tenant1.get_resource_name("net")745 port_name = tenant1.get_resource_name("port")746 key_name = tenant1.get_resource_name("key")747 server_name = tenant1.get_resource_name("server").replace("_", "-")748 project.compile(749 """750 import unittest751 import openstack752 import ssh753 tenant = "%(project)s"754 p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),755 password=std::get_env("OS_PASSWORD"), tenant=tenant)756 project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)757 n = openstack::Network(provider=p, name="%(name)s", project=project)758 subnet = openstack::Subnet(provider=p, project=project, network=n, dhcp=true, name="%(name)s",759 network_address="10.255.255.0/24", dns_servers=["8.8.8.8", "8.8.4.4"])760 os = std::OS(name="cirros", version=0.4, family=std::linux)761 key = ssh::Key(name="%(key_name)s", public_key="")762 vm = openstack::VirtualMachine(provider=p, project=project, key_pair=key, name="%(server_name)s",763 image=openstack::find_image(p, os), flavor=openstack::find_flavor(p, 1, 0.5), user_data="")764 p1 = openstack::AddressPair(address="10.255.255.0/24")765 p2 = openstack::AddressPair(address="10.255.0.0/24", mac="12:23:34:45:56:67")766 port = openstack::HostPort(provider=p, project=project, name="%(port_name)s", subnet=subnet, address="10.255.255.10",767 dhcp=false, allowed_address_pairs=[p1, p2], vm=vm)768 vm.eth0_port = port769 """770 % {771 "name": net_name,772 "project": tenant1._tenant,773 "port_name": port_name,774 "server_name": server_name,775 "key_name": key_name,776 }777 )778 project.deploy_resource("openstack::Network", name=net_name)779 project.deploy_resource("openstack::Subnet", name=net_name)780 project.deploy_resource("openstack::VirtualMachine", name=server_name)781 project.deploy_resource("openstack::HostPort", name=port_name)782 ports = tenant1.neutron.list_ports(name=port_name)["ports"]783 assert len(ports) == 1784 assert len(ports[0]["allowed_address_pairs"]) == 2785 # recheck the config786 project.deploy_resource("openstack::HostPort", name=port_name)787@pytest.mark.parametrize(788 "disable_gateway_ip,gateway_ip",789 [790 (False, None),791 (False, "10.255.255.27"),792 (True, None),793 ( # Should not be used in practice. disable_gateway_ip takes precedence.794 True,795 "10.255.255.111",796 ),797 ],798)799def test_gateway_ip(project, openstack, disable_gateway_ip, gateway_ip):800 """801 Test whether the gateway_ip and the disable_gateway_ip settings of a subnet work correctly.802 """803 tenant1 = openstack.get_project("tenant1")804 net_name = tenant1.get_resource_name("net")805 subnet_name = tenant1.get_resource_name("subnet")806 gateway_ip_model = f'"{gateway_ip}"' if gateway_ip is not None else "null"807 project.compile(808 f"""809 import unittest810 import openstack811 tenant = "{tenant1._tenant}"812 p = openstack::Provider(name="test", connection_url=std::get_env("OS_AUTH_URL"), username=std::get_env("OS_USERNAME"),813 password=std::get_env("OS_PASSWORD"), tenant=tenant)814 project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)815 n = openstack::Network(provider=p, name="{net_name}", project=project)816 subnet = openstack::Subnet(provider=p, project=project, network=n, dhcp=true, name="{subnet_name}",817 network_address="10.255.255.0/24", dns_servers=["8.8.8.8", "8.8.4.4"],818 gateway_ip={gateway_ip_model}, disable_gateway_ip={str(disable_gateway_ip).lower()})819 """820 )821 project.deploy_resource("openstack::Network", name=net_name)822 # Check initial state823 changes = project.dryrun_resource("openstack::Subnet", name=subnet_name)824 assert changes825 # Deploy subnet826 project.deploy_resource("openstack::Subnet", name=subnet_name)827 # Verify state on Openstack828 subnets = tenant1.neutron.list_subnets(name=subnet_name)["subnets"]829 assert len(subnets) == 1830 subnet = subnets[0]831 if disable_gateway_ip:832 assert subnet["gateway_ip"] is None833 elif gateway_ip is None:834 # The first IP of the subnet should be set835 assert subnet["gateway_ip"] == "10.255.255.1"836 else:837 assert subnet["gateway_ip"] == gateway_ip838 # Ensure convergence839 changes = project.dryrun_resource("openstack::Subnet", name=subnet_name)840 assert not changes841def test_issue_7_and_286(project, openstack):842 tenant1 = openstack.get_project("tenant1")843 net_name = tenant1.get_resource_name("net")844 port_name = tenant1.get_resource_name("port")845 key_name = tenant1.get_resource_name("key")846 server_name = tenant1.get_resource_name("server").replace("_", "-")847 def _get_model(purged: bool) -> str:848 return f"""849 import unittest850 import openstack851 import ssh852 tenant = "{tenant1._tenant}"853 p = openstack::Provider(854 name="test",855 connection_url=std::get_env("OS_AUTH_URL"),856 username=std::get_env("OS_USERNAME"),857 password=std::get_env("OS_PASSWORD"),858 tenant=tenant859 )860 project = openstack::Project(provider=p, name=tenant, description="", enabled=true, managed=false)861 n = openstack::Network(provider=p, name="{net_name}", project=project)862 subnet = openstack::Subnet(863 provider=p,864 project=project,865 network=n,866 dhcp=true,867 name="{net_name}",868 network_address="10.255.255.0/24",869 dns_servers=["8.8.8.8", "8.8.4.4"],870 purged={str(purged).lower()},871 )872 os = std::OS(name="cirros", version=0.4, family=std::linux)873 key = ssh::Key(name="{key_name}", public_key="")874 vm = openstack::VirtualMachine(875 provider=p,876 project=project,877 key_pair=key,878 name="{server_name}",879 image=openstack::find_image(p, os),880 flavor=openstack::find_flavor(p, 1, 0.5),881 user_data="",882 purged={str(purged).lower()},883 )884 port = openstack::HostPort(885 provider=p,886 project=project,887 name="{port_name}",888 subnet=subnet,889 address="10.255.255.10",890 dhcp=false,891 vm=vm,892 retries=1,893 wait=0,894 purged={str(purged).lower()},895 )896 vm.eth0_port = port897 """898 project.compile(_get_model(purged=False))899 project.deploy_resource("openstack::Network", name=net_name)900 project.deploy_resource("openstack::Subnet", name=net_name)901 # HostPort doesn't have VM so it will be skipped, but shouldn't fail.902 project.deploy_resource(903 "openstack::HostPort",904 name=port_name,905 status=inmanta.const.ResourceState.skipped,906 )907 project.compile(_get_model(purged=True))908 # HostPort doesn't exist, so it should succeed.909 project.deploy_resource(910 "openstack::HostPort",911 name=port_name,...
aws_deploy.py
Source:aws_deploy.py
1from create_boto_client import create_boto_client2from rds import create_postgresql_instance34def deploy_resource():5 """create RDS client and create postgresql instance"""67 # create and get rds client8 rds_client = create_boto_client('rds')910 # create postgresql instance11 create_postgresql_instance(rds_client)1213if __name__ == '__main__':
...
test_deployment.py
Source:test_deployment.py
...3def get_RDS():4 rds_client = RDSClient().get_client()5 rds = RDS(rds_client)6 return rds7def deploy_resource():8 get_RDS().create_postgres_instance()9 print('RDS instances created ')10if __name__ == '__main__':...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
