[Integration test] Allows for clouds(...) marker and --no-deploy …

…handling (#237)

* Adjust tests to create a test_openstack which can only run on openstack clouds
* Create openstack specific tests
* Repair ceph provisioner name
* repair reader test pod

charms/worker/terraform/README.md

@@ -13,7 +13,7 @@ The module offers the following configurable inputs:
 | Name | Type | Description | Required | Default |
 | - | - | - | - | - |
 | `app_name`| string | Application name | False | k8s-worker | 
-| `base` | string | Ubuntu base to deploy the carm onto | False | [email protected] |
+| `base` | string | Ubuntu base to deploy the charm onto | False | [email protected] |
 | `channel`| string | Channel that the charm is deployed from | False | 1.30/edge |
 | `config`| map(string) | Map of the charm configuration options | False | {} |
 | `constraints` | string | Juju constraints to apply for this application | False | arch=amd64 |

tests/integration/conftest.py

@@ -5,23 +5,27 @@
 import contextlib
 import json
 import logging
+import random
 import shlex
+import string
 from pathlib import Path
 from typing import Optional
 
+import juju.controller
 import juju.utils
+import kubernetes.client.models as k8s_models
 import pytest
 import pytest_asyncio
 import yaml
 from juju.model import Model
 from juju.tag import untag
 from juju.url import URL
 from kubernetes import config as k8s_config
-from kubernetes.client import Configuration
+from kubernetes.client import ApiClient, Configuration, CoreV1Api
 from pytest_operator.plugin import OpsTest
 
 from .cos_substrate import LXDSubstrate
-from .helpers import Bundle, cloud_type, get_unit_cidrs, is_deployed
+from .helpers import Bundle, cloud_type, get_kubeconfig, get_unit_cidrs, is_deployed
 
 log = logging.getLogger(__name__)
 TEST_DATA = Path(__file__).parent / "data"
@@ -77,6 +81,10 @@ def pytest_configure(config):
         "bundle(file='', series='', apps_local={}, apps_channel={}, apps_resources={}): "
         "specify a YAML bundle file for a test.",
     )
+    config.addinivalue_line(
+        "markers",
+        "clouds(*args): mark tests to run only on specific clouds.",
+    )
 
 
 def pytest_collection_modifyitems(config, items):
@@ -95,17 +103,29 @@ def pytest_collection_modifyitems(config, items):
                 item.add_marker(skip_cos)
 
 
+@pytest.fixture(scope="module")
+def module_name(request) -> str:
+    """Get the module name of the test.
+
+    Args:
+        request: Pytest request object.
+
+    Returns:
+        str: The test module name.
+    """
+    return request.module.__name__
+
+
 async def cloud_proxied(ops_test: OpsTest):
     """Setup a cloud proxy settings if necessary
 
-    Test if ghcr.io is reachable through a proxy, if so,
-    Apply expected proxy config to juju model.
+    If ghcr.io is reachable through a proxy apply expected proxy config to juju model.
 
     Args:
         ops_test (OpsTest): ops_test plugin
     """
     assert ops_test.model, "Model must be present"
-    controller = await ops_test.model.get_controller()
+    controller: juju.controller.Controller = await ops_test.model.get_controller()
     controller_model = await controller.get_model("controller")
     proxy_config_file = TEST_DATA / "static-proxy-config.yaml"
     proxy_configs = yaml.safe_load(proxy_config_file.read_text())
@@ -132,6 +152,15 @@ async def cloud_profile(ops_test: OpsTest):
         await ops_test.model.set_config({"container-networking-method": "local", "fan-config": ""})
 
 
+@pytest.fixture(autouse=True)
+async def skip_by_cloud_type(request, ops_test):
+    """Skip tests based on cloud type."""
+    if cloud_markers := request.node.get_closest_marker("clouds"):
+        _type, _ = await cloud_type(ops_test)
+        if _type not in cloud_markers.args:
+            pytest.skip(f"cloud={_type} not among {cloud_markers.args}")
+
+
 @contextlib.asynccontextmanager
 async def deploy_model(
     ops_test: OpsTest,
@@ -183,10 +212,13 @@ async def kubernetes_cluster(request: pytest.FixtureRequest, ops_test: OpsTest):
 
     with ops_test.model_context(model) as the_model:
         if await is_deployed(the_model, bundle.path):
-            log.info("Using existing model.")
+            log.info("Using existing model=%s.", the_model.uuid)
             yield ops_test.model
             return
 
+    if request.config.option.no_deploy:
+        pytest.skip("Skipping because of --no-deploy")
+
     log.info("Deploying new cluster using %s bundle.", bundle.path)
     if request.config.option.apply_proxy:
         await cloud_proxied(ops_test)
@@ -196,6 +228,45 @@ async def kubernetes_cluster(request: pytest.FixtureRequest, ops_test: OpsTest):
         yield the_model
 
 
+def valid_namespace_name(s: str) -> str:
+    """Creates a valid kubernetes namespace name.
+
+    Args:
+        s: The string to sanitize.
+
+    Returns:
+        A valid namespace name.
+    """
+    valid_chars = set(string.ascii_lowercase + string.digits + "-")
+    sanitized = "".join("-" if char not in valid_chars else char for char in s)
+    sanitized = sanitized.strip("-")
+    return sanitized[-63:]
+
+
+@pytest.fixture()
+@pytest.mark.usefixtures("kubernetes_cluster")
+async def api_client(ops_test: OpsTest, module_name: str):
+    """Create a k8s API client and namespace for the test.
+
+    Args:
+        ops_test: The pytest-operator plugin.
+        module_name: The name of the module.
+    """
+    rand_str = "".join(random.choices(string.ascii_lowercase + string.digits, k=5))
+    namespace = valid_namespace_name(f"{module_name}-{rand_str}")
+    kubeconfig_path = await get_kubeconfig(ops_test, module_name)
+    config = type.__call__(Configuration)
+    k8s_config.load_config(client_configuration=config, config_file=str(kubeconfig_path))
+    api_client = ApiClient(configuration=config)
+
+    v1 = CoreV1Api(api_client)
+    v1.create_namespace(
+        body=k8s_models.V1Namespace(metadata=k8s_models.V1ObjectMeta(name=namespace))
+    )
+    yield api_client
+    v1.delete_namespace(name=namespace)
+
+
 @pytest_asyncio.fixture(name="_grafana_agent", scope="module")
 async def grafana_agent(kubernetes_cluster: Model):
     """Deploy Grafana Agent."""

tests/integration/data/test-bundle-openstack.yaml

@@ -0,0 +1,36 @@
+# Copyright 2025 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+name: integration-test
+description: |-
+  Used to deploy or refresh within an integration test model
+series: jammy
+applications:
+  k8s:
+    charm: k8s
+    num_units: 3
+    constraints: cores=2 mem=8G root-disk=16G
+    expose: true
+    options:
+      bootstrap-node-taints: "node-role.kubernetes.io/control-plane=:NoSchedule"
+  k8s-worker:
+    charm: k8s-worker
+    num_units: 2
+    constraints: cores=2 mem=8G root-disk=16G
+  openstack-integrator:
+    charm: openstack-integrator
+    num_units: 1
+    trust: true
+    base: [email protected]
+  openstack-cloud-controller:
+    charm: openstack-cloud-controller
+  cinder-csi:
+    charm: cinder-csi
+relations:
+  - [k8s,                                                k8s-worker:cluster]
+  - [k8s,                                                k8s-worker:containerd]
+  - [openstack-cloud-controller:kube-control,            k8s:kube-control]
+  - [cinder-csi:kube-control,                            k8s:kube-control]
+  - [openstack-cloud-controller:external-cloud-provider, k8s:external-cloud-provider]
+  - [openstack-cloud-controller:openstack,               openstack-integrator:clients]
+  - [cinder-csi:openstack,                               openstack-integrator:clients]

tests/integration/data/test_storage_provider/cinder-pvc.yaml

@@ -0,0 +1,16 @@
+# Copyright 2025 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+apiVersion: v1
+kind: PersistentVolumeClaim
+apiVersion: v1
+metadata:
+  name: raw-block-pvc
+spec:
+  accessModes:
+    - ReadWriteOnce
+  volumeMode: Filesystem
+  resources:
+    requests:
+      storage: 64Mi
+  storageClassName: csi-cinder-default

tests/integration/helpers.py

@@ -8,16 +8,16 @@
 import ipaddress
 import json
 import logging
-import shlex
 from dataclasses import dataclass, field
 from functools import cached_property
 from itertools import chain
 from pathlib import Path
 from typing import Any, Dict, List, Mapping, Optional, Set, Tuple, Union
 
+import juju.application
+import juju.model
+import juju.unit
 import yaml
-from juju import unit
-from juju.model import Model
 from juju.url import URL
 from pytest_operator.plugin import OpsTest
 from tenacity import AsyncRetrying, before_sleep_log, retry, stop_after_attempt, wait_fixed
@@ -26,7 +26,7 @@
 CHARMCRAFT_DIRS = {"k8s": Path("charms/worker/k8s"), "k8s-worker": Path("charms/worker")}
 
 
-async def is_deployed(model: Model, bundle_path: Path) -> bool:
+async def is_deployed(model: juju.model.Model, bundle_path: Path) -> bool:
     """Checks if model has apps defined by the bundle.
     If all apps are deployed, wait for model to be active/idle
 
@@ -60,7 +60,7 @@ async def is_deployed(model: Model, bundle_path: Path) -> bool:
     return True
 
 
-async def get_unit_cidrs(model: Model, app_name: str, unit_num: int) -> List[str]:
+async def get_unit_cidrs(model: juju.model.Model, app_name: str, unit_num: int) -> List[str]:
     """Find unit network cidrs on a unit.
 
     Args:
@@ -87,7 +87,7 @@ async def get_unit_cidrs(model: Model, app_name: str, unit_num: int) -> List[str
     return list(sorted(local_cidrs))
 
 
-async def get_rsc(k8s, resource, namespace=None, labels=None):
+async def get_rsc(k8s, resource, namespace=None, labels=None) -> List[Dict[str, Any]]:
     """Get Resource list optionally filtered by namespace and labels.
 
     Args:
@@ -105,11 +105,18 @@ async def get_rsc(k8s, resource, namespace=None, labels=None):
 
     action = await k8s.run(cmd)
     result = await action.wait()
-    assert result.results["return-code"] == 0, f"Failed to get {resource} with kubectl"
+    stdout, stderr = (result.results.get(field, "").strip() for field in ["stdout", "stderr"])
+    assert result.results["return-code"] == 0, (
+        f"\nFailed to get {resource} with kubectl\n"
+        f"\tstdout: '{stdout}'\n"
+        f"\tstderr: '{stderr}'"
+    )
     log.info("Parsing %s list...", resource)
-    resource_list = json.loads(result.results["stdout"])
-    assert resource_list["kind"] == "List", f"Should have found a list of {resource}"
-    return resource_list["items"]
+    resource_obj = json.loads(stdout)
+    if "/" in resource:
+        return [resource_obj]
+    assert resource_obj["kind"] == "List", f"Should have found a list of {resource}"
+    return resource_obj["items"]
 
 
 @retry(reraise=True, stop=stop_after_attempt(12), wait=wait_fixed(15))
@@ -138,8 +145,8 @@ async def ready_nodes(k8s, expected_count):
 
 
 async def wait_pod_phase(
-    k8s: unit.Unit,
-    name: str,
+    k8s: juju.unit.Unit,
+    name: Optional[str],
     *phase: str,
     namespace: str = "default",
     retry_times: int = 30,
@@ -149,46 +156,27 @@ async def wait_pod_phase(
 
     Args:
         k8s: k8s unit
-        name: the pod name
+        name: the pod name or all pods if None
         phase: expected phase
         namespace: pod namespace
         retry_times: the number of retries
         retry_delay_s: retry interval
 
     """
+    pod_resource = "pod" if name is None else f"pod/{name}"
     async for attempt in AsyncRetrying(
         stop=stop_after_attempt(retry_times),
         wait=wait_fixed(retry_delay_s),
         before_sleep=before_sleep_log(log, logging.WARNING),
     ):
         with attempt:
-            cmd = shlex.join(
-                [
-                    "k8s",
-                    "kubectl",
-                    "get",
-                    "--namespace",
-                    namespace,
-                    "-o",
-                    "jsonpath={.status.phase}",
-                    f"pod/{name}",
-                ]
-            )
-            action = await k8s.run(cmd)
-            result = await action.wait()
-            stdout, stderr = (
-                result.results.get(field, "").strip() for field in ["stdout", "stderr"]
-            )
-            assert result.results["return-code"] == 0, (
-                f"\nPod hasn't reached phase: {phase}\n"
-                f"\tstdout: '{stdout}'\n"
-                f"\tstderr: '{stderr}'"
-            )
-            assert stdout in phase, f"Pod {name} not yet in phase {phase} ({stdout})"
+            for pod in await get_rsc(k8s, pod_resource, namespace=namespace):
+                _phase, _name = pod["status"]["phase"], pod["metadata"]["name"]
+                assert _phase in phase, f"Pod {_name} not yet in phase {phase}"
 
 
 async def get_pod_logs(
-    k8s: unit.Unit,
+    k8s: juju.unit.Unit,
     name: str,
     namespace: str = "default",
 ) -> str:
@@ -228,6 +216,32 @@ async def get_leader(app) -> int:
     raise ValueError("No leader found")
 
 
+async def get_kubeconfig(ops_test, module_name: str):
+    """Retrieve kubeconfig from the k8s leader.
+
+    Args:
+        ops_test: pytest-operator plugin
+        module_name: name of the test module
+
+    Returns:
+        path to the kubeconfig file
+    """
+    kubeconfig_path = ops_test.tmp_path / module_name / "kubeconfig"
+    if kubeconfig_path.exists() and kubeconfig_path.stat().st_size:
+        return kubeconfig_path
+    k8s = ops_test.model.applications["k8s"]
+    leader_idx = await get_leader(k8s)
+    leader = k8s.units[leader_idx]
+    action = await leader.run_action("get-kubeconfig")
+    result = await action.wait()
+    completed = result.status == "completed" or result.results["return-code"] == 0
+    assert completed, f"Failed to get kubeconfig {result=}"
+    kubeconfig_path.parent.mkdir(exist_ok=True, parents=True)
+    kubeconfig_path.write_text(result.results["kubeconfig"])
+    assert Path(kubeconfig_path).stat().st_size, "kubeconfig file is 0 bytes"
+    return kubeconfig_path
+
+
 @dataclass
 class Markings:
     """Test markings for the bundle.