Initial work to create and delete droplets

scripts/cicd-python/utils/droplets.py

@@ -0,0 +1,113 @@
+import aiohttp
+import asyncio
+import sys
+
+# Define droplet configurations for different droplet types.
+DROPLET_CONFIGS = {
+    "amd64-deb": {
+        "name": "build-server-amd64-deb-tmp",
+        "image": 179066895,
+        "size": "c2-16vcpu-32gb"
+    },
+}
+
+async def create_droplet(token: str, droplet_type: str) -> None:
+    config = DROPLET_CONFIGS.get(droplet_type)
+    if not config:
+        print(f"Droplet type '{droplet_type}' not recognized.", file=sys.stderr)
+        sys.exit(1)
+    
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+    }
+    create_url = "https://api.digitalocean.com/v2/droplets"
+    payload = {
+        "name": config["name"],
+        "region": "nyc1",  # Changed default region to "ncy1"
+        "size": config["size"],
+        "image": config["image"],
+        "backups": False,
+    }
+    
+    async with aiohttp.ClientSession() as session:
+        async with session.post(create_url, headers=headers, json=payload) as resp:
+            if resp.status not in (201, 202):
+                error_text = await resp.text()
+                print(f"Error creating droplet: {error_text}", file=sys.stderr)
+                sys.exit(error_text)
+            data = await resp.json()
+            droplet = data.get("droplet")
+            if not droplet:
+                print("No droplet information returned.", file=sys.stderr)
+                sys.exit("No droplet information returned.")
+            droplet_id = droplet.get("id")
+            print(f"Droplet creation initiated. Droplet ID: {droplet_id}")
+            
+            # Poll for droplet status until it becomes "active"
+            status = droplet.get("status", "new")
+            droplet_url = f"https://api.digitalocean.com/v2/droplets/{droplet_id}"
+            while status != "active":
+                await asyncio.sleep(2)
+                async with session.get(droplet_url, headers=headers) as poll_resp:
+                    if poll_resp.status != 200:
+                        error_text = await poll_resp.text()
+                        print(f"Error polling droplet status: {error_text}", file=sys.stderr)
+                        sys.exit(error_text)
+                    droplet_data = await poll_resp.json()
+                    droplet = droplet_data.get("droplet")
+                    if droplet:
+                        status = droplet.get("status", status)
+                        print(f"Droplet status: {status}")
+                    else:
+                        print("Droplet data missing in polling response", file=sys.stderr)
+                        sys.exit("Droplet data missing in polling response")
+            
+            print("Droplet is up and running.")
+            # Once active, send a final GET request to output the droplet's information.
+            async with session.get(droplet_url, headers=headers) as final_resp:
+                if final_resp.status != 200:
+                    error_text = await final_resp.text()
+                    print(f"Error retrieving droplet information: {error_text}", file=sys.stderr)
+                    sys.exit(error_text)
+                final_data = await final_resp.json()
+                print("Droplet Information:")
+                print(final_data)
+
+async def delete_droplet(token: str, droplet_type: str) -> None:
+    config = DROPLET_CONFIGS.get(droplet_type)
+    if not config:
+        print(f"Droplet type '{droplet_type}' not recognized.", file=sys.stderr)
+        sys.exit(1)
+    
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+    }
+    droplets_url = "https://api.digitalocean.com/v2/droplets"
+    
+    async with aiohttp.ClientSession() as session:
+        async with session.get(droplets_url, headers=headers) as resp:
+            if resp.status != 200:
+                error_text = await resp.text()
+                print(f"Error retrieving droplets: {error_text}", file=sys.stderr)
+                sys.exit(error_text)
+            data = await resp.json()
+            droplets = data.get("droplets", [])
+            target_droplet = None
+            for droplet in droplets:
+                if droplet.get("name") == config["name"]:
+                    target_droplet = droplet
+                    break
+            if not target_droplet:
+                print(f"No droplet found with name '{config['name']}'.")
+                return
+            
+            droplet_id = target_droplet.get("id")
+            delete_url = f"https://api.digitalocean.com/v2/droplets/{droplet_id}"
+            async with session.delete(delete_url, headers=headers) as delete_resp:
+                if delete_resp.status != 204:
+                    error_text = await delete_resp.text()
+                    print(f"Error deleting droplet: {error_text}", file=sys.stderr)
+                    sys.exit(error_text)
+                print(f"Droplet '{config['name']}' deleted successfully.")

scripts/cicd-python/utils/test_credentials.py

@@ -0,0 +1,20 @@
+import aiohttp
+import sys
+
+async def test_api_credentials(token: str) -> None:
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+    }
+    url = "https://api.digitalocean.com/v2/droplets"
+    async with aiohttp.ClientSession() as session:
+        async with session.get(url, headers=headers) as resp:
+            if resp.status == 200:
+                data = await resp.json()
+                droplets = data.get("droplets", [])
+                print("API credentials are valid.")
+                print(f"Retrieved {len(droplets)} droplet(s).")
+            else:
+                error_text = await resp.text()
+                print("Failed to authenticate API credentials:", error_text, file=sys.stderr)
+                sys.exit(error_text)