From 2f4a8d85c5a93d0104944e965ce8ef9ff2797c0c Mon Sep 17 00:00:00 2001 From: dushu Date: Sun, 11 Feb 2024 16:33:57 -0500 Subject: [PATCH] test: add integration testcases --- tests/integration/conftest.py | 86 ++++++++++++++++++++++ tests/integration/test_charm.py | 124 +++++++++++++++++++++++++++----- tests/integration/tester.py | 46 ++++++++++++ 3 files changed, 240 insertions(+), 16 deletions(-) create mode 100644 tests/integration/conftest.py create mode 100644 tests/integration/tester.py diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 00000000..7e09d696 --- /dev/null +++ b/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") diff --git a/tests/integration/test_charm.py b/tests/integration/test_charm.py index 9c84ac0f..9ddf4fd4 100644 --- a/tests/integration/test_charm.py +++ b/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, ) diff --git a/tests/integration/tester.py b/tests/integration/tester.py new file mode 100644 index 00000000..5134112a --- /dev/null +++ b/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 +""" +)