Skip to content
Snippets Groups Projects
ipam.py 16.8 KiB
Newer Older
Erik Reid's avatar
Erik Reid committed
import ipaddress
import re
from os import PathLike
import pytest
import responses
from gso.services import ipam
Erik Reid's avatar
Erik Reid committed

@responses.activate
def test_allocate_networks(data_config_filename: PathLike):
    responses.add(
        method=responses.POST,
        url=re.compile(r".*/wapi.*/network.*"),
        json={
            "_ref": "network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.20/32/default",  # noqa: E501
            "network": "10.255.255.20/32",
        },
    )
    responses.add(
        method=responses.POST,
        url=re.compile(r".*/wapi.*/ipv6network.*"),
        json={
            "_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default",  # noqa: E501
            "network": "dead:beef::18/128",
        },
    )
    service_networks = ipam.allocate_networks(service_type="TRUNK")
    assert service_networks == ipam.ServiceNetworks(
        v4=ipaddress.ip_network("10.255.255.20/32"), v6=ipaddress.ip_network("dead:beef::18/128")
    )
    # should fail because this service type has networks instead of containers
    with pytest.raises(AssertionError):
        service_networks = ipam.allocate_networks(service_type="LO")
        assert service_networks is None
@responses.activate
def test_allocate_host(data_config_filename: PathLike):
    responses.add(
        method=responses.POST,
        url=re.compile(r".*/wapi.*/record:host$"),
        json="record:host/ZG5zLmhvc3QkLm5vbl9ETlNfaG9zdF9yb290LjAuMTY4MzcwNTU4MzY3MC5nc28udGVzdA:test.lo/%20",  # noqa: E501
    )
    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*/network.*10.255.255.*"),
        json=[
            {
                "_ref": "network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.20/32/default",  # noqa: E501
                "network": "10.255.255.20/32",
                "network_view": "default",
            }
        ],
    )
    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*/network.*10.255.254.*"),
        json=[
            {
                "_ref": "network/ZG5zLm5Gd0VHQkRQUjMzLjMwNzIuMzE1LzAyLzI:10.255.254.20/32/default",  # noqa: E501
                "network": "10.255.254.20/32",
                "network_view": "default",
            }
        ],
    )
    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*/ipv6network.*dead.*beef.*"),
        json=[
            {
                "_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default",  # noqa: E501
                "network": "dead:beef::18/128",
                "network_view": "default",
            }
        ],
    )
    responses.add(method=responses.GET, url=re.compile(r".*/wapi.*/ipv6network.*beef.*dead.*"), json=[])
    responses.add(
        method=responses.POST,
        url=re.compile(r".*/wapi.*/network/.*10.255.255.*?_function=next_available_ip&num=1$"),  # noqa: E501
        json={"ips": ["10.255.255.20"]},
    )
    responses.add(
        method=responses.POST,
        url=re.compile(r".*/wapi.*/network/.*10.255.254.*?_function=next_available_ip&num=1$"),  # noqa: E501
        body="Cannot find 1 available IP address(es) in this network",
        status=400,
    )
    responses.add(
        method=responses.POST,
        url=re.compile(r".*/wapi.*/ipv6network/.*?_function=next_available_ip&num=1$"),  # noqa: E501
        json={"ips": ["dead:beef::18"]},
    )
    responses.add(
        method=responses.POST,
        url=re.compile(r".*/wapi.*/network.*_return_fields.*"),
        json={
            "_ref": "network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.20/32/default",  # noqa: E501
            "network": "10.255.255.20/32",
        },
    )
    responses.add(
        method=responses.POST,
        url=re.compile(r".*/wapi.*/ipv6network.*_return_fields.*"),
        json={
            "_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default",  # noqa: E501
            "network": "dead:beef::18/128",
        },
    )
    # test host creation by IP addresses
    service_hosts = ipam.allocate_host(
        hostname="test",
        service_type="TRUNK",
        host_addresses=ipam.HostAddresses(
            v4=ipaddress.ip_address("10.255.255.20"), v6=ipaddress.ip_address("dead:beef::18")
        ),
    )
    assert service_hosts == ipam.HostAddresses(
        v4=ipaddress.ip_address("10.255.255.20"), v6=ipaddress.ip_address("dead:beef::18")
    )
    # test host creation by network addresses
    service_hosts = ipam.allocate_host(
        hostname="test",
        service_type="TRUNK",
        service_networks=ipam.ServiceNetworks(
            v4=ipaddress.ip_network("10.255.255.20/32"), v6=ipaddress.ip_network("dead:beef::18/128")
        ),
    )
    assert service_hosts == ipam.HostAddresses(
        v4=ipaddress.ip_address("10.255.255.20"), v6=ipaddress.ip_address("dead:beef::18")
    )
    # test host creation by just service_type when service cfg uses networks
    service_hosts = ipam.allocate_host(hostname="test", service_type="LO")
    assert service_hosts == ipam.HostAddresses(
        v4=ipaddress.ip_address("10.255.255.20"), v6=ipaddress.ip_address("dead:beef::18")
    )
    # test host creation by just service_type when service cfg uses containers
    service_hosts = ipam.allocate_host(hostname="test", service_type="TRUNK")
    assert service_hosts == ipam.HostAddresses(
        v4=ipaddress.ip_address("10.255.255.20"), v6=ipaddress.ip_address("dead:beef::18")
    )
    # test host creation that should return a no available IP error
    with pytest.raises(AssertionError):
        service_hosts = ipam.allocate_host(
            hostname="test",
            service_type="TRUNK",
            service_networks=ipam.ServiceNetworks(
                v4=ipaddress.ip_network("10.255.254.20/32"), v6=ipaddress.ip_network("dead:beef::18/128")
            ),
        assert service_hosts is None

    # test host creation that should return a network not exist error
    with pytest.raises(AssertionError):
        service_hosts = ipam.allocate_host(
            hostname="test",
            service_type="TRUNK",
            service_networks=ipam.ServiceNetworks(
                v4=ipaddress.ip_network("10.255.255.20/32"), v6=ipaddress.ip_network("beef:dead::18/128")
            ),
        assert service_hosts is None


@responses.activate
def test_delete_network(data_config_filename: PathLike):
    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*/network.*10.255.255.0.*"),
        json=[
            {
                "_ref": "network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.0/26/default",  # noqa: E501
                "network": "10.255.255.0/26",
                "network_view": "default",
            }
        ],
    )

    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*/network.*10.255.255.20.*"),
        json=[
            {
                "_ref": "network/ZG5zLm5Gd0VHQkRQUjMzLjMwNzIuMzE1LzAyLzI:100.255.255.20/32/default",  # noqa: E501
                "network": "100.255.255.20/32",
                "network_view": "default",
            }
        ],
    )

    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*/ipv6network.*dead.*beef.*"),
        json=[
            {
                "_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default",  # noqa: E501
                "network": "dead:beef::18/128",
                "network_view": "default",
            }
        ],
    )

    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*/ipv6network.*beef.*dead.*"),
        json=[
            {
                "_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:beef%3Adead%3A%3A18/128/default",  # noqa: E501
                "network": "beef:dead::18/128",
                "network_view": "default",
            }
        ],
    )

    responses.add(
        method=responses.DELETE,
        url=re.compile(r".*/wapi.*/network.*10.255.255.0.*"),
        body="network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.0/26/default",  # noqa: E501
    )

    responses.add(
        method=responses.DELETE,
        url=re.compile(r".*/wapi.*/network.*100.255.255.*"),
        body="network/ZG5zLm5Gd0VHQkRQUjMzLjMwNzIuMzE1LzAyLzI:100.255.255.20/32/default",  # noqa: E501
    )

    responses.add(
        method=responses.DELETE,
        url=re.compile(r".*/wapi.*/ipv6network.*dead.*beef.*"),
        body="ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default",  # noqa: E501
    )

    responses.add(
        method=responses.DELETE,
        url=re.compile(r".*/wapi.*/ipv6network.*beef.*dead.*"),
        body="ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:beef%3Adead%3A%3A18/128/default",  # noqa: E501
    )

    service_network = ipam.delete_network(network=ipaddress.ip_network("10.255.255.0/26"), service_type="LO")
    assert service_network == ipam.V4ServiceNetwork(v4=ipaddress.ip_network("10.255.255.0/26"))

    with pytest.raises(AssertionError):
        service_network = ipam.delete_network(network=ipaddress.ip_network("10.255.255.20/32"), service_type="LO")
        assert service_network is None

    service_network = ipam.delete_network(network=ipaddress.ip_network("dead:beef::18/128"), service_type="TRUNK")
    assert service_network == ipam.V6ServiceNetwork(v6=ipaddress.ip_network("dead:beef::18/128"))

    with pytest.raises(AssertionError):
        service_network = ipam.delete_network(network=ipaddress.ip_network("beef:dead::18/128"), service_type="TRUNK")
        assert service_network is None


@responses.activate
def test_delete_host(data_config_filename: PathLike):
    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*record:host.*"),
        json=[
            {
                "_ref": "record:host/ZG5zLmhvc3QkLl9kZWZhdWx0Lmdzby5oYV9sbw:ha_lo.gso/default",  # noqa: E501
                "ipv4addrs": [
                    {
                        "_ref": "record:host_ipv4addr/ZG5zLmhvc3RfYWRkcmVzcyQuX2RlZmF1bHQuZ3NvLmhhX2xvLjEwLjI1NS4yNTUuMS40.255.255.1/ha_lo.gso/default",  # noqa: E501
                        "configure_for_dhcp": False,
                        "host": "ha_lo.gso",
                        "ipv4addr": "10.255.255.1",
                    }
                ],
                "ipv6addrs": [
                    {
                        "_ref": "record:host_ipv6addr/ZG5zLmhvc3RfYWRkcmVzcyQuX2RlZmF1bHQuZvLmhhX2xvLmRlYWQ6YmVlZjo6MS4:dead%3Abeef%3A%3A1/ha_lo.gso/default",  # noqa: E501
                        "configure_for_dhcp": False,
                        "host": "ha_lo.gso",
                        "ipv6addr": "dead:beef::1",
                    }
                ],
                "name": "ha_lo.gso",
                "view": "default",
            }
        ],
    )

    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*record:cname.*"),
        json=[
            {
                "_ref": "record:cname/ZG5zLmJpbmRfY25hbWUkLl9kZWZhdWx0Lmdzby5oYS5hbGlhczE:alias1.ha.gso/default",  # noqa: E501
                "canonical": "hA_LO.lo",
                "name": "alias1.ha.lo",
                "view": "default",
            },
            {
                "_ref": "record:cname/5zLmJpbmRfY25hbWUkLl9kZWZhdWx0Lmdzby5oYS5hbGlhczI:alias2.ha.gso/default",  # noqa: E501
                "canonical": "hA_LO.lo",
                "name": "alias2.ha.lo",
                "view": "default",
            },
        ],
    )

    responses.add(
        method=responses.DELETE,
        url=re.compile(r".*/wapi.*record:host.*"),
        body="record:host/ZG5zLmhvc3QkLl9kZWZhdWx0Lmdzby5oYl9sbw:hb_lo.gso/default",  # noqa: E501
    )

    responses.add(
        method=responses.DELETE,
        url=re.compile(r".*/wapi.*record:cname.*"),
        body="record:cname/ZG5zLmJpbmRfY25hbWUkLl9kZWZhdWx0Lmdzby5oYi5hbGlhczE:alias1.hb.gso/default",  # noqa: E501
    )

    input_host_addresses = ipam.HostAddresses(
        v4=ipaddress.ip_address("10.255.255.1"), v6=ipaddress.ip_address("dead:beef::1")
    )
    host_addresses = ipam.delete_host(
        hostname="ha_lo",
        host_addresses=input_host_addresses,
        cname_aliases=["alias1.ha", "alias2.ha"],
        service_type="LO",
    )
    assert host_addresses == ipam.HostAddresses(
        v4=ipaddress.ip_address("10.255.255.1"), v6=ipaddress.ip_address("dead:beef::1")
    )
    # Fail because missing CNAME
    with pytest.raises(AssertionError):
        host_addresses = ipam.delete_host(
            hostname="ha_lo", host_addresses=input_host_addresses, cname_aliases=["alias1.ha"], service_type="LO"
        assert host_addresses is None

    # Fail because non-matching CNAME
    with pytest.raises(AssertionError):
        host_addresses = ipam.delete_host(
            hostname="ha_lo",
            host_addresses=input_host_addresses,
            cname_aliases=["alias1.ha", "alias2.ha", "alias3.ha"],
            service_type="LO",
        assert host_addresses is None


@responses.activate
def test_validate_network(data_config_filename: PathLike):
    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*/network.*10.255.255.0.*"),
        json=[
            {
                "_ref": "network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.0/26/default",  # noqa: E501
                "network": "10.255.255.0/26",
                "network_view": "default",
                "comment": "the subscription id is 0123456789abcdef",
            }
        ],
    )
    service_network = ipam.validate_network(
        gso_subscription_id="0123456789abcdef", network=ipam.ipaddress.ip_network("10.255.255.0/26")
    )
    assert service_network == ipam.V4ServiceNetwork(v4=ipaddress.ip_network("10.255.255.0/26"))

    # Fail because non-matching subscription id
    with pytest.raises(AssertionError):
        service_network = ipam.validate_network(
            gso_subscription_id="1a2b3c4d5e6f7890", network=ipam.ipaddress.ip_network("10.255.255.0/26")
        assert service_network is None


@responses.activate
def test_validate_host(data_config_filename: PathLike):
    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*record:host.*"),
        json=[
            {
                "_ref": "record:host/ZG5zLmhvc3QkLl9kZWZhdWx0Lmdzby5oYV9sbw:ha_lo.gso/default",  # noqa: E501
                "ipv4addrs": [
                    {
                        "_ref": "record:host_ipv4addr/ZG5zLmhvc3RfYWRkcmVzcyQuX2RlZmF1bHQuZ3NvLmhhX2xvLjEwLjI1NS4yNTUuMS40.255.255.1/ha_lo.gso/default",  # noqa: E501
                        "configure_for_dhcp": False,
                        "host": "ha_lo.gso",
                        "ipv4addr": "10.255.255.1",
                    }
                ],
                "ipv6addrs": [
                    {
                        "_ref": "record:host_ipv6addr/ZG5zLmhvc3RfYWRkcmVzcyQuX2RlZmF1bHQuZvLmhhX2xvLmRlYWQ6YmVlZjo6MS4:dead%3Abeef%3A%3A1/ha_lo.gso/default",  # noqa: E501
                        "configure_for_dhcp": False,
                        "host": "ha_lo.gso",
                        "ipv6addr": "dead:beef::1",
                    }
                ],
                "name": "ha_lo.gso",
                "view": "default",
            }
        ],
    responses.add(
        method=responses.GET,
        url=re.compile(r".*/wapi.*record:cname.*"),
        json=[
            {
                "_ref": "record:cname/ZG5zLmJpbmRfY25hbWUkLl9kZWZhdWx0Lmdzby5oYS5hbGlhczE:alias1.ha.gso/default",  # noqa: E501
                "canonical": "hA_LO.lo",
                "name": "alias1.ha.lo",
                "view": "default",
            },
            {
                "_ref": "record:cname/5zLmJpbmRfY25hbWUkLl9kZWZhdWx0Lmdzby5oYS5hbGlhczI:alias2.ha.gso/default",  # noqa: E501
                "canonical": "hA_LO.lo",
                "name": "alias2.ha.lo",
                "view": "default",
            },
        ],
    )

    input_host_addresses = ipam.HostAddresses(
        v4=ipaddress.ip_address("10.255.255.1"), v6=ipaddress.ip_address("dead:beef::1")
    )
    host_addresses = ipam.validate_host(
        hostname="ha_lo",
        host_addresses=input_host_addresses,
        cname_aliases=["alias1.ha", "alias2.ha"],
        service_type="LO",
    )
    assert host_addresses == ipam.HostAddresses(
        v4=ipaddress.ip_address("10.255.255.1"), v6=ipaddress.ip_address("dead:beef::1")
    )

    # Fail because non-matching hostname
    host_addresses = ipam.validate_host(
        hostname="wrong_hostname",
        host_addresses=input_host_addresses,
        cname_aliases=["alias1.ha", "alias2.ha"],
        service_type="LO",
    )
    with pytest.raises(AssertionError):
        host_addresses = ipam.HostAddresses(
            v4=ipaddress.ip_address("10.255.255.1"), v6=ipaddress.ip_address("dead:beef::1")
        )
        assert host_addresses is None