[DPE-5588] Check against invalid arch (#194)

src/architecture.py

@@ -0,0 +1,61 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Architecture utilities module"""
+
+import logging
+import os
+import pathlib
+import platform
+
+import yaml
+from ops.charm import CharmBase
+from ops.model import BlockedStatus
+
+logger = logging.getLogger(__name__)
+
+
+class WrongArchitectureWarningCharm(CharmBase):
+    """A fake charm class that only signals a wrong architecture deploy."""
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        hw_arch = platform.machine()
+        self.unit.status = BlockedStatus(
+            f"Charm incompatible with {hw_arch} architecture. "
+            f"If this app is being refreshed, rollback"
+        )
+        raise RuntimeError(
+            f"Incompatible architecture: this charm revision does not support {hw_arch}. "
+            f"If this app is being refreshed, rollback with instructions from Charmhub docs. "
+            f"If this app is being deployed for the first time, remove it and deploy it again "
+            f"using a compatible revision."
+        )
+
+
+def is_wrong_architecture() -> bool:
+    """Checks if charm was deployed on wrong architecture."""
+    charm_path = os.environ.get("CHARM_DIR", "")
+    manifest_path = pathlib.Path(charm_path, "manifest.yaml")
+
+    if not manifest_path.exists():
+        logger.error("Cannot check architecture: manifest file not found in %s", manifest_path)
+        return False
+
+    manifest = yaml.safe_load(manifest_path.read_text())
+
+    manifest_archs = []
+    for base in manifest["bases"]:
+        base_archs = base.get("architectures", [])
+        manifest_archs.extend(base_archs)
+
+    hardware_arch = platform.machine()
+    if ("amd64" in manifest_archs and hardware_arch == "x86_64") or (
+        "arm64" in manifest_archs and hardware_arch == "aarch64"
+    ):
+        logger.debug("Charm architecture matches")
+        return False
+
+    logger.error("Charm architecture does not match")
+    return True

src/machine_charm.py

@@ -6,11 +6,17 @@
 
 """MySQL Router machine charm"""
 
+import ops
+
+from architecture import WrongArchitectureWarningCharm, is_wrong_architecture
+
+if is_wrong_architecture() and __name__ == "__main__":
+    ops.main.main(WrongArchitectureWarningCharm)
+
 import logging
 import socket
 import typing
 
-import ops
 import tenacity
 from charms.tempo_coordinator_k8s.v0.charm_tracing import trace_charm
 

tests/conftest.py

@@ -2,6 +2,7 @@
 # See LICENSE file for licensing details.
 
 import argparse
+import subprocess
 
 import pytest
 
@@ -32,6 +33,37 @@ def pytest_configure(config):
         config.option.mysql_router_charm_bases_index = 1
 
 
+@pytest.fixture(autouse=True)
+def architecture() -> str:
+    return subprocess.run(
+        ["dpkg", "--print-architecture"],
+        capture_output=True,
+        check=True,
+        encoding="utf-8",
+    ).stdout.strip()
+
+
+@pytest.fixture
+def only_amd64(architecture):
+    """Pretty way to skip ARM tests."""
+    if architecture != "amd64":
+        pytest.skip("Requires amd64 architecture")
+
+
+@pytest.fixture
+def only_arm64(architecture):
+    """Pretty way to skip AMD tests."""
+    if architecture != "arm64":
+        pytest.skip("Requires arm64 architecture")
+
+
+@pytest.fixture
+def only_ubuntu_jammy(mysql_router_charm_series):
+    """Pretty way to skip < Ubuntu 22.04 tests."""
+    if mysql_router_charm_series != "jammy":
+        pytest.skip("Requires Ubuntu Jammy")
+
+
 @pytest.fixture
 def only_with_juju_secrets(juju_has_secrets):
     """Pretty way to skip Juju 3 tests."""

tests/integration/helpers.py

@@ -5,9 +5,11 @@
 import logging
 import subprocess
 import tempfile
-from typing import Dict, List, Optional
+from pathlib import Path
+from typing import Dict, List, Optional, Union
 
 import tenacity
+import yaml
 from juju.model import Model
 from juju.unit import Unit
 from pytest_operator.plugin import OpsTest
@@ -486,3 +488,17 @@ async def get_machine_address(ops_test: OpsTest, unit: Unit) -> str:
             return line.split()[2]
 
     assert False, "Unable to find the unit's machine"
+
+
+async def get_charm(charm_path: Union[str, Path], architecture: str, bases_index: int) -> Path:
+    """Fetches packed charm from CI runner without checking for architecture."""
+    charm_path = Path(charm_path)
+    charmcraft_yaml = yaml.safe_load((charm_path / "charmcraft.yaml").read_text())
+    assert charmcraft_yaml["type"] == "charm"
+
+    base = charmcraft_yaml["bases"][bases_index]
+    build_on = base.get("build-on", [base])[0]
+    version = build_on["channel"]
+    packed_charms = list(charm_path.glob(f"*{version}-{architecture}.charm"))
+
+    return packed_charms[0].resolve(strict=True)

tests/integration/test_architecture.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+import asyncio
+
+import pytest
+from pytest_operator.plugin import OpsTest
+
+from .helpers import get_charm
+
+MYSQL_ROUTER_APP_NAME = "mysql-router"
+MYSQL_TEST_APP_NAME = "mysql-test-app"
+
+
+@pytest.mark.group(1)
+@pytest.mark.usefixtures("only_amd64", "only_ubuntu_jammy")
+async def test_arm_charm_on_amd_host(ops_test: OpsTest, mysql_router_charm_series: str) -> None:
+    """Tries deploying an arm64 charm on amd64 host."""
+    charm = await get_charm(".", "arm64", 2)
+
+    await asyncio.gather(
+        ops_test.model.deploy(
+            charm,
+            application_name=MYSQL_ROUTER_APP_NAME,
+            num_units=0,
+            series=mysql_router_charm_series,
+        ),
+        ops_test.model.deploy(
+            MYSQL_TEST_APP_NAME,
+            application_name=MYSQL_TEST_APP_NAME,
+            num_units=1,
+            channel="latest/edge",
+            series=mysql_router_charm_series,
+        ),
+    )
+
+    await ops_test.model.relate(
+        f"{MYSQL_ROUTER_APP_NAME}:database",
+        f"{MYSQL_TEST_APP_NAME}:database",
+    )
+
+    await ops_test.model.wait_for_idle(
+        apps=[MYSQL_ROUTER_APP_NAME],
+        status="error",
+        raise_on_error=False,
+    )
+
+
+@pytest.mark.group(1)
+@pytest.mark.usefixtures("only_arm64", "only_ubuntu_jammy")
+async def test_amd_charm_on_arm_host(ops_test: OpsTest, mysql_router_charm_series: str) -> None:
+    """Tries deploying an amd64 charm on arm64 host."""
+    charm = await get_charm(".", "amd64", 1)
+
+    await asyncio.gather(
+        ops_test.model.deploy(
+            charm,
+            application_name=MYSQL_ROUTER_APP_NAME,
+            num_units=0,
+            series=mysql_router_charm_series,
+        ),
+        ops_test.model.deploy(
+            MYSQL_TEST_APP_NAME,
+            application_name=MYSQL_TEST_APP_NAME,
+            num_units=1,
+            channel="latest/edge",
+            series=mysql_router_charm_series,
+        ),
+    )
+
+    await ops_test.model.relate(
+        f"{MYSQL_ROUTER_APP_NAME}:database",
+        f"{MYSQL_TEST_APP_NAME}:database",
+    )
+
+    await ops_test.model.wait_for_idle(
+        apps=[MYSQL_ROUTER_APP_NAME],
+        status="error",
+        raise_on_error=False,
+    )

tests/unit/test_architecture.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+from architecture import is_wrong_architecture
+
+TEST_MANIFEST = """
+    bases:
+        - architectures:
+            - {arch}
+          channel: '22.04'
+          name: ubuntu
+"""
+
+
+def test_wrong_architecture_file_not_found(monkeypatch):
+    """Tests if the function returns False when the charm file doesn't exist."""
+    monkeypatch.setattr("os.environ", {"CHARM_DIR": "/tmp"})
+    monkeypatch.setattr("pathlib.Path.exists", lambda *args, **kwargs: False)
+    assert not is_wrong_architecture()
+
+
+def test_wrong_architecture_amd64(monkeypatch):
+    """Tests if the function correctly identifies arch when charm is AMD."""
+    manifest = TEST_MANIFEST.format(arch="amd64")
+    monkeypatch.setattr("os.environ", {"CHARM_DIR": "/tmp"})
+    monkeypatch.setattr("pathlib.Path.exists", lambda *args, **kwargs: True)
+    monkeypatch.setattr("pathlib.Path.read_text", lambda *args, **kwargs: manifest)
+
+    monkeypatch.setattr("platform.machine", lambda *args, **kwargs: "x86_64")
+    assert not is_wrong_architecture()
+
+    monkeypatch.setattr("platform.machine", lambda *args, **kwargs: "aarch64")
+    assert is_wrong_architecture()
+
+
+def test_wrong_architecture_arm64(monkeypatch):
+    """Tests if the function correctly identifies arch when charm is ARM."""
+    manifest = TEST_MANIFEST.format(arch="arm64")
+    monkeypatch.setattr("os.environ", {"CHARM_DIR": "/tmp"})
+    monkeypatch.setattr("pathlib.Path.exists", lambda *args, **kwargs: True)
+    monkeypatch.setattr("pathlib.Path.read_text", lambda *args, **kwargs: manifest)
+
+    monkeypatch.setattr("platform.machine", lambda *args, **kwargs: "x86_64")
+    assert is_wrong_architecture()
+
+    monkeypatch.setattr("platform.machine", lambda *args, **kwargs: "aarch64")
+    assert not is_wrong_architecture()