Merge pull request #24 from canonical/IAM-667

test: add integration testcases

tests/integration/conftest.py

@@ -0,0 +1,86 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+import functools
+import subprocess
+from pathlib import Path
+from typing import Callable, Optional
+
+import pytest
+import yaml
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from pytest_operator.plugin import OpsTest
+
+METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
+CERTIFICATE_PROVIDER_APP = "self-signed-certificates"
+DB_APP = "postgresql-k8s"
+GLAUTH_APP = METADATA["name"]
+GLAUTH_IMAGE = METADATA["resources"]["oci-image"]["upstream-source"]
+GLAUTH_CLIENT_APP = "any-charm"
+
+
+def extract_certificate_common_name(certificate: str) -> Optional[str]:
+    cert_data = certificate.encode()
+    cert = x509.load_pem_x509_certificate(cert_data, default_backend())
+    if not (rdns := cert.subject.rdns):
+        return None
+
+    return rdns[0].rfc4514_string()
+
+
+def get_unit_data(unit_name: str, model_name: str) -> dict:
+    res = subprocess.run(
+        ["juju", "show-unit", unit_name, "-m", model_name],
+        check=True,
+        text=True,
+        capture_output=True,
+    )
+    cmd_output = yaml.safe_load(res.stdout)
+    return cmd_output[unit_name]
+
+
+def get_integration_data(model_name: str, app_name: str, integration_name: str) -> Optional[dict]:
+    unit_data = get_unit_data(f"{app_name}/0", model_name)
+    return next(
+        (
+            integration
+            for integration in unit_data["relation-info"]
+            if integration["endpoint"] == integration_name
+        ),
+        None,
+    )
+
+
+def get_app_integration_data(
+    model_name: str, app_name: str, integration_name: str
+) -> Optional[dict]:
+    data = get_integration_data(model_name, app_name, integration_name)
+    return data["application-data"] if data else None
+
+
+def get_unit_integration_data(
+    model_name: str, app_name: str, remote_app_name: str, integration_name: str
+) -> Optional[dict]:
+    data = get_integration_data(model_name, app_name, integration_name)
+    return data["related-units"][f"{remote_app_name}/0"]["data"] if data else None
+
+
+@pytest.fixture
+def app_integration_data(ops_test: OpsTest) -> Callable:
+    return functools.partial(get_app_integration_data, ops_test.model_name)
+
+
+@pytest.fixture
+def unit_integration_data(ops_test: OpsTest) -> Callable:
+    return functools.partial(get_unit_integration_data, ops_test.model_name)
+
+
+@pytest.fixture
+def database_integration_data(app_integration_data: Callable) -> Optional[dict]:
+    return app_integration_data(GLAUTH_APP, "pg-database")
+
+
+@pytest.fixture
+def certificate_integration_data(app_integration_data: Callable) -> Optional[dict]:
+    return app_integration_data(GLAUTH_APP, "certificates")

tests/integration/test_charm.py

@@ -1,49 +1,142 @@
 #!/usr/bin/env python3
-# Copyright 2023 Canonical Ltd.
+# Copyright 2024 Canonical Ltd.
 # See LICENSE file for licensing details.
 
+import asyncio
+import json
 import logging
 from pathlib import Path
+from typing import Callable, Optional
 
 import pytest
-import yaml
+from conftest import (
+    CERTIFICATE_PROVIDER_APP,
+    DB_APP,
+    GLAUTH_APP,
+    GLAUTH_CLIENT_APP,
+    GLAUTH_IMAGE,
+    extract_certificate_common_name,
+)
 from pytest_operator.plugin import OpsTest
+from tester import ANY_CHARM
 
 logger = logging.getLogger(__name__)
 
-METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
-GLAUTH_APP = METADATA["name"]
-GLAUTH_IMAGE = METADATA["resources"]["oci-image"]["upstream-source"]
-DB_APP = "postgresql-k8s"
-
 
 @pytest.mark.skip_if_deployed
 @pytest.mark.abort_on_fail
 async def test_build_and_deploy(ops_test: OpsTest) -> None:
-    await ops_test.model.deploy(
-        "postgresql-k8s",
-        channel="14/stable",
-        trust=True,
+    charm_lib_path = Path("lib/charms")
+    any_charm_src_overwrite = {
+        "any_charm.py": ANY_CHARM,
+        "ldap.py": (charm_lib_path / "glauth_k8s/v0/ldap.py").read_text(),
+        "certificate_transfer.py": (
+            charm_lib_path / "certificate_transfer_interface/v0/certificate_transfer.py"
+        ).read_text(),
+    }
+
+    await asyncio.gather(
+        ops_test.model.deploy(
+            DB_APP,
+            channel="14/stable",
+            trust=True,
+        ),
+        ops_test.model.deploy(
+            CERTIFICATE_PROVIDER_APP,
+            channel="stable",
+            trust=True,
+        ),
+        ops_test.model.deploy(
+            GLAUTH_CLIENT_APP,
+            channel="beta",
+            config={
+                "src-overwrite": json.dumps(any_charm_src_overwrite),
+                "python-packages": "pydantic ~= 2.0\njsonschema",
+            },
+        ),
     )
+
     charm_path = await ops_test.build_charm(".")
     await ops_test.model.deploy(
         str(charm_path),
         resources={"oci-image": GLAUTH_IMAGE},
         application_name=GLAUTH_APP,
-        config={"starttls_enabled": False},
+        config={"starttls_enabled": True},
         trust=True,
         series="jammy",
     )
+
+    await ops_test.model.integrate(GLAUTH_APP, CERTIFICATE_PROVIDER_APP)
     await ops_test.model.integrate(GLAUTH_APP, DB_APP)
 
     await ops_test.model.wait_for_idle(
-        apps=[GLAUTH_APP, DB_APP],
+        apps=[CERTIFICATE_PROVIDER_APP, DB_APP, GLAUTH_CLIENT_APP, GLAUTH_APP],
         status="active",
         raise_on_blocked=False,
         timeout=1000,
     )
 
 
+def test_database_integration(
+    ops_test: OpsTest,
+    database_integration_data: Optional[dict],
+) -> None:
+    assert database_integration_data
+    assert f"{ops_test.model_name}_{GLAUTH_APP}" == database_integration_data["database"]
+    assert database_integration_data["password"]
+
+
+def test_certification_integration(
+    certificate_integration_data: Optional[dict],
+) -> None:
+    assert certificate_integration_data
+    certificates = json.loads(certificate_integration_data["certificates"])
+    certificate = certificates[0]["certificate"]
+    assert "CN=ldap.glauth.com" == extract_certificate_common_name(certificate)
+
+
+async def test_ldap_integration(
+    ops_test: OpsTest,
+    app_integration_data: Callable,
+) -> None:
+    await ops_test.model.integrate(
+        f"{GLAUTH_CLIENT_APP}:ldap",
+        f"{GLAUTH_APP}:ldap",
+    )
+
+    await ops_test.model.wait_for_idle(
+        apps=[GLAUTH_APP, GLAUTH_CLIENT_APP],
+        status="active",
+        timeout=1000,
+    )
+
+    ldap_integration_data = app_integration_data(
+        GLAUTH_CLIENT_APP,
+        "ldap",
+    )
+    assert ldap_integration_data
+    assert ldap_integration_data["bind_dn"].startswith(
+        f"cn={GLAUTH_CLIENT_APP},ou={ops_test.model_name}"
+    )
+
+
+async def test_certificate_transfer_integration(
+    ops_test: OpsTest,
+    unit_integration_data: Callable,
+) -> None:
+    await ops_test.model.integrate(
+        f"{GLAUTH_CLIENT_APP}:send-ca-cert",
+        f"{GLAUTH_APP}:send-ca-cert",
+    )
+
+    certificate_transfer_integration_data = unit_integration_data(
+        GLAUTH_CLIENT_APP,
+        GLAUTH_APP,
+        "send-ca-cert",
+    )
+    assert certificate_transfer_integration_data
+
+
 async def test_glauth_scale_up(ops_test: OpsTest) -> None:
     app, target_unit_num = ops_test.model.applications[GLAUTH_APP], 3
 
@@ -52,8 +145,7 @@ async def test_glauth_scale_up(ops_test: OpsTest) -> None:
     await ops_test.model.wait_for_idle(
         apps=[GLAUTH_APP],
         status="active",
-        raise_on_blocked=True,
-        timeout=600,
+        timeout=1000,
         wait_for_exact_units=target_unit_num,
     )
 
@@ -65,5 +157,5 @@ async def test_glauth_scale_down(ops_test: OpsTest) -> None:
     await ops_test.model.wait_for_idle(
         apps=[GLAUTH_APP],
         status="active",
-        timeout=300,
+        timeout=1000,
     )

tests/integration/tester.py

@@ -0,0 +1,46 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+import textwrap
+
+ANY_CHARM = textwrap.dedent(
+    """
+from typing import Any
+
+from any_charm_base import AnyCharmBase
+from certificate_transfer import CertificateAvailableEvent, CertificateTransferRequires
+from ldap import (
+    LdapReadyEvent,
+    LdapRequirer,
+)
+
+
+class AnyCharm(AnyCharmBase):
+    def __init__(self, *args: Any):
+        super().__init__(*args)
+        self.ldap_requirer = LdapRequirer(
+            self,
+            relation_name="ldap",
+        )
+        self.framework.observe(
+            self.ldap_requirer.on.ldap_ready,
+            self._on_ldap_ready,
+        )
+        self.certificate_transfer = CertificateTransferRequires(
+            self,
+            relationship_name="send-ca-cert",
+        )
+        self.framework.observe(
+            self.certificate_transfer.on.certificate_available,
+            self._on_certificate_available,
+        )
+
+    def _on_ldap_ready(self, event: LdapReadyEvent) -> None:
+        ldap_data = self.ldap_requirer.consume_ldap_relation_data(
+            event.relation.id,
+        )
+
+    def _on_certificate_available(self, event: CertificateAvailableEvent) -> None:
+        pass
+"""
+)