Adjust tests to create a test_openstack which can only run on opensta…

…ck clouds

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

@@ -9,6 +9,7 @@
 from pathlib import Path
 from typing import Optional
 
+import juju.controller
 import juju.utils
 import pytest
 import pytest_asyncio
@@ -77,6 +78,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):
@@ -98,14 +103,13 @@ def pytest_collection_modifyitems(config, items):
 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 +136,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 +196,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)

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/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: