import ipaddress import re from os import PathLike import pytest import responses from requests import codes from gso.services import infoblox from gso.services.infoblox import AllocationError, DeletionError def _set_up_network_responses(): responses.add(method=responses.GET, url=re.compile(r".+/wapi/v2\.12/network\?network=10\.255\.255\.0.+"), json=[]) responses.add(method=responses.GET, url=re.compile(r".+/wapi/v2\.12/ipv6network\?network=dead%3Abeef.+"), json=[]) responses.add( method=responses.POST, url=re.compile(r".+/wapi/v2\.12/network.+"), json={ "_ref": "network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.20/32/default", "network": "10.255.255.20/32", }, status=codes.CREATED, ) responses.add( method=responses.POST, url=re.compile(r".+/wapi/v2\.12/ipv6network.+"), json={ "_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default", "network": "dead:beef::18/128", }, status=codes.CREATED, ) def _set_up_host_responses(): responses.add( method=responses.GET, url="https://10.0.0.1/wapi/v2.12/record%3Ahost?_return_fields=extattrs%2Cipv6addrs%2Cname%2Cview%2Caliases", json=[], ) responses.add( method=responses.GET, url="https://10.0.0.1/wapi/v2.12/record%3Ahost?name=test.lo.geant.net&ipv6addr=func%3Anextavailableip%3Adead%3A" "beef%3A%3A%2F80%2Cdefault", json=[], ) responses.add( method=responses.GET, url="https://10.0.0.1/wapi/v2.12/record%3Ahost?name=test.lo.geant.net&_return_fields=extattrs%2Cipv4addrs%2Cnam" "e%2Cview%2Caliases", json=[], ) responses.add( method=responses.GET, url="https://10.0.0.1/wapi/v2.12/record%3Ahost?name=broken&_return_fields=extattrs%2Cipv4addrs%2Cname%2Cview%2C" "aliases", json=[], ) responses.add( method=responses.GET, url=re.compile( r"https://10.0.0.1/wapi/v2.12/record%3Ahost\?name=broken&ipv6addr=func%3Anextavailableip%3Adead%3Abeef%3A%3" r"A%2F80%2Cdefault.*" ), json=[], status=codes.BAD, ) responses.add( method=responses.POST, url=re.compile(r".+/wapi/v2\.12/record%3Ahost\?_return_fields=extattrs%2Cipv6addrs.+"), json={ "_ref": "record:host/ZG5zLmhvc3QkLl9kZWZhdWx0Lm5ldC5nZWFudC5sby50ZXN0:test.lo.geant.net/default", "ipv6addrs": [ { "_ref": "record:host_ipv6addr/ZG5zLmhvc3RfYWRkcmVzcyQuX2RlZmF1bHQubmV0LmdlYW50LmxvLnRlc3QuZGVhZDpiZ" "WVmOjoxLg:dead%3Abeef%3A%3A1/test.lo.geant.net/default", "configure_for_dhcp": False, "duid": "00:00:00:00:00:00:00:00:00:00", "host": "test.lo.geant.net", "ipv6addr": "dead:beef::1", } ], "ip": "dead:beef::1", "name": "test.lo.geant.net", "view": "default", }, status=codes.CREATED, ) responses.add( method=responses.PUT, url="https://10.0.0.1/wapi/v2.12/record%3Ahost/ZG5zLmhvc3QkLl9kZWZhdWx0Lm5ldC5nZWFudC5sd28udGVzdA%3Atest.lo.gea" "nt.net/default?_return_fields=extattrs%2Cipv4addrs%2Cname%2Cview%2Caliases", json={ "ipv4addrs": [ { "_ref": "record:host_ipv4addr/ZG5zLmhvc3RfYWRkcmVzcyQuX2RlZmF1bHQubmV0LmdlYW50Lmx3by50ZXN0LjEwLjI1N" "S4yNTUuMTI5Lg:10.255.255.129/test.lo.geant.net/default", "configure_for_dhcp": False, "host": "test.lo.geant.net", "ipv4addr": "10.255.255.129", "mac": "00:00:00:00:00:00", } ], "name": "test.lo.geant.net", "view": "default", }, ) responses.add( method=responses.GET, url="https://10.0.0.1/wapi/v2.12/record%3Ahost?name=test.lo.geant.net&_return_fields=extattrs%2Cipv4addrs%2Cnam" "e%2Cview%2Caliases", json=[ { "_ref": "record:host/ZG5zLmhvc3QkLl9kZWZhdWx0Lm5ldC5nZWFudC5sd28udGVzdA:test.lo.geant.net/default", "ipv4addrs": [ { "_ref": "record:host_ipv4addr/ZG5zLmhvc3RfYWRkcmVzcyQuX2RlZmF1bHQubmV0LmdlYW50Lmx3by50ZXN0LjEwL" "jI1N" "S4yNTUuMTI5Lg:10.255.255.129/test.lo.geant.net/default", "configure_for_dhcp": False, "host": "test.lo.geant.net", "ipv4addr": "10.255.255.129", "mac": "00:00:00:00:00:00", } ], "name": "test.lo.geant.net", "view": "default", } ], ) @responses.activate def test_allocate_networks(data_config_filename: PathLike): _set_up_network_responses() new_v4_network = infoblox.allocate_v4_network("TRUNK") new_v6_network = infoblox.allocate_v6_network("TRUNK") assert new_v4_network == ipaddress.IPv4Network("10.255.255.20/32") assert new_v6_network == ipaddress.IPv6Network("dead:beef::18/128") @responses.activate def test_allocate_bad_network(data_config_filename: PathLike): _set_up_network_responses() with pytest.raises(AllocationError) as e: infoblox.allocate_v4_network("LO") assert e.value.args[0] == "Cannot allocate anything in [], check whether any IP space is available." with pytest.raises(AllocationError) as e: infoblox.allocate_v6_network("LO") assert e.value.args[0] == "Cannot allocate anything in [], check whether any IP space is available." @responses.activate def test_allocate_good_host(data_config_filename: PathLike): _set_up_host_responses() new_host = infoblox.allocate_host("test.lo.geant.net", "LO", [], "test host") assert new_host == (ipaddress.ip_address("10.255.255.129"), ipaddress.ip_address("dead:beef::1")) @responses.activate def test_allocate_bad_host(data_config_filename: PathLike): _set_up_host_responses() with pytest.raises(AllocationError) as e: infoblox.allocate_host("broken", "TRUNK", [], "Unavailable host") assert e.value.args[0] == "Cannot find 1 available IP address in networks []." @responses.activate def test_delete_good_network(data_config_filename: PathLike): responses.add( method=responses.GET, url="https://10.0.0.1/wapi/v2.12/network?network=10.255.255.0%2F26&_return_fields=comment%2Cextattrs%2Cnetwork%" "2Cnetwork_view", json=[ { "_ref": "network/ZG5zLm5ldHdvcmskNjIuNDAuOTYuMC8yNC8w:10.255.255.0/26/default", "network": "10.255.255.0/26", "network_view": "default", } ], ) responses.add( method=responses.DELETE, url="https://10.0.0.1/wapi/v2.12/network/ZG5zLm5ldHdvcmskNjIuNDAuOTYuMC8yNC8w%3A10.255.255.0/26/default", json=[], ) infoblox.delete_network(ipaddress.IPv4Network("10.255.255.0/26")) @responses.activate def test_delete_non_existent_network(data_config_filename: PathLike): responses.add( method=responses.GET, url="https://10.0.0.1/wapi/v2.12/network?network=10.255.255.0%2F26&_return_fields=comment%2Cextattrs%2Cnetwork%" "2Cnetwork_view", json=[], ) with pytest.raises(DeletionError) as e: infoblox.delete_network(ipaddress.IPv4Network("10.255.255.0/26")) assert e.value.args[0] == "Could not find network 10.255.255.0/26, nothing has been deleted." @responses.activate def test_delete_good_host(data_config_filename: PathLike): responses.add( method=responses.GET, url=re.compile( r"https://10\.0\.0\.1/wapi/v2\.12/record%3Ahost\?(?:name=ha_lo\.gso|ipv4addr=10\.255\.255\.1)?.+" ), json=[ { "_ref": "record:host/ZG5zLmhvc3QkLl9kZWZhdWx0Lmdzby5oYV9sbw:ha_lo.gso/default", "ipv4addrs": [ { "_ref": "record:host_ipv4addr/ZG5zLmhvc3RfYWRkcmVzcyQuX2RlZmF1bHQuZ3NvLmhhX2xvLjEwLjI1NS4yNTUuM" "S40.255.255.1/ha_lo.gso/default", "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", "configure_for_dhcp": False, "host": "ha_lo.gso", "ipv6addr": "dead:beef::1", } ], "name": "ha_lo.gso", "view": "default", } ], ) responses.add( method=responses.DELETE, url=re.compile( r"https://10\.0\.0\.1/wapi/v2\.12/record%3Ahost/.+(ha_lo\.gso|dead:beef::1|10\.255\.255\.1)/default" ), json=[], ) infoblox.delete_host_by_fqdn("ha_lo.gso") infoblox.delete_host_by_ip(ipaddress.IPv4Address("10.255.255.1")) infoblox.delete_host_by_ip(ipaddress.IPv6Address("dead:beef::1")) @responses.activate def test_delete_bad_host(data_config_filename: PathLike): responses.add( method=responses.GET, url=re.compile(r".+"), json=[], ) with pytest.raises(DeletionError) as e: infoblox.delete_host_by_ip(ipaddress.IPv4Address("10.255.255.1")) assert e.value.args[0] == "Could not find host at 10.255.255.1, nothing has been deleted." with pytest.raises(DeletionError) as e: infoblox.delete_host_by_ip(ipaddress.IPv6Address("dead:beef::1")) assert e.value.args[0] == "Could not find host at dead:beef::1, nothing has been deleted." with pytest.raises(DeletionError) as e: infoblox.delete_host_by_fqdn("fake.host.net") assert e.value.args[0] == "Could not find host at fake.host.net, nothing has been deleted."