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.