follow ck best practices for integration tests (overlay k8s-core with…

… locally built charm/resources)

tests/data/charm.yaml

@@ -0,0 +1,26 @@
+description: A minimal Kubernetes cluster with two machines with virtual networks provided by Canal.
+series: &series {{ series }}
+applications:
+  calico: null
+  kubernetes-control-plane:
+    options:
+      channel: {{ snap_channel }}
+  kubernetes-worker:
+    options:
+      channel: {{ snap_channel }}
+  canal:
+    charm: {{ charm }}
+    channel: null
+    resources:
+      flannel: {{flannel_amd64|default("0")}}
+      flannel-arm64: {{flannel_arm64|default("0")}}
+      calico: {{calico_amd64|default("0")}}
+      calico-arm64: {{calico_arm64|default("0")}}
+      calico-node-image: {{calico_node_image|default("0")}}
+relations:
+- - canal:etcd
+  - etcd:db
+- - canal:cni
+  - kubernetes-control-plane:cni
+- - canal:cni
+  - kubernetes-worker:cni

tests/integration/conftest.py

@@ -1,56 +1,29 @@
 import logging
-from pathlib import Path
 import pytest
-import shlex
 
 log = logging.getLogger(__name__)
 
 
-def ensure_arch_names(current_resources):
-    for resource in current_resources:
-        if any(arch in resource.name for arch in ['s390x', 'arm64', 'amd64']):
-            continue
-        if resource.name == 'calico-node-image.tar.gz':
-            # doesn't have an associated arch
-            continue
-        # without an arch, assume 'amd64'
-        head, tail = resource.name.split('.', 1)
-        arched_name = "{}-amd64.{}".format(head, tail)
-        (resource.parent / arched_name).symlink_to(resource)
+def pytest_addoption(parser):
+    parser.addoption(
+        "--series",
+        type=str,
+        default="focal",
+        help="Set series for the machine units",
+    )
+    parser.addoption(
+        "--snap-channel",
+        type=str,
+        default="1.24/stable",
+        help="Set snap channel for the control-plane & worker units",
+    )
 
 
 @pytest.fixture()
-async def setup_resources(ops_test):
-    """Provides the flannel resources needed to deploy the charm."""
-    script_path = resource_path = Path.cwd()
-    current_resources = list(resource_path.glob("*.tar.gz"))
-    tmpdir = ops_test.tmp_path / "resources"
-    tmpdir.mkdir(parents=True, exist_ok=True)
-    if not current_resources:
-        # If they are not locally available, try to build them
-        log.info("Build Resources...")
-        build_script = script_path / "build-canal-resources.sh"
-        rc, stdout, stderr = await ops_test.run(
-            *shlex.split("sudo {}".format(build_script)), cwd=tmpdir, check=False
-        )
-        if rc != 0:
-            err = (stderr or stdout).strip()
-            log.warning("build-flannel-resources failed: {}".format(err))
-        current_resources = list(Path(tmpdir).glob("*.tar.gz"))
-        resource_path = tmpdir
-    if not current_resources:
-        # if we couldn't build them, just download a fixed version
-        log.info("Downloading Resources...")
-        fetch_script = script_path / "fetch-charm-store-resources.sh"
-        rc, stdout, stderr = await ops_test.run(
-            *shlex.split(str(fetch_script)), cwd=tmpdir, check=False
-        )
-        if rc != 0:
-            err = (stderr or stdout).strip()
-            log.warning("fetch-charm-store-resources failed: {}".format(err))
-        current_resources = list(Path(tmpdir).glob("*.tar.gz"))
-        resource_path = tmpdir
-    if not current_resources:
-        pytest.fail("Could not prepare necessary resources for testing charm")
-    ensure_arch_names(current_resources)
-    yield resource_path
+def series(request):
+    return request.config.getoption("--series")
+
+
+@pytest.fixture()
+def snap_channel(request):
+    return request.config.getoption("--snap-channel")

tests/integration/test_canal_integration.py

@@ -0,0 +1,64 @@
+import logging
+import shlex
+from pathlib import Path
+
+import pytest
+
+log = logging.getLogger(__name__)
+
+
+def remove_ext(path: Path) -> str:
+    suffixes = "".join(path.suffixes)
+    return path.name.replace(suffixes, "")
+
+
+@pytest.mark.abort_on_fail
+async def test_build_and_deploy(ops_test, series: str, snap_channel: str):
+    """Build and deploy Canal in bundle."""
+    charm = next(Path.cwd().glob("canal*.charm"), None)
+    if not charm:
+        log.info("Build Charm...")
+        charm = await ops_test.build_charm(".")
+
+    resources = list(Path.cwd().glob("flannel*.tar.gz")) + list(
+        Path.cwd().glob("calico*.tar.gz")
+    )
+    if not resources:
+        log.info("Build Resources...")
+        build_script = Path.cwd() / "build-canal-resources.sh"
+        resources = await ops_test.build_resources(build_script, with_sudo=False)
+    expected_resources = {
+        "flannel-amd64",
+        "flannel-arm64",
+        "calico-amd64",
+        "calico-arm64",
+        "calico-node-image",
+    }
+
+    if resources and all(remove_ext(rsc) in expected_resources for rsc in resources):
+        resources = {remove_ext(rsc).replace("-", "_"): rsc for rsc in resources}
+    else:
+        log.info("Failed to build resources, downloading from latest/edge")
+        arch_resources = ops_test.arch_specific_resources(charm)
+        resources = await ops_test.download_resources(charm, resources=arch_resources)
+        resources = {name.replace("-", "_"): rsc for name, rsc in resources.items()}
+
+    assert resources, "Failed to build or download charm resources."
+
+    log.info("Build Bundle...")
+    context = dict(charm=charm, series=series, snap_channel=snap_channel, **resources)
+    overlays = [
+        ops_test.Bundle("kubernetes-core", channel="edge"),
+        Path("tests/data/charm.yaml"),
+    ]
+    bundle, *overlays = await ops_test.async_render_bundles(*overlays, **context)
+
+    log.info("Deploy Bundle...")
+    model = ops_test.model_full_name
+    cmd = f"juju deploy -m {model} {bundle} " + " ".join(
+        f"--overlay={f}" for f in overlays
+    )
+    rc, stdout, stderr = await ops_test.run(*shlex.split(cmd))
+    assert rc == 0, "Bundle deploy failed: {}".format((stderr or stdout).strip())
+
+    await ops_test.model.wait_for_idle(status="active", timeout=60 * 60, idle_period=60)