ENH: check missing GCP services

src/_nebari/provider/cloud/google_cloud.py

@@ -2,7 +2,7 @@
 import json
 import os
 import subprocess
-from typing import Dict, List
+from typing import Dict, List, Set
 
 from _nebari import constants
 from _nebari.provider.cloud.commons import filter_by_highest_supported_k8s_version
@@ -89,6 +89,22 @@ def instances(project: str) -> Dict[str, str]:
     return {_["description"]: _["name"] for _ in data}
 
 
+def activated_services() -> Set[str]:
+    """Return a list of activated services."""
+    check_credentials()
+    output = subprocess.check_output(
+        [
+            "gcloud",
+            "services",
+            "list",
+            "--enabled",
+            "--format=json(config.title)",
+        ]
+    )
+    data = json.loads(output)
+    return {service["config"]["title"] for service in data}
+
+
 def cluster_exists(cluster_name: str, project_id: str, region: str) -> bool:
     """Check if a GKE cluster exists."""
     try:
@@ -250,6 +266,26 @@ def gcp_cleanup(config: schema.Main):
     delete_service_account(service_account_name, project_id)
 
 
+def check_missing_service() -> None:
+    """Check if all required services are activated."""
+    required = {
+        "Compute Engine API",
+        "Kubernetes Engine API",
+        "Cloud Monitoring API",
+        "Cloud Autoscaling API",
+        "Identity and Access Management (IAM) API",
+        "Cloud Resource Manager API",
+    }
+    activated = activated_services()
+    common = required.intersection(activated)
+    missing = required.difference(common)
+    if missing:
+        raise ValueError(
+            f"""Missing required services: {missing}\n
+            Please see the documentation for more information: {constants.GCP_ENV_DOCS}"""
+        )
+
+
 # Getting pricing data could come from here
 # https://cloudpricingcalculator.appspot.com/static/data/pricelist.json
 

tests/tests_unit/test_provider.py

@@ -0,0 +1,54 @@
+from contextlib import nullcontext
+
+import pytest
+
+from _nebari.provider.cloud.google_cloud import check_missing_service
+
+
+@pytest.mark.parametrize(
+    "activated_services, exception",
+    [
+        (
+            {
+                "Compute Engine API",
+                "Kubernetes Engine API",
+                "Cloud Monitoring API",
+                "Cloud Autoscaling API",
+                "Identity and Access Management (IAM) API",
+                "Cloud Resource Manager API",
+            },
+            nullcontext(),
+        ),
+        (
+            {
+                "Compute Engine API",
+                "Kubernetes Engine API",
+                "Cloud Monitoring API",
+                "Cloud Autoscaling API",
+                "Identity and Access Management (IAM) API",
+                "Cloud Resource Manager API",
+                "Cloud SQL Admin API",
+            },
+            nullcontext(),
+        ),
+        (
+            {
+                "Compute Engine API",
+                "Kubernetes Engine API",
+                "Cloud Monitoring API",
+                "Cloud Autoscaling API",
+                "Cloud SQL Admin API",
+            },
+            pytest.raises(ValueError, match=r"Missing required services:.*"),
+        ),
+    ],
+)
+def test_gcp_missing_service(monkeypatch, activated_services, exception):
+    def mock_return():
+        return activated_services
+
+    monkeypatch.setattr(
+        "_nebari.provider.cloud.google_cloud.activated_services", mock_return
+    )
+    with exception:
+        check_missing_service()