add integration test

.github/workflows/integration_test.yaml

@@ -0,0 +1,28 @@
+name: Integration tests
+
+on:
+  pull_request:
+
+jobs:
+  integration-test-microk8s:
+    name: Integration tests (microk8s)
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+      - name: Setup operator environment
+        uses: charmed-kubernetes/actions-operator@main
+        with:
+          juju-channel: 3.1/stable
+          provider: microk8s
+          microk8s-addons: "ingress storage dns rbac registry"
+          channel: 1.25-strict/stable
+      - name: Run integration tests
+        # set a predictable model name so it can be consumed by charm-logdump-action
+        run: tox -e integration -- --model testing
+      - name: Dump logs
+        uses: canonical/charm-logdump-action@main
+        if: failure()
+        with:
+          app: airbyte-ui-k8s
+          model: testing

tests/integration/conftest.py

@@ -0,0 +1,56 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Charm integration test config."""
+
+import asyncio
+import logging
+
+import pytest_asyncio
+from helpers import (
+    APP_NAME_AIRBYTE_SERVER,
+    APP_NAME_TEMPORAL_ADMIN,
+    APP_NAME_TEMPORAL_SERVER,
+    create_default_namespace,
+    get_airbyte_charm_resources,
+    perform_airbyte_integrations,
+    perform_temporal_integrations,
+)
+from pytest_operator.plugin import OpsTest
+
+logger = logging.getLogger(__name__)
+
+
+@pytest_asyncio.fixture(name="deploy", scope="module")
+async def deploy(ops_test: OpsTest):
+    """Test the app is up and running."""
+    charm = await ops_test.build_charm(".")
+    resources = get_airbyte_charm_resources()
+
+    asyncio.gather(
+        ops_test.model.deploy(charm, resources=resources, application_name=APP_NAME_AIRBYTE_SERVER, trust=True),
+        ops_test.model.deploy(
+            APP_NAME_TEMPORAL_SERVER,
+            channel="edge",
+            config={"num-history-shards": 1},
+        ),
+        ops_test.model.deploy(APP_NAME_TEMPORAL_ADMIN, channel="edge"),
+        ops_test.model.deploy("postgresql-k8s", channel="14/stable", trust=True),
+        ops_test.model.deploy("minio", channel="edge"),
+    )
+
+    async with ops_test.fast_forward():
+        await ops_test.model.wait_for_idle(
+            apps=["postgresql-k8s", "minio"], status="active", raise_on_blocked=False, timeout=1200
+        )
+        await ops_test.model.wait_for_idle(
+            apps=[APP_NAME_TEMPORAL_SERVER, APP_NAME_TEMPORAL_ADMIN],
+            status="blocked",
+            raise_on_blocked=False,
+            timeout=600,
+        )
+
+        await perform_temporal_integrations(ops_test)
+        await create_default_namespace(ops_test)
+
+        await perform_airbyte_integrations(ops_test)

tests/integration/helpers.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Temporal charm integration test helpers."""
+
+import logging
+from pathlib import Path
+
+import yaml
+from pytest_operator.plugin import OpsTest
+from temporal_client.activities import say_hello
+from temporal_client.workflows import SayHello
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+logger = logging.getLogger(__name__)
+
+METADATA = yaml.safe_load(Path("./charmcraft.yaml").read_text())
+APP_NAME_AIRBYTE_SERVER = METADATA["name"]
+APP_NAME_TEMPORAL_SERVER = "temporal-k8s"
+APP_NAME_TEMPORAL_ADMIN = "temporal-admin-k8s"
+APP_NAME_TEMPORAL_UI = "temporal-ui-k8s"
+
+
+def get_airbyte_charm_resources():
+    return {
+        "airbyte-api-server": METADATA["resources"]["airbyte-api-server"]["upstream-source"],
+        "airbyte-bootloader": METADATA["resources"]["airbyte-bootloader"]["upstream-source"],
+        "airbyte-connector-builder-server": METADATA["resources"]["airbyte-connector-builder-server"][
+            "upstream-source"
+        ],
+        "airbyte-cron": METADATA["resources"]["airbyte-cron"]["upstream-source"],
+        "airbyte-pod-sweeper": METADATA["resources"]["airbyte-pod-sweeper"]["upstream-source"],
+        "airbyte-server": METADATA["resources"]["airbyte-server"]["upstream-source"],
+        "airbyte-workers": METADATA["resources"]["airbyte-workers"]["upstream-source"],
+    }
+
+
+async def run_sample_workflow(ops_test: OpsTest):
+    """Connect a client and runs a basic Temporal workflow.
+
+    Args:
+        ops_test: PyTest object.
+    """
+    url = await get_application_url(ops_test, application=APP_NAME_TEMPORAL_SERVER, port=7233)
+    logger.info("running workflow on app address: %s", url)
+
+    client = await Client.connect(url)
+
+    # Run a worker for the workflow
+    async with Worker(client, task_queue="my-task-queue", workflows=[SayHello], activities=[say_hello]):
+        name = "Jean-luc"
+        result = await client.execute_workflow(SayHello.run, name, id="my-workflow-id", task_queue="my-task-queue")
+        logger.info(f"result: {result}")
+        assert result == f"Hello, {name}!"
+
+
+async def create_default_namespace(ops_test: OpsTest):
+    """Create default namespace on Temporal server using tctl.
+
+    Args:
+        ops_test: PyTest object.
+    """
+    # Register default namespace from admin charm.
+    action = (
+        await ops_test.model.applications[APP_NAME_TEMPORAL_ADMIN]
+        .units[0]
+        .run_action("tctl", args="--ns default namespace register -rd 3")
+    )
+    result = (await action.wait()).results
+    logger.info(f"tctl result: {result}")
+    assert "result" in result and result["result"] == "command succeeded"
+
+
+async def get_application_url(ops_test: OpsTest, application, port):
+    """Return application URL from the model.
+
+    Args:
+        ops_test: PyTest object.
+        application: Name of the application.
+        port: Port number of the URL.
+
+    Returns:
+        Application URL of the form {address}:{port}
+    """
+    status = await ops_test.model.get_status()  # noqa: F821
+    address = status["applications"][application].public_address
+    return f"{address}:{port}"
+
+
+async def get_unit_url(ops_test: OpsTest, application, unit, port, protocol="http"):
+    """Return unit URL from the model.
+
+    Args:
+        ops_test: PyTest object.
+        application: Name of the application.
+        unit: Number of the unit.
+        port: Port number of the URL.
+        protocol: Transfer protocol (default: http).
+
+    Returns:
+        Unit URL of the form {protocol}://{address}:{port}
+    """
+    status = await ops_test.model.get_status()  # noqa: F821
+    address = status["applications"][application]["units"][f"{application}/{unit}"]["address"]
+    return f"{protocol}://{address}:{port}"
+
+
+async def perform_temporal_integrations(ops_test: OpsTest):
+    """Integrate Temporal charm with postgresql, admin and ui charms.
+
+    Args:
+        ops_test: PyTest object.
+    """
+    await ops_test.model.integrate(f"{APP_NAME_TEMPORAL_SERVER}:db", "postgresql-k8s:database")
+    await ops_test.model.integrate(f"{APP_NAME_TEMPORAL_SERVER}:visibility", "postgresql-k8s:database")
+    await ops_test.model.integrate(f"{APP_NAME_TEMPORAL_SERVER}:admin", f"{APP_NAME_TEMPORAL_ADMIN}:admin")
+    await ops_test.model.wait_for_idle(
+        apps=[APP_NAME_TEMPORAL_SERVER], status="active", raise_on_blocked=False, timeout=180
+    )
+
+    assert ops_test.model.applications[APP_NAME_TEMPORAL_SERVER].units[0].workload_status == "active"
+
+
+async def perform_airbyte_integrations(ops_test: OpsTest):
+    """Perform Airbyte charm integrations.
+
+    Args:
+        ops_test: PyTest object.
+    """
+    await ops_test.model.integrate(APP_NAME_AIRBYTE_SERVER, "postgresql-k8s")
+    await ops_test.model.integrate(APP_NAME_AIRBYTE_SERVER, "minio")
+    await ops_test.model.wait_for_idle(
+        apps=[APP_NAME_AIRBYTE_SERVER], status="active", raise_on_blocked=False, timeout=600
+    )
+
+    assert ops_test.model.applications[APP_NAME_AIRBYTE_SERVER].units[0].workload_status == "active"

tests/integration/temporal_client/activities.py

@@ -0,0 +1,20 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+
+"""Temporal client activity."""
+
+from temporalio import activity
+
+
+@activity.defn
+async def say_hello(name: str) -> str:
+    """Temporal activity.
+
+    Args:
+        name: used to run the dynamic activity.
+
+    Returns:
+        String in the form "Hello, {name}!
+    """
+    return f"Hello, {name}!"

tests/integration/temporal_client/workflows.py

@@ -0,0 +1,32 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+
+"""Temporal client sample workflow."""
+
+import asyncio
+from datetime import timedelta
+from typing import List
+
+from temporalio import workflow
+
+# Import our activity, passing it through the sandbox
+with workflow.unsafe.imports_passed_through():
+    from .activities import say_hello
+
+
+@workflow.defn
+class SayHello:
+    """Temporal workflow class."""
+
+    @workflow.run
+    async def run(self, name: str) -> str:
+        """Workflow execution method.
+
+        Args:
+            name: used to run the dynamic activity.
+
+        Returns:
+            Workflow execution
+        """
+        return await workflow.execute_activity(say_hello, name, schedule_to_close_timeout=timedelta(seconds=5))

tests/integration/test_charm.py

@@ -1,32 +1,28 @@
+#!/usr/bin/env python3
 # Copyright 2024 Canonical Ltd.
 # See LICENSE file for licensing details.
 
-import asyncio
 import logging
-from pathlib import Path
 
 import pytest
-import yaml
+import requests
+from conftest import deploy  # noqa: F401, pylint: disable=W0611
+from helpers import APP_NAME_AIRBYTE_SERVER, get_unit_url
 from pytest_operator.plugin import OpsTest
 
 logger = logging.getLogger(__name__)
 
-METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
-APP_NAME = METADATA["name"]
-
 
 @pytest.mark.abort_on_fail
-async def test_build_and_deploy(ops_test: OpsTest):
-    """Build the charm-under-test and deploy it together with related charms.
-
-    Assert on the unit status before any relations/configurations take place.
-    """
-    # Build and deploy charm from local source folder
-    charm = await ops_test.build_charm(".")
-    resources = {"httpbin-image": METADATA["resources"]["httpbin-image"]["upstream-source"]}
-
-    # Deploy the charm and wait for active/idle status
-    await asyncio.gather(
-        ops_test.model.deploy(charm, resources=resources, application_name=APP_NAME),
-        ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", raise_on_blocked=True, timeout=1000),
-    )
+@pytest.mark.usefixtures("deploy")
+class TestDeployment:
+    """Integration tests for charm."""
+
+    async def test_deployment(self, ops_test: OpsTest):
+        url = await get_unit_url(ops_test, application=APP_NAME_AIRBYTE_SERVER, unit=0, port=8001)
+        logger.info("curling app address: %s", url)
+
+        response = requests.get(f"{url}/api/v1/health", timeout=300)
+        print(response.json())
+        assert response.status_code == 200
+        assert response.json().get("available")

tox.ini

@@ -112,4 +112,4 @@ commands =
            --tb native \
            --log-cli-level=INFO \
            {posargs} \
-           {[vars]tests_path}/integration/new_test_charm.py
+           {[vars]tests_path}/integration/test_charm.py