[DPE-6730] Make CI cloud independent

tests/integration/conftest.py

@@ -0,0 +1,13 @@
+#!/usr/bin/env python3
+# Copyright 2025 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+import pytest
+
+
+@pytest.fixture
+def charm():
+    # Return str instead of pathlib.Path since python-libjuju's model.deploy(), juju deploy, and
+    # juju bundle files expect local charms to begin with `./` or `/` to distinguish them from
+    # Charmhub charms.
+    return "./[email protected]"

tests/integration/ha/conftest.py

@@ -14,14 +14,6 @@
 logger = logging.getLogger(__name__)
 
 
-@pytest.fixture
-def charm():
-    # Return str instead of pathlib.Path since python-libjuju's model.deploy(), juju deploy, and
-    # juju bundle files expect local charms to begin with `./` or `/` to distinguish them from
-    # Charmhub charms.
-    return "./[email protected]"
-
-
 @pytest.fixture(scope="function")
 async def reset_restart_delay(ops_test: OpsTest):
     """Resets service file delay on all units."""

tests/integration/ha/test_backups.py

@@ -17,6 +17,7 @@
 
 import asyncio
 import logging
+import os
 import random
 import string
 import subprocess
@@ -328,10 +329,28 @@ async def test_small_deployment_build_and_deploy(
 
     await asyncio.gather(
         ops_test.model.deploy(
-            TLS_CERTIFICATES_APP_NAME, channel=TLS_STABLE_CHANNEL, config=config
+            TLS_CERTIFICATES_APP_NAME,
+            channel=TLS_STABLE_CHANNEL,
+            config=config,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
         ),
-        ops_test.model.deploy(backup_integrator, channel=backup_integrator_channel),
-        ops_test.model.deploy(charm, num_units=3, series=SERIES, config=CONFIG_OPTS),
+        ops_test.model.deploy(
+            backup_integrator,
+            channel=backup_integrator_channel,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
+        ),
+        ops_test.model.deploy(
+            charm,
+            num_units=3,
+            series=SERIES,
+            config=CONFIG_OPTS,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
+        ),
+    )
+
+    subprocess.call(
+        f"juju expose -m {ops_test.model.name} {APP_NAME}",
+        shell=True,
     )
 
     # Relate it to OpenSearch to set up TLS.
@@ -386,32 +405,55 @@ async def test_large_deployment_build_and_deploy(
 
     await asyncio.gather(
         ops_test.model.deploy(
-            TLS_CERTIFICATES_APP_NAME, channel=TLS_STABLE_CHANNEL, config=tls_config
+            TLS_CERTIFICATES_APP_NAME,
+            channel=TLS_STABLE_CHANNEL,
+            config=tls_config,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
+        ),
+        ops_test.model.deploy(
+            backup_integrator,
+            channel=backup_integrator_channel,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
         ),
-        ops_test.model.deploy(backup_integrator, channel=backup_integrator_channel),
         ops_test.model.deploy(
             charm,
             application_name="main",
             num_units=1,
             series=SERIES,
             config=main_orchestrator_conf | CONFIG_OPTS,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
         ),
         ops_test.model.deploy(
             charm,
             application_name="failover",
             num_units=2,
             series=SERIES,
             config=failover_orchestrator_conf | CONFIG_OPTS,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
         ),
         ops_test.model.deploy(
             charm,
             application_name=APP_NAME,
             num_units=1,
             series=SERIES,
             config=data_hot_conf | CONFIG_OPTS,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
         ),
     )
 
+    subprocess.call(
+        f"juju expose -m {ops_test.model.name} {APP_NAME}",
+        shell=True,
+    )
+    subprocess.call(
+        f"juju expose -m {ops_test.model.name} main",
+        shell=True,
+    )
+    subprocess.call(
+        f"juju expose -m {ops_test.model.name} failover",
+        shell=True,
+    )
+
     # Large deployment setup
     await ops_test.model.integrate("main:peer-cluster-orchestrator", "failover:peer-cluster")
     await ops_test.model.integrate("main:peer-cluster-orchestrator", f"{APP_NAME}:peer-cluster")
@@ -694,10 +736,28 @@ async def test_restore_to_new_cluster(
 
     await asyncio.gather(
         ops_test.model.deploy(
-            TLS_CERTIFICATES_APP_NAME, channel=TLS_STABLE_CHANNEL, config=config
+            TLS_CERTIFICATES_APP_NAME,
+            channel=TLS_STABLE_CHANNEL,
+            config=config,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
+        ),
+        ops_test.model.deploy(
+            backup_integrator,
+            channel=backup_integrator_channel,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
+        ),
+        ops_test.model.deploy(
+            charm,
+            num_units=3,
+            series=SERIES,
+            config=CONFIG_OPTS,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
         ),
-        ops_test.model.deploy(backup_integrator, channel=backup_integrator_channel),
-        ops_test.model.deploy(charm, num_units=3, series=SERIES, config=CONFIG_OPTS),
+    )
+
+    subprocess.call(
+        f"juju expose -m {ops_test.model.name} {APP_NAME}",
+        shell=True,
     )
 
     # Relate it to OpenSearch to set up TLS.
@@ -800,10 +860,28 @@ async def test_build_deploy_and_test_status(ops_test: OpsTest, charm) -> None:
     config = {"ca-common-name": "CN_CA"}
     await asyncio.gather(
         ops_test.model.deploy(
-            TLS_CERTIFICATES_APP_NAME, channel=TLS_STABLE_CHANNEL, config=config
+            TLS_CERTIFICATES_APP_NAME,
+            channel=TLS_STABLE_CHANNEL,
+            config=config,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
         ),
-        ops_test.model.deploy(S3_INTEGRATOR, channel=S3_INTEGRATOR_CHANNEL),
-        ops_test.model.deploy(charm, num_units=3, series=SERIES, config=CONFIG_OPTS),
+        ops_test.model.deploy(
+            S3_INTEGRATOR,
+            channel=S3_INTEGRATOR_CHANNEL,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
+        ),
+        ops_test.model.deploy(
+            charm,
+            num_units=3,
+            series=SERIES,
+            config=CONFIG_OPTS,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
+        ),
+    )
+
+    subprocess.call(
+        f"juju expose -m {ops_test.model.name} {APP_NAME}",
+        shell=True,
     )
 
     # Relate it to OpenSearch to set up TLS.

tests/integration/ha/test_ha.py

@@ -4,6 +4,8 @@
 
 import asyncio
 import logging
+import os
+import subprocess
 import time
 
 import pytest
@@ -51,22 +53,36 @@
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
 @pytest.mark.skip_if_deployed
-async def test_build_and_deploy(ops_test: OpsTest) -> None:
+async def test_build_and_deploy(ops_test: OpsTest, charm) -> None:
     """Build and deploy one unit of OpenSearch."""
     # it is possible for users to provide their own cluster for HA testing.
     # Hence, check if there is a pre-existing cluster.
     if await app_name(ops_test):
         return
 
-    my_charm = await ops_test.build_charm(".")
     await ops_test.model.set_config(MODEL_CONFIG)
     # Deploy TLS Certificates operator.
     config = {"ca-common-name": "CN_CA"}
     await asyncio.gather(
         ops_test.model.deploy(
-            TLS_CERTIFICATES_APP_NAME, channel=TLS_STABLE_CHANNEL, config=config
+            TLS_CERTIFICATES_APP_NAME,
+            channel=TLS_STABLE_CHANNEL,
+            config=config,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
         ),
-        ops_test.model.deploy(my_charm, num_units=NUM_HA_UNITS, series=SERIES, config=CONFIG_OPTS),
+        ops_test.model.deploy(
+            charm,
+            application_name="opensearch",
+            num_units=NUM_HA_UNITS,
+            series=SERIES,
+            config=CONFIG_OPTS,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
+        ),
+    )
+
+    subprocess.call(
+        f"juju expose -m {ops_test.model.name} opensearch",
+        shell=True,
     )
 
     # Relate it to OpenSearch to set up TLS.

tests/integration/ha/test_ha_multi_clusters.py

@@ -4,6 +4,8 @@
 
 import asyncio
 import logging
+import os
+import subprocess
 
 import pytest
 from pytest_operator.plugin import OpsTest
@@ -43,9 +45,23 @@ async def test_build_and_deploy(ops_test: OpsTest, charm) -> None:
     config = {"ca-common-name": "CN_CA"}
     await asyncio.gather(
         ops_test.model.deploy(
-            TLS_CERTIFICATES_APP_NAME, channel=TLS_STABLE_CHANNEL, config=config
+            TLS_CERTIFICATES_APP_NAME,
+            channel=TLS_STABLE_CHANNEL,
+            config=config,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
         ),
-        ops_test.model.deploy(charm, num_units=2, series=SERIES, config=CONFIG_OPTS),
+        ops_test.model.deploy(
+            charm,
+            num_units=2,
+            series=SERIES,
+            config=CONFIG_OPTS,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
+        ),
+    )
+
+    subprocess.call(
+        f"juju expose -m {ops_test.model.name} {APP_NAME}",
+        shell=True,
     )
 
     # Relate it to OpenSearch to set up TLS.
@@ -75,7 +91,11 @@ async def test_multi_clusters_db_isolation(
 
     # deploy new cluster
     await ops_test.model.deploy(
-        charm, num_units=1, application_name=SECOND_APP_NAME, config=CONFIG_OPTS
+        charm,
+        num_units=1,
+        application_name=SECOND_APP_NAME,
+        config=CONFIG_OPTS,
+        constraints=os.environ.get("TEST_CONSTRAINTS"),
     )
     await ops_test.model.integrate(SECOND_APP_NAME, TLS_CERTIFICATES_APP_NAME)
 

tests/integration/ha/test_ha_networking.py

@@ -4,6 +4,8 @@
 
 import asyncio
 import logging
+import os
+import subprocess
 
 import pytest
 from pytest_operator.plugin import OpsTest
@@ -27,6 +29,7 @@
     SERIES,
     app_name,
     check_cluster_formation_successful,
+    filter_cloud,
     get_application_unit_ids_hostnames,
     get_application_unit_ids_ips,
     get_application_unit_names,
@@ -42,6 +45,7 @@
 logger = logging.getLogger(__name__)
 
 
+@filter_cloud(["lxd"])
 @pytest.mark.runner(["self-hosted", "linux", "X64", "jammy"])
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
@@ -58,9 +62,23 @@ async def test_build_and_deploy(ops_test: OpsTest, charm) -> None:
     config = {"ca-common-name": "CN_CA"}
     await asyncio.gather(
         ops_test.model.deploy(
-            TLS_CERTIFICATES_APP_NAME, channel=TLS_STABLE_CHANNEL, config=config
+            TLS_CERTIFICATES_APP_NAME,
+            channel=TLS_STABLE_CHANNEL,
+            config=config,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
         ),
-        ops_test.model.deploy(charm, num_units=3, series=SERIES, config=CONFIG_OPTS),
+        ops_test.model.deploy(
+            charm,
+            num_units=3,
+            series=SERIES,
+            config=CONFIG_OPTS,
+            constraints=os.environ.get("TEST_CONSTRAINTS"),
+        ),
+    )
+
+    subprocess.call(
+        f"juju expose -m {ops_test.model.name} {APP_NAME}",
+        shell=True,
     )
 
     # Relate it to OpenSearch to set up TLS.
@@ -74,6 +92,7 @@ async def test_build_and_deploy(ops_test: OpsTest, charm) -> None:
     assert len(ops_test.model.applications[APP_NAME].units) == 3
 
 
+@filter_cloud(["lxd"])
 @pytest.mark.runner(["self-hosted", "linux", "X64", "jammy"])
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
@@ -177,6 +196,7 @@ async def test_full_network_cut_with_ip_change_node_with_elected_cm(
     await assert_continuous_writes_consistency(ops_test, c_writes, [app])
 
 
+@filter_cloud(["lxd"])
 @pytest.mark.runner(["self-hosted", "linux", "X64", "jammy"])
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
@@ -296,6 +316,7 @@ async def test_full_network_cut_with_ip_change_node_with_primary_shard(
     await assert_continuous_writes_consistency(ops_test, c_writes, [app])
 
 
+@filter_cloud(["lxd"])
 @pytest.mark.runner(["self-hosted", "linux", "X64", "jammy"])
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
@@ -385,6 +406,7 @@ async def test_full_network_cut_without_ip_change_node_with_elected_cm(
     await assert_continuous_writes_consistency(ops_test, c_writes, [app])
 
 
+@filter_cloud(["lxd"])
 @pytest.mark.runner(["self-hosted", "linux", "X64", "jammy"])
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail