From 5a9fa05d5006532c471a1de48497bd12c64e9c1d Mon Sep 17 00:00:00 2001 From: Marcelo Henrique Neppel Date: Wed, 1 Jul 2026 15:49:17 -0300 Subject: [PATCH] test(tls): unit tests for the operator-cert manager and events handler Cover TLSManager live-fetch getters, CA rotation, push, and the TLS events handler (certificate_available / relation_broken / defer paths). Stacked on the manager+events PR whose code these exercise. Signed-off-by: Marcelo Henrique Neppel --- tests/unit/test_tls_events.py | 295 ++++++++++++++++++++++++++++++++ tests/unit/test_tls_manager.py | 299 +++++++++++++++++++++++++++++++++ 2 files changed, 594 insertions(+) create mode 100644 tests/unit/test_tls_events.py create mode 100644 tests/unit/test_tls_manager.py diff --git a/tests/unit/test_tls_events.py b/tests/unit/test_tls_events.py new file mode 100644 index 0000000..c04742c --- /dev/null +++ b/tests/unit/test_tls_events.py @@ -0,0 +1,295 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. +"""Tests for the TLS events handler (single_kernel_postgresql/events/tls.py). + +The handler is live-fetch: operator cert/key are read from the requirer on demand +(get_assigned_certificates), never persisted. Only the peer CA is tracked in state +(current-ca / old-ca) for rotation, mirroring postgresql-operator/src/relations/tls.py +_on_peer_certificate_available (which stashes current-ca/old-ca and otherwise reads live). +""" + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from single_kernel_postgresql.config.exceptions import PostgreSQLFileOperationError + + +class _FakePrivateKey: + """Minimal stand-in for PrivateKey: str(key) returns the raw PEM string.""" + + def __init__(self, raw: str) -> None: + self._raw = raw + + def __str__(self) -> str: + return self._raw + + +def _fake_assigned(cert, ca, key): + """Mimic TLSCertificatesRequiresV4.get_assigned_certificates() return shape. + + Returns (list[ProviderCertificate], PrivateKey | None). + """ + provider_cert = SimpleNamespace(certificate=cert, ca=ca) + return [provider_cert], _FakePrivateKey(key) + + +def test_handler_is_wired(harness): + tls = harness.charm.tls + assert tls.client_certificate is not None + assert tls.peer_certificate is not None + # the handler wires its requirers onto the manager so the live-fetch getters work + assert harness.charm.tls_manager.client_certificate is tls.client_certificate + assert harness.charm.tls_manager.peer_certificate is tls.peer_certificate + + +def test_client_certificate_available_pushes(harness): + charm = harness.charm + with ( + patch.object(charm.cluster_manager, "configure_system_passwords"), + patch.object(charm.config_manager, "update_config"), + ): + harness.set_leader(True) + + tls = charm.tls + tls.client_certificate.get_assigned_certificates = MagicMock( + return_value=_fake_assigned("CC", "CA", "CK") + ) + charm.tls_manager.push_tls_files = MagicMock() + + tls._on_certificate_available(MagicMock()) + + # no operator client cert/key is persisted; the push reads live + charm.tls_manager.push_tls_files.assert_called_once() + assert charm.tls_manager.get_client_tls_files() == ("CK", "CA", "CC") + + +def test_peer_certificate_available_rotates_ca_and_pushes(harness): + charm = harness.charm + with ( + patch.object(charm.cluster_manager, "configure_system_passwords"), + patch.object(charm.config_manager, "update_config"), + ): + harness.set_leader(True) + + tls = charm.tls + tls.peer_certificate.get_assigned_certificates = MagicMock( + return_value=_fake_assigned("PC", "PCA", "PK") + ) + charm.tls_manager.push_tls_files = MagicMock() + + tls._on_peer_certificate_available(MagicMock()) + + peer = charm.state.peer + # only the CA is tracked in state for rotation; cert/key stay live + assert peer.current_ca == "PCA" + charm.tls_manager.push_tls_files.assert_called_once() + key, _, cert = charm.tls_manager.get_peer_tls_files() + assert (key, cert) == ("PK", "PC") + + +def test_certificate_available_pushes_on_empty(harness): + charm = harness.charm + with ( + patch.object(charm.cluster_manager, "configure_system_passwords"), + patch.object(charm.config_manager, "update_config"), + ): + harness.set_leader(True) + + tls = charm.tls + # requirer reports no assigned cert (relation gone) -> getters return nothing + tls.client_certificate.get_assigned_certificates = MagicMock(return_value=([], None)) + charm.tls_manager.push_tls_files = MagicMock() + + tls._on_certificate_available(MagicMock()) + + assert charm.tls_manager.get_client_tls_files() == (None, None, None) + charm.tls_manager.push_tls_files.assert_called_once() + + +def test_peer_certificate_available_clears_ca_on_empty(harness): + charm = harness.charm + with ( + patch.object(charm.cluster_manager, "configure_system_passwords"), + patch.object(charm.config_manager, "update_config"), + ): + harness.set_leader(True) + + tls = charm.tls + # seed a current CA via a prior rotation, then the relation goes empty + charm.tls_manager.rotate_peer_ca("PCA") + tls.peer_certificate.get_assigned_certificates = MagicMock(return_value=([], None)) + charm.tls_manager.push_tls_files = MagicMock() + + tls._on_peer_certificate_available(MagicMock()) + + peer = charm.state.peer + assert peer.old_ca == "PCA" + assert peer.current_ca is None + charm.tls_manager.push_tls_files.assert_called_once() + + +def test_relation_broken_client_wired(harness): + """relation_broken on TLS_CLIENT_RELATION routes to _on_certificate_available (live push).""" + charm = harness.charm + client_rel_id = harness.add_relation("client-certificates", "tls-provider") + charm.tls_manager.push_tls_files = MagicMock() + harness.remove_relation(client_rel_id) + + # With the relation gone, the live getter reports nothing. + assert charm.tls_manager.get_client_tls_files() == (None, None, None) + + +def test_relation_broken_peer_wired(harness): + """relation_broken on TLS_PEER_RELATION routes to _on_peer_certificate_available (clears CA).""" + charm = harness.charm + # seed a current CA so the broken path has something to retire + charm.tls_manager.rotate_peer_ca("PCA") + + peer_rel_id = harness.add_relation("peer-certificates", "tls-provider") + charm.tls_manager.push_tls_files = MagicMock() + harness.remove_relation(peer_rel_id) + + # The broken handler retired the current CA into old-ca and cleared current. + assert charm.state.peer.current_ca is None + assert charm.state.peer.old_ca == "PCA" + + +def _set_unit_db(harness, key, value): + rel_id = harness.model.get_relation("database-peers").id + harness.update_relation_data(rel_id, harness.charm.unit.name, {key: value}) + + +def test_client_and_peer_requesters_have_distinct_common_names(substrate, harness): + """Client and peer requesters use distinct CNs drawn from different databag keys. + + certificate_requests are baked at init time (before the test updates the databag), so + we verify distinctness via the live state properties (which read directly from the + databag) and confirm both requester objects were constructed. + + On VM the CNs are host-derived (database-address / database-peers-address). On K8s + both CNs collapse to the endpoints FQDN (original charm parity) — still distinct from + each other is not required there, only that both are the endpoints FQDN. + """ + _set_unit_db(harness, "database-address", "10.1.2.3") + _set_unit_db(harness, "database-peers-address", "10.4.5.6") + + state = harness.charm.state + if substrate == "vm": + # VM: client CN reads database-address, peer CN reads database-peers-address. + assert state.client_common_name == "10.1.2.3" + assert state.peer_common_name == "10.4.5.6" + assert state.client_common_name != state.peer_common_name + else: + # K8s: both CNs are the unit endpoints FQDN (operator-cert parity). + app = state.model.app.name + expected = f"{app}-0.{app}-endpoints" + assert state.client_common_name == expected + assert state.peer_common_name == expected + + tls = harness.charm.tls + assert tls.client_certificate is not None + assert tls.peer_certificate is not None + + +def test_refresh_event_defined_and_wired(harness): + """refresh_tls_certificates_event exists; peer relation_changed emits it without error.""" + tls = harness.charm.tls + assert hasattr(tls, "refresh_tls_certificates_event") + + charm = harness.charm + peer_rel_id = harness.model.get_relation("database-peers").id + with ( + patch.object(charm.cluster_manager, "configure_system_passwords", return_value=None), + patch.object(charm.config_manager, "update_config", return_value=None), + ): + harness.update_relation_data( + peer_rel_id, charm.unit.name, {"database-address": "10.9.8.7"} + ) + + +def test_certificate_available_defers_when_internal_ca_absent(harness): + """When internal-ca is not yet set, _on_certificate_available defers and skips push. + + Mirrors postgresql-operator/src/relations/tls.py lines 157-161: the handler must + not attempt file writes before the CA is present (K8s Pebble may not be ready). + """ + tls = harness.charm.tls + tls.client_certificate.get_assigned_certificates = MagicMock( + return_value=_fake_assigned("CC", "CA", "CK") + ) + harness.charm.tls_manager.push_tls_files = MagicMock() + + event = MagicMock() + # internal_ca is None because no leader has set it yet + assert harness.charm.state.application.internal_ca is None + tls._on_certificate_available(event) + + event.defer.assert_called_once() + harness.charm.tls_manager.push_tls_files.assert_not_called() + + +def test_peer_certificate_available_defers_when_internal_ca_absent(harness): + """When internal-ca is not yet set, _on_peer_certificate_available defers and skips push.""" + tls = harness.charm.tls + tls.peer_certificate.get_assigned_certificates = MagicMock( + return_value=_fake_assigned("PC", "PCA", "PK") + ) + harness.charm.tls_manager.push_tls_files = MagicMock() + + event = MagicMock() + assert harness.charm.state.application.internal_ca is None + tls._on_peer_certificate_available(event) + + event.defer.assert_called_once() + harness.charm.tls_manager.push_tls_files.assert_not_called() + + +def test_certificate_available_defers_on_workload_file_error(harness): + """When push_tls_files raises PostgreSQLFileOperationError, the handler defers. + + Mirrors postgresql-operator/src/relations/tls.py lines 162-170: workload + file-write failures (e.g. Pebble not yet ready on K8s) must defer rather + than crash the hook. + """ + charm = harness.charm + with ( + patch.object(charm.cluster_manager, "configure_system_passwords"), + patch.object(charm.config_manager, "update_config"), + ): + harness.set_leader(True) + + tls = charm.tls + tls.client_certificate.get_assigned_certificates = MagicMock( + return_value=_fake_assigned("CC", "CA", "CK") + ) + charm.tls_manager.push_tls_files = MagicMock( + side_effect=PostgreSQLFileOperationError("disk full") + ) + + event = MagicMock() + tls._on_certificate_available(event) + + event.defer.assert_called_once() + + +def test_peer_certificate_available_defers_on_workload_file_error(harness): + """When push_tls_files raises PostgreSQLFileOperationError, the peer handler defers.""" + charm = harness.charm + with ( + patch.object(charm.cluster_manager, "configure_system_passwords"), + patch.object(charm.config_manager, "update_config"), + ): + harness.set_leader(True) + + tls = charm.tls + tls.peer_certificate.get_assigned_certificates = MagicMock( + return_value=_fake_assigned("PC", "PCA", "PK") + ) + charm.tls_manager.push_tls_files = MagicMock( + side_effect=PostgreSQLFileOperationError("pebble not ready") + ) + + event = MagicMock() + tls._on_peer_certificate_available(event) + + event.defer.assert_called_once() diff --git a/tests/unit/test_tls_manager.py b/tests/unit/test_tls_manager.py new file mode 100644 index 0000000..a0cea90 --- /dev/null +++ b/tests/unit/test_tls_manager.py @@ -0,0 +1,299 @@ +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. + +from datetime import timedelta +from types import SimpleNamespace +from unittest.mock import MagicMock, patch, sentinel + + +class _FakePrivateKey: + """Minimal stand-in for PrivateKey: str(key) returns the raw PEM string.""" + + def __init__(self, raw: str) -> None: + self._raw = raw + + def __str__(self) -> str: + return self._raw + + +def _fake_assigned(cert, ca, key): + """Mimic TLSCertificatesRequiresV4.get_assigned_certificates() -> (list, PrivateKey|None).""" + return [SimpleNamespace(certificate=cert, ca=ca)], _FakePrivateKey(key) + + +def _set_client(mgr, cert, ca, key): + mgr.client_certificate.get_assigned_certificates = MagicMock( + return_value=_fake_assigned(cert, ca, key) + ) + + +def _set_peer(mgr, cert, ca, key): + mgr.peer_certificate.get_assigned_certificates = MagicMock( + return_value=_fake_assigned(cert, ca, key) + ) + + +def _set_empty(requirer): + requirer.get_assigned_certificates = MagicMock(return_value=([], None)) + + +# --- CA rotation (current-ca / old-ca are 16/edge secrets, kept; cert/key are NOT stored) --- + + +def test_rotate_peer_ca_sets_current(harness): + mgr = harness.charm.tls_manager + mgr.rotate_peer_ca("CA1") + + peer = harness.charm.state.peer + assert peer.current_ca == "CA1" + assert peer.old_ca is None + + +def test_rotate_peer_ca_rotates_on_change(harness): + mgr = harness.charm.tls_manager + mgr.rotate_peer_ca("CA1") + mgr.rotate_peer_ca("CA2") + + peer = harness.charm.state.peer + assert peer.current_ca == "CA2" + assert peer.old_ca == "CA1" + + +def test_rotate_peer_ca_no_rotation_when_unchanged(harness): + mgr = harness.charm.tls_manager + mgr.rotate_peer_ca("CA1") + mgr.rotate_peer_ca("CA1") + + peer = harness.charm.state.peer + assert peer.current_ca == "CA1" + assert peer.old_ca is None + + +def test_clear_peer_ca_rotates_current_to_old(harness): + mgr = harness.charm.tls_manager + mgr.rotate_peer_ca("CA") + mgr.clear_peer_ca() + + peer = harness.charm.state.peer + assert peer.old_ca == "CA" + assert peer.current_ca is None + + +def test_rotate_peer_ca_clears_stale_old_ca_on_reenable(harness): + """Re-enabling after a disable must not leave a stale pre-disable CA in the bundle.""" + mgr = harness.charm.tls_manager + mgr.rotate_peer_ca("A") + mgr.rotate_peer_ca("B") # rotate A -> B + assert mgr.state.peer.current_ca == "B" + assert mgr.state.peer.old_ca == "A" + mgr.clear_peer_ca() # disable stashes B as old and clears current + assert mgr.state.peer.current_ca is None + assert mgr.state.peer.old_ca == "B" + mgr.rotate_peer_ca("C") # re-enable with a fresh CA + assert mgr.state.peer.current_ca == "C" + assert mgr.state.peer.old_ca is None + + +# --- operator cert/key are fetched LIVE from the requirer, never persisted --- + + +def test_get_client_tls_files_none_when_absent(harness): + mgr = harness.charm.tls_manager + _set_empty(mgr.client_certificate) + assert mgr.get_client_tls_files() == (None, None, None) + + +def test_get_client_tls_files_returns_live(harness): + mgr = harness.charm.tls_manager + _set_client(mgr, cert="C", ca="A", key="K") + assert mgr.get_client_tls_files() == ("K", "A", "C") + + +def test_get_peer_ca_bundle_composes_live_old_internal(harness): + charm = harness.charm + # leadership is required to write the app-scoped internal-ca secret + with ( + patch.object(charm.cluster_manager, "configure_system_passwords"), + patch.object(charm.config_manager, "update_config"), + ): + harness.set_leader(True) + + mgr = charm.tls_manager + _set_peer(mgr, cert="PC", ca="CUR", key="PK") # live operator CA = CUR + charm.state.peer.old_ca = "OLD" + charm.state.set_secret("app", "internal-ca", "INT") + + assert mgr.get_peer_ca_bundle() == "CUR\nOLD\nINT" + + +def test_get_peer_tls_files_prefers_live_operator(harness): + mgr = harness.charm.tls_manager + _set_peer(mgr, cert="PC", ca="PCA", key="PK") + key, ca, cert = mgr.get_peer_tls_files() + assert (key, cert) == ("PK", "PC") + assert ca == "PCA" # bundle = live operator CA only (no old, no internal here) + + +def test_get_peer_tls_files_falls_back_to_internal(harness): + charm = harness.charm + # leadership is required to write the app-scoped internal-ca secret + with ( + patch.object(charm.cluster_manager, "configure_system_passwords"), + patch.object(charm.config_manager, "update_config"), + ): + harness.set_leader(True) + + mgr = charm.tls_manager + _set_empty(mgr.peer_certificate) + peer = charm.state.peer + peer.internal_key = "IK" + peer.internal_cert = "IC" + charm.state.set_secret("app", "internal-ca", "ICA") + + assert mgr.get_peer_tls_files() == ("IK", "ICA", "IC") + + +def test_push_tls_files_writes_expected_files(harness): + mgr = harness.charm.tls_manager + _set_client(mgr, cert="CC", ca="CA", key="CK") + _set_peer(mgr, cert="PC", ca="PCA", key="PK") + + mgr.workload.write_text = MagicMock() + mgr.push_tls_files() + + written = {call.args[1].name: call.args[0] for call in mgr.workload.write_text.call_args_list} + assert written["key.pem"] == "CK" + assert written["ca.pem"] == "CA" + assert written["cert.pem"] == "CC" + assert written["peer_key.pem"] == "PK" + assert written["peer_cert.pem"] == "PC" + assert written["peer_ca.pem"] == "PCA" + assert written["peer_ca_bundle.pem"] == "PCA" + # all TLS files are written with the workload's substrate-specific mode + # (VM 0o600, K8s 0o400), user, and group + expected_mode = mgr.workload.tls_file_mode + expected_user = mgr.workload.user + expected_group = mgr.workload.group + for call in mgr.workload.write_text.call_args_list: + assert call.args[2] == expected_mode + assert call.kwargs["user"] == expected_user + assert call.kwargs["group"] == expected_group + # every TLS file is written under the substrate-correct TLS dir (paths.tls) + for call in mgr.workload.write_text.call_args_list: + assert call.args[1].parent == mgr.workload.paths.tls + + +def test_tls_manager_constructor_injects_requirers(harness): + """TLSManager takes the two cert requirers at construction (no post-init mutation). + + The manager holds no PostgreSQL client (it never uses one); it reads operator + cert/key live from the constructor-injected requirers. + """ + from single_kernel_postgresql.managers.tls import TLSManager + + client_req = harness.charm.tls.client_certificate + peer_req = harness.charm.tls.peer_certificate + mgr = TLSManager( + harness.charm.state, + harness.charm.tls_manager.workload, + client_certificate=client_req, + peer_certificate=peer_req, + ) + assert mgr.client_certificate is client_req + assert mgr.peer_certificate is peer_req + # the charm-wired manager gets the same requirers the TLS handler built + assert harness.charm.tls_manager.client_certificate is harness.charm.tls.client_certificate + assert harness.charm.tls_manager.peer_certificate is harness.charm.tls.peer_certificate + + +def test_client_tls_files_on_disk(harness): + """Reports presence of the client key/cert/ca files; container errors count as absent.""" + from single_kernel_postgresql.config.exceptions import PostgreSQLFileOperationError + + mgr = harness.charm.tls_manager + with patch.object(mgr.workload, "exists", return_value=True) as _exists: + assert mgr.client_tls_files_on_disk() is True + assert _exists.call_count == 3 + with patch.object(mgr.workload, "exists", side_effect=[True, False]): + assert mgr.client_tls_files_on_disk() is False + with patch.object(mgr.workload, "exists", side_effect=PostgreSQLFileOperationError("down")): + assert mgr.client_tls_files_on_disk() is False + + +def test_generate_internal_peer_ca_stores_secrets(harness): + """generate_internal_peer_ca persists the generated CA cert/key into app state.""" + charm = harness.charm + # leadership is required to write the app-scoped internal-ca secrets + with ( + patch.object(charm.cluster_manager, "configure_system_passwords"), + patch.object(charm.config_manager, "update_config"), + ): + harness.set_leader(True) + + with ( + patch( + "single_kernel_postgresql.managers.tls.generate_private_key", + return_value=sentinel.ca_key, + ), + patch( + "single_kernel_postgresql.managers.tls.generate_ca", + return_value=sentinel.ca, + ), + ): + charm.tls_manager.generate_internal_peer_ca() + + assert charm.state.application.internal_ca_key == str(sentinel.ca_key) + assert charm.state.application.internal_ca == str(sentinel.ca) + + +def test_generate_internal_peer_cert_stores_material(harness): + """generate_internal_peer_cert persists internal key/cert into peer state and pushes nothing. + + Writing the internal peer cert/CA to disk is owned by the (not-yet-migrated) + config subsystem, so generating the material must not touch the workload. + """ + charm = harness.charm + mgr = charm.tls_manager + # leadership is required to write/read the app-scoped internal-ca secrets the + # internal peer cert is signed against + with ( + patch.object(charm.cluster_manager, "configure_system_passwords"), + patch.object(charm.config_manager, "update_config"), + ): + harness.set_leader(True) + charm.state.set_secret("app", "internal-ca-key", "ca-key-content") + charm.state.set_secret("app", "internal-ca", "ca-content") + + # spy on the only file-writing primitive to prove no push happens + mgr.workload.write_text = MagicMock() + with ( + patch( + "single_kernel_postgresql.managers.tls.generate_private_key", + return_value=sentinel.cert_key, + ), + patch( + "single_kernel_postgresql.managers.tls.generate_csr", + return_value=sentinel.cert_csr, + ), + patch( + "single_kernel_postgresql.managers.tls.generate_certificate", + return_value=sentinel.cert, + ) as _generate_certificate, + patch("single_kernel_postgresql.managers.tls.PrivateKey") as _private_key, + patch("single_kernel_postgresql.managers.tls.Certificate") as _certificate, + ): + _private_key.from_string.return_value = sentinel.ca_key + _certificate.from_string.return_value = sentinel.ca_cert + + mgr.generate_internal_peer_cert() + + # the CSR is signed by the stored internal CA (cert + key), valid 20 years + _generate_certificate.assert_called_once_with( + sentinel.cert_csr, sentinel.ca_cert, sentinel.ca_key, validity=timedelta(days=7300) + ) + + # generated material is persisted in peer-unit state (round-trip) + assert charm.state.peer.internal_cert == str(sentinel.cert) + assert charm.state.peer.internal_key == str(sentinel.cert_key) + # writing the internal peer cert/CA to disk is owned by the config subsystem + mgr.workload.write_text.assert_not_called()