Produce apis to add a k8s cloud and tear down its models first

.github/workflows/tests.yaml

@@ -6,7 +6,7 @@ on:
 jobs:
   call-inclusive-naming-check:
     name: Inclusive naming
-    uses: canonical-web-and-design/Inclusive-naming/.github/workflows/woke.yaml@main
+    uses: canonical/inclusive-naming/.github/workflows/woke.yaml@main
     with:
       fail-on-error: "true"
 

.gitignore

@@ -7,4 +7,6 @@ build
 dist/
 *.orig
 report
-.coverage
+.coverage
+juju-crashdump*
+

pytest_operator/plugin.py

@@ -1,4 +1,5 @@
 import asyncio
+import base64
 import contextlib
 import dataclasses
 import enum
@@ -38,11 +39,15 @@
 from zipfile import Path as ZipPath
 
 import jinja2
+import kubernetes.config
 import pytest
 import pytest_asyncio.plugin
 import yaml
 from _pytest.config import Config
 from _pytest.config.argparsing import Parser
+from kubernetes import client as k8s_client
+from kubernetes.client import Configuration as K8sConfiguration
+from juju.client import client
 from juju.client.jujudata import FileJujuData
 from juju.exceptions import DeadEntityException
 from juju.errors import JujuError
@@ -399,6 +404,10 @@ class ModelInUseError(Exception):
     """Raise when trying to add a model alias which already exists."""
 
 
+BundleOpt = Union[str, Path, "OpsTest.Bundle"]
+Timeout = TypeVar("Timeout", float, int)
+
+
 @dataclasses.dataclass
 class ModelState:
     model: Model
@@ -408,14 +417,18 @@ class ModelState:
     model_name: str
     config: Optional[dict] = None
     tmp_path: Optional[Path] = None
+    timeout: Optional[Timeout] = None
 
     @property
     def full_name(self) -> str:
         return f"{self.controller_name}:{self.model_name}"
 
 
-BundleOpt = TypeVar("BundleOpt", str, Path, "OpsTest.Bundle")
-Timeout = TypeVar("Timeout", float, int)
+@dataclasses.dataclass
+class CloudState:
+    cloud_name: str
+    models: List[str] = dataclasses.field(default_factory=list)
+    timeout: Optional[Timeout] = None
 
 
 class OpsTest:
@@ -510,6 +523,7 @@ def __init__(self, request, tmp_path_factory):
         # use an OrderedDict so that the first model made is destroyed last.
         self._current_alias = None
         self._models: MutableMapping[str, ModelState] = OrderedDict()
+        self._clouds: MutableMapping[str, CloudState] = OrderedDict()
 
     @contextlib.contextmanager
     def model_context(self, alias: str) -> Generator[Model, None, None]:
@@ -597,14 +611,16 @@ def keep_model(self) -> bool:
         current_state = self.current_alias and self._models.get(self.current_alias)
         return current_state.keep if current_state else self._init_keep_model
 
-    def _generate_model_name(self) -> str:
+    def _generate_name(self, kind: str) -> str:
         module_name = self.request.module.__name__.rpartition(".")[-1]
         suffix = "".join(choices(ascii_lowercase + digits, k=4))
+        if kind != "model":
+            suffix = "-".join((kind, suffix))
         return f"{module_name.replace('_', '-')}-{suffix}"
 
     @cached_property
     def default_model_name(self) -> str:
-        return self._generate_model_name()
+        return self._generate_name(kind="model")
 
     async def run(
         self,
@@ -670,23 +686,33 @@ async def _add_model(self, cloud_name, model_name, keep=False, **kwargs):
         """
         controller = self._controller
         controller_name = controller.controller_name
+        credential_name = None
+        timeout = None
         if not cloud_name:
             # if not provided, try the default cloud name
             cloud_name = self._init_cloud_name
         if not cloud_name:
             # if not provided, use the controller's default cloud
             cloud_name = await controller.get_cloud()
+        if ops_cloud := self._clouds.get(cloud_name):
+            credential_name = cloud_name
+            timeout = ops_cloud.timeout
+
         model_full_name = f"{controller_name}:{model_name}"
         log.info(f"Adding model {model_full_name} on cloud {cloud_name}")
 
-        model = await controller.add_model(model_name, cloud_name, **kwargs)
+        model = await controller.add_model(
+            model_name, cloud_name, credential_name=credential_name, **kwargs
+        )
         # NB: This call to `juju models` is needed because libjuju's
         # `add_model` doesn't update the models.yaml cache that the Juju
         # CLI depends on with the model's UUID, which the CLI requires to
         # connect. Calling `juju models` beforehand forces the CLI to
         # update the cache from the controller.
         await self.juju("models")
-        state = ModelState(model, keep, controller_name, cloud_name, model_name)
+        state = ModelState(
+            model, keep, controller_name, cloud_name, model_name, timeout=timeout
+        )
         state.config = await model.get_config()
         return state
 
@@ -820,11 +846,13 @@ async def track_model(
             )
         else:
             cloud_name = cloud_name or self.cloud_name
-            model_name = model_name or self._generate_model_name()
+            model_name = model_name or self._generate_name(kind="model")
             model_state = await self._add_model(
                 cloud_name, model_name, keep_val, **kwargs
             )
         self._models[alias] = model_state
+        if ops_cloud := self._clouds.get(cloud_name):
+            ops_cloud.models.append(alias)
         return model_state.model
 
     async def log_model(self):
@@ -886,6 +914,13 @@ async def forget_model(
         if alias not in self.models:
             raise ModelNotFoundError(f"{alias} not found")
 
+        model_state: ModelState = self._models[alias]
+        if timeout is None and model_state.timeout:
+            timeout = model_state.timeout
+
+        async def is_model_alive():
+            return model_name in await self._controller.list_models()
+
         with self.model_context(alias) as model:
             await self.log_model()
             model_name = model.info.name
@@ -896,13 +931,23 @@ async def forget_model(
             if not self.keep_model:
                 await self._reset(model, allow_failure, timeout=timeout)
                 await self._controller.destroy_model(
-                    model_name, force=True, destroy_storage=destroy_storage
+                    model_name,
+                    force=True,
+                    destroy_storage=destroy_storage,
+                    max_wait=timeout,
                 )
+                if timeout and await is_model_alive():
+                    log.warning("Waiting for model %s to die...", model_name)
+                    while await is_model_alive():
+                        await asyncio.sleep(5)
+
             await model.disconnect()
 
         # stop managing this model now
-        log.info(f"Forgetting {alias}...")
+        log.info(f"Forgetting model {alias}...")
         self._models.pop(alias)
+        if ops_cloud := self._clouds.get(model_state.cloud_name):
+            ops_cloud.models.remove(alias)
         if alias is self.current_alias:
             self._current_alias = None
 
@@ -933,7 +978,9 @@ async def _destroy(entity_name: str, **kwargs):
 
         try:
             await model.block_until(
-                lambda: len(model.machines) == 0 and len(model.applications) == 0,
+                lambda: len(model.units) == 0
+                and len(model.machines) == 0
+                and len(model.applications) == 0,
                 timeout=timeout,
             )
         except asyncio.TimeoutError:
@@ -948,10 +995,15 @@ async def _destroy(entity_name: str, **kwargs):
             log.info(f"Reset {model.info.name} completed successfully.")
 
     async def _cleanup_models(self):
+        # remove clouds from most recently made, to first made
+        # each model in the cloud will be forgotten
+        for cloud in reversed(self._clouds):
+            await self.forget_cloud(cloud)
+
         # remove models from most recently made, to first made
         aliases = list(reversed(self._models.keys()))
-        for models in aliases:
-            await self.forget_model(models)
+        for model in aliases:
+            await self.forget_model(model)
 
         await self._controller.disconnect()
 
@@ -1491,3 +1543,126 @@ def is_crash_dump_enabled(self) -> bool:
             return True
         else:
             return False
+
+    async def add_k8s(
+        self,
+        cloud_name: Optional[str] = None,
+        kubeconfig: Optional[K8sConfiguration] = None,
+        context: Optional[str] = None,
+        skip_storage: bool = True,
+        storage_class: Optional[str] = None,
+    ) -> str:
+        """
+        Add a new k8s cloud in the existing controller.
+
+        @param Optional[str] cloud_name:
+            Name for the new cloud
+            None will autogenerate a name
+        @param Optional[kubernetes.client.configuration.Configuration] kubeconfig:
+            Configuration object from kubernetes.config.load_config
+            None will read from the usual kubeconfig locations like
+                os.environ.get('KUBECONFIG', '$HOME/.kube/config')
+        @param Optional[str] context:
+            context to use within the kubeconfig
+            None will use the default context
+        @param bool skip_storage:
+            True will not use cloud storage,
+            False either finds storage or uses storage_class
+        @param Optional[str] skip_storage:
+            cluster storage-class to use for juju storage
+            None will look for a default storage class within the cluster
+
+        @returns str: cloud_name
+
+        Common Examples:
+        ----------------------------------
+        # make a new k8s cloud with any juju name and destroy it when the tests are over
+        await ops_test.add_k8s()
+
+        # make a cloud known to juju as "bob"
+        await ops_test.add_k8s(cloud_name="my-k8s")
+        ----------------------------------
+        """
+
+        if kubeconfig is None:
+            # kubeconfig should be auto-detected from the usual places
+            kubeconfig = type.__call__(K8sConfiguration)
+            kubernetes.config.load_config(
+                client_configuration=kubeconfig,
+                context=context,
+                temp_file_path=self.tmp_path,
+            )
+        juju_cloud_config = {}
+        if not skip_storage and storage_class is None:
+            # lookup default storage-class
+            api_client = kubernetes.client.ApiClient(configuration=kubeconfig)
+            cluster = k8s_client.StorageV1Api(api_client=api_client)
+            for sc in cluster.list_storage_class().items:
+                if (
+                    sc.metadata.annotations.get(
+                        "storageclass.kubernetes.io/is-default-class"
+                    )
+                    == "true"
+                ):
+                    storage_class = sc.metadata.name
+        if not skip_storage and storage_class:
+            juju_cloud_config["workload-storage"] = storage_class
+            juju_cloud_config["operator-storage"] = storage_class
+
+        controller = self._controller
+        cloud_name = cloud_name or self._generate_name("k8s-cloud")
+        log.info(f"Adding k8s cloud {cloud_name}")
+
+        cloud_def = client.Cloud(
+            auth_types=[
+                "certificate",
+                "clientcertificate",
+                "oauth2",
+                "oauth2withcert",
+                "userpass",
+            ],
+            ca_certificates=[Path(kubeconfig.ssl_ca_cert).read_text()],
+            endpoint=kubeconfig.host,
+            host_cloud_region="kubernetes/ops-test",
+            regions=[client.CloudRegion(endpoint=kubeconfig.host, name="default")],
+            skip_tls_verify=not kubeconfig.verify_ssl,
+            type_="kubernetes",
+            config=juju_cloud_config,
+        )
+
+        if kubeconfig.cert_file and kubeconfig.key_file:
+            auth_type = "clientcertificate"
+            attrs = dict(
+                ClientCertificateData=Path(kubeconfig.cert_file).read_text(),
+                ClientKeyData=Path(kubeconfig.key_file).read_text(),
+            )
+        elif token := kubeconfig.api_key["authorization"]:
+            if token.startswith("Bearer "):
+                auth_type = "oauth2"
+                attrs = {"Token": token.split(" ")[1]}
+            elif token.startswith("Basic "):
+                auth_type, userpass = "userpass", token.split(" ")[1]
+                user, passwd = base64.b64decode(userpass).decode().split(":", 1)
+                attrs = {"username": user, "password": passwd}
+            else:
+                raise ValueError("Failed to find credentials in authorization token")
+        else:
+            raise ValueError("Failed to find credentials in kubernetes.Configuration")
+
+        await controller.add_cloud(cloud_name, cloud_def)
+        await controller.add_credential(
+            cloud_name,
+            credential=client.CloudCredential(attrs, auth_type),
+            cloud=cloud_name,
+        )
+        self._clouds[cloud_name] = CloudState(cloud_name, timeout=5 * 60)
+        return cloud_name
+
+    async def forget_cloud(self, cloud_name: str):
+        if cloud_name not in self._clouds:
+            raise KeyError(f"{cloud_name} not in clouds")
+        for model in reversed(self._clouds[cloud_name].models):
+            await self.forget_model(model, destroy_storage=True)
+        log.info(f"Forgetting cloud: {cloud_name}...")
+        await self._controller.remove_cloud(cloud_name)
+        del self._clouds[cloud_name]

tests/data/test_lib/setup.py

@@ -3,9 +3,9 @@
 from setuptools import setup
 
 setup(
-      name="pytest-operator-test-lib",
-      version="1.0.0",
-      description="Pytest-operator test library.",
-      author="tester",
-      packages=[],
+    name="pytest-operator-test-lib",
+    version="1.0.0",
+    description="Pytest-operator test library.",
+    author="tester",
+    packages=[],
 )

tests/integration/test_add_k8s.py

@@ -0,0 +1,20 @@
+# test that pytest operator supports adding a k8s to an existing controller
+# This is a new k8s cloud created/managed by pytest-operator
+
+from kubernetes.config.config_exception import ConfigException
+import pytest
+from pytest_operator.plugin import OpsTest
+
+
+async def test_add_k8s(ops_test: OpsTest):
+    try:
+        k8s_cloud = await ops_test.add_k8s(skip_storage=False)
+    except (ConfigException, TypeError):
+        pytest.skip("No Kubernetes config found to add-k8s")
+
+    k8s_model = await ops_test.track_model(
+        "secondary", cloud_name=k8s_cloud, keep=ops_test.ModelKeep.NEVER
+    )
+    with ops_test.model_context("secondary"):
+        await k8s_model.deploy("grafana-k8s", trust=True)
+        await k8s_model.wait_for_idle(apps=["grafana-k8s"], status="active")

tests/integration/test_pytest_operator.py

@@ -91,6 +91,9 @@ async def test_2_create_delete_new_model(self, ops_test):
         }
 
         # track the main model with a second alias, don't do this other than testing
+        # this will forget the duplicate without resetting/deleting the main model
+        # because "duplicate" will be in "keep_model" mode since ops_tests believes
+        # it's an existing model.
         model_name = prior_model.info.name
         duplicate = await ops_test.track_model("duplicate", model_name=model_name)
         assert duplicate.info.uuid == prior_model.info.uuid