Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 56 additions & 0 deletions single_kernel_postgresql/core/peer_relation.py
Original file line number Diff line number Diff line change
Expand Up @@ -225,6 +225,48 @@ def is_connectivity_enabled(self) -> bool:
return True
return self.relation.data[self.unit].get("connectivity", "on") == "on"

@property
def config_hash(self) -> str | None:
"""Get the last-applied PostgreSQL config hash from the peer relation data."""
if not self.relation:
return None
return self.relation.data[self.unit].get("config_hash")

@config_hash.setter
def config_hash(self, value: str) -> None:
"""Set the last-applied PostgreSQL config hash in the peer relation data."""
if not self.relation:
return
self.relation.data[self.unit]["config_hash"] = value

@property
def user_hash(self) -> str | None:
"""Get the last-applied users hash from the peer relation data."""
if not self.relation:
return None
return self.relation.data[self.unit].get("user_hash")

@user_hash.setter
def user_hash(self, value: str) -> None:
"""Set the last-applied users hash in the peer relation data."""
if not self.relation:
return
self.relation.data[self.unit]["user_hash"] = value

@property
def tls(self) -> bool:
"""Get the last-rendered TLS flag from the peer relation data."""
if not self.relation:
return False
return self.relation.data[self.unit].get("tls") == "enabled"

@tls.setter
def tls(self, value: bool) -> None:
"""Set the last-rendered TLS flag in the peer relation data."""
if not self.relation:
return
self.relation.data[self.unit]["tls"] = "enabled" if value else ""

@cached_property
def data(self) -> MutableMapping[str, str]:
"""Escape hatch method to access the peer data directly."""
Expand Down Expand Up @@ -418,6 +460,20 @@ def is_ldap_enabled(self) -> bool:
"""Return whether this unit has LDAP enabled."""
return self.is_ldap_charm_related and self.is_cluster_initialised

@property
def user_hash(self) -> str | None:
"""Get the last-applied users hash from the peer relation data."""
if not self.relation:
return None
return self.relation.data[self.app].get("user_hash")

@user_hash.setter
def user_hash(self, value: str) -> None:
"""Set the last-applied users hash in the peer relation data."""
if not self.relation:
return
self.relation.data[self.app]["user_hash"] = value

def get_secret(self, key: str) -> str | None:
"""Get the secret value for 'key' from the peer relation data."""
if not self.relation:
Expand Down
24 changes: 23 additions & 1 deletion tests/unit/conftest.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,36 @@
# Copyright 2026 Canonical Ltd.
# See LICENSE file for licensing details.

from unittest.mock import patch
from unittest.mock import patch, sentinel

import pytest
from ops.testing import Harness
from single_kernel_postgresql.charms import k8s_charm, vm_charm
from single_kernel_postgresql.config.literals import PEER_RELATION


@pytest.fixture
def patch_crypto():
"""Stub the tls-crypto generators so leader-elected paths skip real RSA keygen.

set_leader() fires _on_leader_elected -> configure_internal_peer_ca, which would
otherwise run real generate_private_key/generate_ca. The dedicated generate_* tests
patch with their own sentinels to assert call args; this fixture is for the
incidental paths that only need internal-ca to be present, not its content.
"""
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,
),
):
yield


@pytest.fixture
def harness(substrate, test_charm_path):
"""A begun Harness for the substrate's test charm, with the peer relation added."""
Expand Down
110 changes: 110 additions & 0 deletions tests/unit/test_peer_relation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
#!/usr/bin/env python3

# Copyright 2026 Canonical Ltd.
# See LICENSE file for licensing details.

"""Tests for the config/user hash + tls plain-databag accessors on peer state.

These accessors scaffold the config-subsystem port: PostgreSQLPeer.config_hash
and .user_hash persist the last-applied config/user hashes, PostgreSQLPeer.tls
persists the last-rendered TLS flag, and PostgreSQLApplication.user_hash
persists the app-wide user hash. All are plain (non-secret) databag values,
matching the charm's unit_peer_data / app_peer_data storage.
"""


def _get_unit_db(harness, key):
rel_id = harness.model.get_relation("database-peers").id
return harness.get_relation_data(rel_id, harness.charm.unit.name).get(key)


def _get_app_db(harness, key):
rel_id = harness.model.get_relation("database-peers").id
return harness.get_relation_data(rel_id, harness.charm.app.name).get(key)


# -- PostgreSQLPeer.config_hash ----------------------------------------------


def test_config_hash_unset_is_none(harness):
assert harness.charm.state.peer.config_hash is None


def test_config_hash_roundtrip(harness):
peer = harness.charm.state.peer
peer.config_hash = "abc123"
assert peer.config_hash == "abc123"


def test_config_hash_writes_exact_databag_key(harness):
harness.charm.state.peer.config_hash = "abc123"
assert _get_unit_db(harness, "config_hash") == "abc123"


# -- PostgreSQLPeer.user_hash -------------------------------------------------


def test_peer_user_hash_unset_is_none(harness):
assert harness.charm.state.peer.user_hash is None


def test_peer_user_hash_roundtrip(harness):
peer = harness.charm.state.peer
peer.user_hash = "def456"
assert peer.user_hash == "def456"


def test_peer_user_hash_writes_exact_databag_key(harness):
harness.charm.state.peer.user_hash = "def456"
assert _get_unit_db(harness, "user_hash") == "def456"


# -- PostgreSQLPeer.tls --------------------------------------------------------


def test_tls_unset_is_false(harness):
assert harness.charm.state.peer.tls is False


def test_tls_set_true_roundtrip(harness):
peer = harness.charm.state.peer
peer.tls = True
assert peer.tls is True


def test_tls_set_true_writes_enabled(harness):
harness.charm.state.peer.tls = True
assert _get_unit_db(harness, "tls") == "enabled"


def test_tls_set_false_roundtrip(harness):
peer = harness.charm.state.peer
peer.tls = True
peer.tls = False
assert peer.tls is False


def test_tls_set_false_clears_enabled_key(harness):
"""tls=False writes "", which ops relation data drops on write — observed as key absence."""
peer = harness.charm.state.peer
peer.tls = True
peer.tls = False
assert _get_unit_db(harness, "tls") is None


# -- PostgreSQLApplication.user_hash ------------------------------------------


def test_app_user_hash_unset_is_none(harness):
assert harness.charm.state.application.user_hash is None


def test_app_user_hash_roundtrip(harness):
application = harness.charm.state.application
application.user_hash = "ghi789"
assert application.user_hash == "ghi789"


def test_app_user_hash_writes_exact_databag_key(harness):
harness.charm.state.application.user_hash = "ghi789"
assert _get_app_db(harness, "user_hash") == "ghi789"
Loading
Loading