Merge pull request #59 from duplocloud/integration-tests

Integration tests

src/duplo_resource/infrastructure.py

@@ -1,5 +1,5 @@
 from duplocloud.client import DuploClient
-from duplocloud.errors import DuploError
+from duplocloud.errors import DuploError, DuploFailedResource
 from duplocloud.resource import DuploResource
 from duplocloud.commander import Command, Resource
 import duplocloud.args as args
@@ -38,6 +38,9 @@ def create(self,
     def wait_check():
       i = self.find(body["Name"])
       if i["ProvisioningStatus"] != "Complete":
+        # stop waiting if the status contains failed
+        if "Failed" in i["ProvisioningStatus"]:
+          raise DuploFailedResource(f"Infrastructure '{body['Name']}'")
         raise DuploError(f"Infrastructure '{body['Name']}' not ready", 404)
     self.duplo.post("adminproxy/CreateInfrastructureConfig", body)
     if wait:

src/duplocloud/errors.py

@@ -14,3 +14,8 @@ class DuploExpiredCache(DuploError):
   def __init__(self, key: str):
     self.key = key
     super().__init__("Cache item {key} is expired", 404)
+
+class DuploFailedResource(DuploError):
+  """Raised when a Duplo resource is in a failed state."""
+  def __init__(self, name: str):
+    super().__init__(f"{name} is in a failed state", 412)

src/duplocloud/resource.py

@@ -1,5 +1,5 @@
 from .client import DuploClient
-from .errors import DuploError
+from .errors import DuploError, DuploFailedResource
 from .commander import get_parser
 import math
 import time
@@ -34,6 +34,8 @@ def wait(self, wait_check: callable, timeout: int=None, poll: int=10):
       try:
         wait_check()
         break
+      except DuploFailedResource as e:
+        raise e
       except DuploError:
         time.sleep(poll)
     else:

src/tests/test_infra.py

@@ -1,60 +1,75 @@
 import math
 import pytest
-import time
 import random
 
-from duplocloud.errors import DuploError
+from duplocloud.errors import DuploError, DuploFailedResource
 from duplocloud.client import DuploClient
 
 duplo, _ = DuploClient.from_env()
 
-@pytest.mark.integration
-def test_listing_infrastructures():
-  r = duplo.load("infrastructure")
-  try:
-    lot = r("list")
-  except DuploError as e:
-    pytest.fail(f"Failed to list infrastructures: {e}")
+@pytest.fixture(scope="module")
+def infra_name():
+  inc = random.randint(1, 100)
+  return f"duploctl{inc}"
 
-@pytest.mark.integration
-def test_finding_infra():
-  r = duplo.load("tenant")
-  try:
-    t = r("find", "default")
-  except DuploError as e:
-    pytest.fail(f"Failed to find default infra: {e}")
-  assert t["AccountName"] == "default"
+class TestInfra:
 
-@pytest.mark.integration
-def test_creating_infrastructures():
-  r = duplo.load("infrastructure")
-  inc = random.randint(1, 100)
-  vnum = math.ceil(random.randint(1, 9))
-  name = f"duploctl{inc}"
-  try:
-    r.create({
-      "Name": name,
-      "Accountid": "",
-      "EnableK8Cluster": False,
-      "AzCount": 2,
-      "Vnet": { 
-        "SubnetCidr": 22, 
-        "AddressPrefix": f"11.2{vnum}0.0.0/16"
-      },
-      "Cloud": 0,
-      "OnPremConfig": None,
-      "Region": "us-west-2",
-      "CustomData": [],
-    }, wait=True)
-  except DuploError as e:
-    pytest.fail(f"Failed to create tenant: {e}")
-  try:
-    i = r.find(name)
-    assert i["Name"] == name
-  except DuploError as e:
-    pytest.fail(f"Failed to find infrastructure {name}: {e}")
-  try:
-    r("delete", name)
-  except DuploError as e:
-    pytest.fail(f"Failed to delete infrastructure: {e}")
+  @pytest.mark.integration
+  def test_listing_infrastructures(self):
+    r = duplo.load("infrastructure")
+    try:
+      lot = r("list")
+    except DuploError as e:
+      pytest.fail(f"Failed to list infrastructures: {e}")
+
+  @pytest.mark.integration
+  def test_finding_infra(self):
+    r = duplo.load("tenant")
+    try:
+      t = r("find", "default")
+    except DuploError as e:
+      pytest.fail(f"Failed to find default infra: {e}")
+    assert t["AccountName"] == "default"
 
+  @pytest.mark.integration
+  @pytest.mark.dependency(name = "create_infra")
+  def test_creating_infrastructures(self, infra_name):
+    r = duplo.load("infrastructure")
+    vnum = math.ceil(random.randint(1, 9))
+    name = infra_name
+    print(f"Creating infra '{name}'")
+    try:
+      r.create({
+        "Name": name,
+        "Accountid": "",
+        "EnableK8Cluster": False,
+        "AzCount": 2,
+        "Vnet": { 
+          "SubnetCidr": 22, 
+          "AddressPrefix": f"11.1{vnum}0.0.0/16"
+        },
+        "Cloud": 0,
+        "OnPremConfig": None,
+        "Region": "us-east-1",
+        "CustomData": [],
+      }, wait=True)
+    except DuploFailedResource as e:
+      pytest.fail(f"Infrastructure is in a failed state: {e}")
+    except DuploError as e:
+      pytest.fail(f"Failed to create tenant: {e}")
+
+  @pytest.mark.integration
+  @pytest.mark.dependency(depends=["create_infra"])
+  def test_find_delete_infra(self, infra_name):
+    r = duplo.load("infrastructure")
+    name = infra_name
+    print(f"Deleting infra '{name}'")
+    try:
+      i = r.find(name)
+      assert i["Name"] == name
+    except DuploError as e:
+      pytest.fail(f"Failed to find infrastructure {name}: {e}")
+    try:
+      r("delete", name)
+    except DuploError as e:
+      pytest.fail(f"Failed to delete infrastructure: {e}")

src/tests/test_tenant.py

@@ -7,50 +7,64 @@
 
 duplo, _ = DuploClient.from_env()
 
-@pytest.mark.integration
-def test_listing_tenants():
-  r = duplo.load("tenant")
-  try:
-    lot = r("list")
-  except DuploError as e:
-    pytest.fail(f"Failed to list tenants: {e}")
-  # there is at least one tenant
-  assert len(lot) > 0
-
-@pytest.mark.integration
-def test_finding_tenants():
-  r = duplo.load("tenant")
-  try:
-    t = r("find", "default")
-  except DuploError as e:
-    pytest.fail(f"Failed to list tenants: {e}")
-  assert t["AccountName"] == "default"
-
-@pytest.mark.integration
-def test_creating_tenants():
-  t = duplo.load("tenant")
-  # create a random tenant and delete it from the default plan
+@pytest.fixture(scope="module")
+def tenant_name():
   inc = random.randint(1, 100)
-  name = f"duploctl{inc}"
-  try:
-    t.create({
-      "AccountName": name,
-      "PlanID": "default",
-      "TenantBlueprint": None
-    }, wait=True)
-    print(f"Tenant '{name}' created")
-  except DuploError as e:
-    pytest.fail(f"Failed to create tenant: {e}")
-  # now find it
-  try:
-    nt = t("find", name)
-    assert nt["AccountName"] == name
-  except DuploError as e:
-    pytest.fail(f"Failed to find tenant {name}: {e}")
-  # now delete the tenant
-  try:
-    t("config", name, "-D", "delete_protection")
-    t("delete", name)
-  except DuploError as e:
-    pytest.fail(f"Failed to delete tenant: {e}")
+  return f"duploctl{inc}"
+
+class TestTenant:
+
+  @pytest.mark.integration
+  def test_listing_tenants(self):
+    r = duplo.load("tenant")
+    try:
+      lot = r("list")
+    except DuploError as e:
+      pytest.fail(f"Failed to list tenants: {e}")
+    # there is at least one tenant
+    assert len(lot) > 0
+
+  @pytest.mark.integration
+  def test_finding_tenants(self):
+    r = duplo.load("tenant")
+    try:
+      t = r("find", "default")
+    except DuploError as e:
+      pytest.fail(f"Failed to list tenants: {e}")
+    assert t["AccountName"] == "default"
+
+  @pytest.mark.integration
+  @pytest.mark.dependency(name = "create_tenant")
+  def test_creating_tenants(self, tenant_name):
+    t = duplo.load("tenant")
+    # create a random tenant and delete it from the default plan
+    name = tenant_name
+    try:
+      t.create({
+        "AccountName": name,
+        "PlanID": "default",
+        "TenantBlueprint": None
+      }, wait=True)
+      print(f"Tenant '{name}' created")
+    except DuploError as e:
+      pytest.fail(f"Failed to create tenant: {e}")
+
+  @pytest.mark.integration
+  @pytest.mark.dependency(depends=["create_tenant"])
+  def test_find_delete_tenant(self, tenant_name):
+    # now find it
+    t = duplo.load("tenant")
+    name = tenant_name
+    print(f"Delete tenant '{name}'")
+    try:
+      nt = t("find", name)
+      assert nt["AccountName"] == name
+    except DuploError as e:
+      pytest.fail(f"Failed to find tenant {name}: {e}")
+    # now delete the tenant
+    try:
+      t("config", name, "-D", "delete_protection")
+      t("delete", name)
+    except DuploError as e:
+      pytest.fail(f"Failed to delete tenant: {e}")