[draft] Add cross compat tests for mongodb, mysql, postgresql charms

s3/tests/integration/conftest.py

@@ -14,8 +14,15 @@
 
 import jubilant
 import pytest
-from domain import S3ConnectionInfo
-from helpers import create_bucket, create_iam_user, delete_bucket, local_tmp_folder
+
+from .domain import S3ConnectionInfo
+from .helpers import (
+    create_bucket,
+    create_iam_user,
+    delete_bucket,
+    get_s3_charm_path,
+    local_tmp_folder,
+)
 
 logger = logging.getLogger(__name__)
 logging.getLogger("jubilant.wait").setLevel(logging.WARNING)
@@ -43,8 +50,7 @@ def platform() -> str:
 @pytest.fixture
 def s3_charm(platform: str) -> Path:
     """Path to the packed s3-integrator charm."""
-    if not (path := next(iter(Path.cwd().glob(f"*-{platform}.charm")), None)):
-        raise FileNotFoundError("Could not find packed s3-integrator charm.")
+    path = get_s3_charm_path()
     logger.info(f"Using s3-integrator charm at: {path}")
     return path
 

s3/tests/integration/helpers.py

@@ -12,11 +12,13 @@
 import tempfile
 from contextlib import contextmanager
 from pathlib import Path
+from platform import machine
 
 import boto3
 import jubilant
 from botocore.exceptions import ClientError, ConnectTimeoutError, ParamValidationError, SSLError
-from domain import S3ConnectionInfo
+
+from .domain import S3ConnectionInfo
 
 logger = logging.getLogger(__name__)
 
@@ -208,3 +210,20 @@ def b64_to_ca_chain_json_dumps(ca_chain: str) -> str:
     if not chain_list:
         raise ValueError("No certificate found in chain file")
     return str(chain_list)
+
+
+def get_platform() -> str:
+    """Detect the current platform architecture."""
+    platforms = {
+        "x86_64": "amd64",
+        "aarch64": "arm64",
+    }
+    return platforms.get(machine(), "amd64")
+
+
+def get_s3_charm_path() -> Path:
+    """Get the path to the packed s3-integrator charm for the current platform."""
+    platform = get_platform()
+    if not (path := next(iter(Path.cwd().glob(f"*-{platform}.charm")), None)):
+        raise FileNotFoundError("Could not find packed s3-integrator charm.")
+    return path

s3/tests/integration/test_config.py

@@ -8,8 +8,9 @@
 
 import jubilant
 import pytest
-from domain import S3ConnectionInfo
-from helpers import delete_bucket, get_bucket
+
+from .domain import S3ConnectionInfo
+from .helpers import delete_bucket, get_bucket
 
 S3 = "s3-integrator"
 SECRET_LABEL = "s3-creds-secret-config"

s3/tests/integration/test_consumers/conftest.py

@@ -0,0 +1,172 @@
+import random
+import string
+from pathlib import Path
+
+import jubilant
+import pytest
+
+from ..domain import S3ConnectionInfo
+from .helpers import CharmSpec
+
+
+def pytest_addoption(parser):
+    group = parser.getgroup("charm options", "Charm selection and configuration options")
+    group.addoption(
+        "--charm",
+        action="store",
+        required=True,
+        help="Name of the charm to test (required)",
+    )
+    group.addoption(
+        "--channel-v0",
+        action="store",
+        required=True,
+        help="Channel for v0 charm (required)",
+    )
+    group.addoption(
+        "--revision-v0",
+        action="store",
+        default=None,
+        help="Revision for v0 charm (default: None)",
+    )
+    group.addoption(
+        "--channel-v1",
+        action="store",
+        required=True,
+        help="Channel for v1 charm (required)",
+    )
+    group.addoption(
+        "--revision-v1",
+        action="store",
+        default=None,
+        help="Revision for v1 charm (default: None)",
+    )
+    group.addoption(
+        "--trust",
+        action="store_true",
+        default=False,
+        help="Whether to use --trust when deploying charms (default: False)",
+    )
+    parser.addoption(
+        "--model",
+        action="store",
+        default=None,
+        help="Specify the model to use for testing",
+    )
+
+
+@pytest.fixture(scope="function")
+def juju(request: pytest.FixtureRequest):
+    keep_models = bool(request.config.getoption("--keep-models"))
+    model_name = request.config.getoption("--model")
+    if model_name is None:
+        with jubilant.temp_model(keep=keep_models) as juju:
+            juju.wait_timeout = 10 * 60
+            yield juju  # run the test
+            if request.session.testsfailed:
+                log = juju.debug_log(limit=30)
+                print(log, end="")
+    else:
+        juju = jubilant.Juju()
+        juju.set_model(str(model_name))
+        juju.wait_timeout = 10 * 60
+        yield juju
+        if request.session.testsfailed:
+            log = juju.debug_log(limit=30)
+            print(log, end="")
+
+
+@pytest.fixture(scope="function")
+def bucket_name() -> str:
+    return f"s3-integrator-{''.join(random.sample(string.ascii_lowercase, 6))}"
+
+
+@pytest.fixture
+def provider_charm_v1(
+    s3_charm: Path, s3_root_user: S3ConnectionInfo, bucket_name: str, platform: str
+) -> CharmSpec:
+    return CharmSpec(
+        charm=s3_charm,
+        app="s3-integrator-v1",
+        config={
+            "bucket": bucket_name,
+            "endpoint": s3_root_user.endpoint,
+            "region": s3_root_user.region,
+            "tls-ca-chain": s3_root_user.tls_ca_chain,
+            "s3-uri-style": "path",
+            "path": "custompath/",
+        },
+        secret_config={
+            "credentials": {
+                "access-key": s3_root_user.access_key,
+                "secret-key": s3_root_user.secret_key,
+            },
+        },
+        constraints={
+            "arch": platform,
+        },
+    )
+
+
+@pytest.fixture
+def provider_charm_v0(
+    request: pytest.FixtureRequest, s3_root_user: S3ConnectionInfo, bucket_name: str, platform: str
+) -> CharmSpec:
+    return CharmSpec(
+        charm="s3-integrator",
+        app="s3-integrator-v0",
+        channel="1/stable",
+        config={
+            "bucket": bucket_name,
+            "endpoint": s3_root_user.endpoint,
+            "region": s3_root_user.region,
+            "tls-ca-chain": s3_root_user.tls_ca_chain,
+            "s3-uri-style": "path",
+            "path": "custompath/",
+        },
+        action_config={
+            "sync-s3-credentials": {
+                "access-key": s3_root_user.access_key,
+                "secret-key": s3_root_user.secret_key,
+            },
+        },
+        constraints={
+            "arch": platform,
+        },
+    )
+
+
+@pytest.fixture
+def requirer_charm_v0(request: pytest.FixtureRequest, platform: str) -> CharmSpec:
+    channel = request.config.getoption("--channel-v0")
+    revision = request.config.getoption("--revision-v0")
+    trust = request.config.getoption("--trust")
+    charm = request.config.getoption("--charm")
+    return CharmSpec(
+        charm=charm,
+        app=f"{charm}-v0",
+        channel=channel,
+        trust=trust,
+        revision=int(revision) if revision else None,
+        constraints={
+            "arch": platform,
+        },
+    )
+
+
+@pytest.fixture
+def requirer_charm_v1(request: pytest.FixtureRequest, platform: str) -> CharmSpec:
+    charm = request.config.getoption("--charm")
+    channel = request.config.getoption("--channel-v1")
+    revision = request.config.getoption("--revision-v1")
+    trust = request.config.getoption("--trust")
+    return CharmSpec(
+        charm=charm,
+        app=f"{charm}-v1",
+        channel=channel,
+        trust=trust,
+        revision=int(revision) if revision else None,
+        constraints={
+            "arch": platform,
+        },
+    )

s3/tests/integration/test_consumers/helpers.py

@@ -0,0 +1,114 @@
+import dataclasses
+import re
+
+import jubilant
+
+
+@dataclasses.dataclass
+class CharmSpec:
+    charm: str
+    app: str
+    channel: str | None = None
+    trust: bool = False
+    num_units: int = 1
+    revision: int | None = None
+    config: dict = dataclasses.field(default_factory=dict)
+    secret_config: dict[str, dict[str, str]] = dataclasses.field(default_factory=dict)
+    action_config: dict[str, dict[str, str]] = dataclasses.field(default_factory=dict)
+    constraints: dict[str, str] = dataclasses.field(default_factory=dict)
+
+    def __str__(self):
+        """Provide a human-readable string representation."""
+        return f"{self.charm}-{self.channel.replace('/', '-')}"
+
+    def __repr__(self):
+        """Provide a concise representation for debugging."""
+        return f"CharmSpec(charm={self.charm}, channel={self.channel}, revision={self.revision})"
+
+
+def wait_active_idle(juju: jubilant.Juju, delay: int = 5):
+    """Wait for all applications to be active and all agents to be idle."""
+    juju.wait(
+        lambda status: jubilant.all_active(status) and jubilant.all_agents_idle(status),
+        delay=delay,
+    )
+
+
+def deploy_and_configure_charm(juju: jubilant.Juju, charm: CharmSpec):
+    """Deploy a charm from given spec."""
+    juju.deploy(
+        charm=charm.charm,
+        app=charm.app,
+        channel=charm.channel,
+        revision=charm.revision,
+        trust=charm.trust,
+        num_units=charm.num_units,
+        constraints=charm.constraints,
+    )
+    juju.wait(jubilant.all_agents_idle, delay=5)
+    charm_config = charm.config
+    for config_name, secret_content in charm.secret_config.items():
+        secret_uri = juju.add_secret(
+            name=f"{charm.app}-{config_name}",
+            content=secret_content,
+        )
+        juju.grant_secret(identifier=secret_uri, app=charm.app)
+        charm_config[config_name] = secret_uri
+    if charm_config:
+        juju.config(app=charm.app, values=charm_config)
+    for action_name, params in charm.action_config.items():
+        juju.wait(jubilant.all_agents_idle, delay=5)
+        juju.run(f"{charm.app}/0", action_name, params)
+    wait_active_idle(juju)
+
+
+def integrate_charms(juju: jubilant.Juju, provider: CharmSpec, requirer: CharmSpec):
+    """Integrate provider and requirer charms."""
+    juju.integrate(f"{provider.app}:s3-credentials", requirer.app)
+    wait_active_idle(juju, delay=15)
+
+
+def remove_charm_relations(juju: jubilant.Juju, provider: CharmSpec, requirer: CharmSpec):
+    """Remove the relation between provider and requirer charms."""
+    juju.remove_relation(f"{provider.app}:s3-credentials", requirer.app)
+    wait_active_idle(juju, delay=15)
+
+
+def list_backups(juju: jubilant.Juju, database: CharmSpec) -> list[str]:
+    """List backups using the requirer charm's action."""
+    action = juju.run(f"{database.app}/0", "list-backups")
+    assert action.return_code == 0, f"list-backups action failed: {action.stderr}"
+    lines = action.results["backups"].splitlines()
+    idx = next(i for i, line in enumerate(lines) if re.match(r"^-{5,}$", line.strip()))
+    backup_lines = [line for line in lines[idx + 1 :] if line.strip()]
+    backup_ids = [line.split()[0] for line in backup_lines if len(line.split()) > 0]
+    return backup_ids
+
+
+def create_backup(juju: jubilant.Juju, database: CharmSpec) -> str:
+    """Create a backup using the requirer charm's action and return the backup ID."""
+    action = juju.run(f"{database.app}/0", "create-backup")
+    assert action.return_code == 0, f"create-backup action failed: {action.stderr}"
+    print(action.results)
+    # assert action.results["backup-status"] == "backup created", f"Unexpected backup status: {action.results['backup-status']}"
+    wait_active_idle(juju)
+
+
+def restore_backup(juju: jubilant.Juju, database: CharmSpec, backup_id: str):
+    """Restore a backup using the requirer charm's action."""
+    action = juju.run(f"{database.app}/0", "restore", {"backup-id": backup_id})
+    assert action.return_code == 0, f"restore-backup action failed: {action.stderr}"
+    print(action.results)
+    # assert action.results["restore-status"] == "restore started", f"Unexpected restore status: {action.results['restore-status']}"
+    wait_active_idle(juju)
+
+
+def upgrade_charm(juju: jubilant.Juju, old_charm: CharmSpec, new_charm: CharmSpec):
+    """Upgrade a charm from old spec to new spec."""
+    juju.refresh(
+        app=old_charm.app,
+        channel=new_charm.channel,
+        revision=new_charm.revision,
+        trust=new_charm.trust,
+    )
+    wait_active_idle(juju, delay=15)