Create integration tests for NodeBalancer Front-End IP & 40Gbps

Author: mawilk90

test/integration/conftest.py

@@ -457,7 +457,7 @@ def create_vpc(test_linode_client):
     vpc = client.vpcs.create(
         label=label,
         region=get_region(
-            test_linode_client, {"VPCs", "VPC IPv6 Stack", "Linode Interfaces"}
+            test_linode_client, {"VPCs", "VPC IPv6 Stack", "VPC Dual Stack", "Linode Interfaces", "NodeBalancers"}
         ),
         description="test description",
         ipv6=[{"range": "auto"}],
@@ -467,6 +467,24 @@ def create_vpc(test_linode_client):
     vpc.delete()
 
 
+@pytest.fixture(scope="session")
+def create_vpc_ipv4(test_linode_client):
+    client = test_linode_client
+
+    label = get_test_label(length=10) + "-ipv4"
+
+    vpc = client.vpcs.create(
+        label=label,
+        region=get_region(
+            test_linode_client, {"VPCs", "Linode Interfaces", "NodeBalancers"}
+        ),
+        description="test description"
+    )
+    yield vpc
+
+    vpc.delete()
+
+
 @pytest.fixture(scope="session")
 def create_vpc_with_subnet(test_linode_client, create_vpc):
     subnet = create_vpc.subnet_create(
@@ -480,6 +498,18 @@ def create_vpc_with_subnet(test_linode_client, create_vpc):
     subnet.delete()
 
 
+@pytest.fixture(scope="session")
+def create_vpc_with_subnet_ipv4(test_linode_client, create_vpc_ipv4):
+    subnet = create_vpc_ipv4.subnet_create(
+        label="test-subnet",
+        ipv4="10.0.0.0/24"
+    )
+
+    yield create_vpc_ipv4, subnet
+
+    subnet.delete()
+
+
 @pytest.fixture(scope="session")
 def create_vpc_with_subnet_and_linode(
     test_linode_client, create_vpc_with_subnet, e2e_test_firewall

test/integration/models/nodebalancer/test_nodebalancer.py

@@ -1,3 +1,4 @@
+import ipaddress
 import re
 from test.integration.conftest import (
     get_api_ca_file,
@@ -272,3 +273,194 @@ def test_nodebalancer_types(test_linode_client):
                     isinstance(region_price.monthly, (float, int))
                     and region_price.monthly >= 0
                 )
+
+
+def test_nb_with_backend_only(test_linode_client, create_vpc_with_subnet):
+    client = test_linode_client
+    vpc = create_vpc_with_subnet[0].id
+    vpc_region = create_vpc_with_subnet[0].region
+    subnet = create_vpc_with_subnet[1].id
+    label = get_test_label(8)
+
+    nb = client.nodebalancer_create(
+        region=vpc_region, label=label, vpcs=[{"vpc_id": vpc,"subnet_id": subnet}]
+    )
+
+    assert isinstance(ipaddress.ip_address(nb.ipv4.address), ipaddress.IPv4Address)
+    assert isinstance(ipaddress.ip_address(nb.ipv6), ipaddress.IPv6Address)
+    assert nb.frontend_address_type == 'public'
+    assert nb.frontend_vpc_subnet_id is None
+
+    nb_get = NodeBalancer(client, nb.id)
+    nb_vpcs = nb_get.vpcs()
+
+    assert len(nb_vpcs) == 1
+    assert nb_vpcs[0].purpose == 'backend'
+
+    nb_vpc = nb_get.vpc(nb_vpcs[0].id)
+
+    assert nb_vpc.purpose == 'backend'
+
+    # TODO: There is no API implementation yet for /backend_vpcs and /frontend_vpcs
+    # nb_backend_vpcs = nb_get.backend_vpcs()
+    # assert len(nb_backend_vpcs) == 1
+    # assert nb_backend_vpcs[0].purpose == 'backend'
+    #
+    # nb_frontend_vpcs = nb_get.frontend_vpcs()
+    # assert len(nb_frontend_vpcs) == 0
+
+    nb.delete()
+
+
+def test_nb_with_frontend_ipv4_only_in_single_stack_vpc(test_linode_client, create_vpc_with_subnet_ipv4):
+    client = test_linode_client
+    vpc_region = create_vpc_with_subnet_ipv4[0].region
+    subnet = create_vpc_with_subnet_ipv4[1].id
+    label = get_test_label(8)
+    ipv4_address = "10.0.0.2" # first available address
+
+    nb = client.nodebalancer_create(
+        region=vpc_region,
+        label=label,
+        frontend_vpcs=[{"subnet_id": subnet, "ipv4_range": "10.0.0.0/24"}],
+        type="premium"
+    )
+    assert nb.ipv4.address == ipv4_address
+    assert nb.ipv6 is None
+    assert nb.frontend_address_type == 'vpc'
+    assert nb.frontend_vpc_subnet_id == subnet
+
+    # TODO: There is no API implementation yet for /backend_vpcs and /frontend_vpcs
+    # nb_frontend_vpcs = nb_get.frontend_vpcs()
+    # assert len(nb_frontend_vpcs) == 1
+    # assert nb_frontend_vpcs[0].purpose == 'frontend'
+    #
+    # nb_backend_vpcs = nb_get.backend_vpcs()
+    # assert len(nb_backend_vpcs) == 0
+
+    nb.delete()
+
+
+def test_nb_with_frontend_ipv6_only_in_single_stack_vpc(test_linode_client, create_vpc_with_subnet_ipv4):
+    client = test_linode_client
+    vpc_region = create_vpc_with_subnet_ipv4[0].region
+    subnet = create_vpc_with_subnet_ipv4[1].id
+    label = get_test_label(8)
+
+    with pytest.raises(ApiError) as err:
+        client.nodebalancer_create(
+            region=vpc_region,
+            label=label,
+            frontend_vpcs=[{"subnet_id": subnet, "ipv6_range": "/62"}],
+            type="premium"
+        )
+        assert "No IPv6 subnets available in VPC" in str(err.json)
+
+
+def test_nb_with_frontend_and_default_type(test_linode_client, create_vpc_with_subnet):
+    client = test_linode_client
+    vpc_region = create_vpc_with_subnet[0].region
+    subnet = create_vpc_with_subnet[1].id
+    label = get_test_label(8)
+
+    with pytest.raises(ApiError) as err:
+        client.nodebalancer_create(
+            region=vpc_region,
+            label=label,
+            frontend_vpcs=[{"subnet_id": subnet}],
+        )
+        assert "Nodebalancer with frontend VPC IP must be premium" in str(err.json)
+
+
+def test_nb_with_frontend_and_premium40gb_type(test_linode_client):
+    region = "us-iad"  # premium_40gb type can be used in specific regions only
+    client = test_linode_client
+
+    vpc = client.vpcs.create(
+        label=get_test_label(length=10),
+        region=region,
+        description="test description",
+        ipv6=[{"range": "auto"}],
+    )
+
+    subnet = vpc.subnet_create(
+        label="test-subnet",
+        ipv4="10.0.0.0/24",
+        ipv6=[{"range": "auto"}],
+    )
+
+    nb = client.nodebalancer_create(
+        region=region,
+        label=get_test_label(length=8),
+        frontend_vpcs=[{"subnet_id": subnet.id}],
+        type="premium_40gb"
+    )
+    assert nb.type == 'premium_40gb'
+
+    nb_get = test_linode_client.load(
+        NodeBalancer,
+        nb.id,
+    )
+    assert nb_get.type == 'premium_40gb'
+
+    nb.delete()
+    vpc.delete()
+
+
+def test_nb_with_frontend_and_backend_in_different_vpcs(test_linode_client, create_vpc_with_subnet):
+    client = test_linode_client
+    region = create_vpc_with_subnet[0].region
+    vpc_backend = create_vpc_with_subnet[0].id
+    subnet_backend = create_vpc_with_subnet[1].id
+    label = get_test_label(8)
+    ipv4_range = "10.0.0.0/24"
+    ipv4_address = "10.0.0.2" # first available address
+
+    vpc_frontend = client.vpcs.create(
+        label=get_test_label(length=10),
+        region=region,
+        description="test description",
+        ipv6=[{"range": "auto"}],
+    )
+
+    subnet_frontend = vpc_frontend.subnet_create(
+        label="test-subnet",
+        ipv4=ipv4_range,
+        ipv6=[{"range": "auto"}],
+    )
+    ipv6_range = subnet_frontend.ipv6[0].range
+    ipv6_address = ipv6_range.split("::")[0] + ":1::1"
+
+    nb = client.nodebalancer_create(
+        region=region,
+        label=label,
+        vpcs=[{"vpc_id": vpc_backend,"subnet_id": subnet_backend}],
+        frontend_vpcs=[{"subnet_id": subnet_frontend.id, "ipv4_range": ipv4_range, "ipv6_range": ipv6_range}],
+        type="premium"
+    )
+
+    assert nb.ipv4.address == ipv4_address
+    assert nb.ipv6 == ipv6_address
+    assert nb.frontend_address_type == 'vpc'
+    assert nb.frontend_vpc_subnet_id == subnet_frontend.id
+
+    nb_get = NodeBalancer(client, nb.id)
+    nb_vpcs = nb_get.vpcs()
+
+    assert len(nb_vpcs) == 2
+    assert nb_vpcs[0].ipv4_range == f"{ipv4_address}/32"
+    assert nb_vpcs[0].ipv6_range == f"{ipv6_address[:-1]}/64"
+    assert nb_vpcs[0].purpose == 'frontend'
+    assert nb_vpcs[1].purpose == 'backend'
+
+    # TODO: There is no API implementation yet for /backend_vpcs and /frontend_vpcs
+    # nb_backend_vpcs = nb_get.backend_vpcs()
+    # assert len(nb_backend_vpcs) == 1
+    # assert nb_backend_vpcs[0].purpose == 'backend'
+    #
+    # nb_frontend_vpcs = nb_get.frontend_vpcs()
+    # assert len(nb_frontend_vpcs) == 1
+    # assert nb_frontend_vpcs[0].purpose == 'frontend'
+
+    nb.delete()
+    vpc_frontend.delete()