Update tls_certificates_interface lib to fix permission deny error during scale operation

lib/charms/tls_certificates_interface/v2/tls_certificates.py

@@ -298,7 +298,7 @@ def _on_all_certificates_invalidated(self, event: AllCertificatesInvalidatedEven
 )
 from ops.framework import EventBase, EventSource, Handle, Object
 from ops.jujuversion import JujuVersion
-from ops.model import SecretNotFoundError
+from ops.model import Relation, SecretNotFoundError
 
 # The unique Charmhub library identifier, never change it
 LIBID = "afd8c2bccf834997afce12c2706d2ede"
@@ -308,7 +308,7 @@ def _on_all_certificates_invalidated(self, event: AllCertificatesInvalidatedEven
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 12
+LIBPATCH = 13
 
 PYDEPS = ["cryptography", "jsonschema"]
 
@@ -897,6 +897,22 @@ def __init__(self, charm: CharmBase, relationship_name: str):
         self.charm = charm
         self.relationship_name = relationship_name
 
+    def _load_app_relation_data(self, relation: Relation) -> dict:
+        """Loads relation data from the application relation data bag.
+
+        Json loads all data.
+
+        Args:
+            relation_object: Relation data from the application databag
+
+        Returns:
+            dict: Relation data in dict format.
+        """
+        # If unit is not leader, it does not try to reach relation data.
+        if not self.model.unit.is_leader():
+            return {}
+        return _load_relation_data(relation.data[self.charm.app])
+
     def _add_certificate(
         self,
         relation_id: int,
@@ -931,7 +947,7 @@ def _add_certificate(
             "ca": ca,
             "chain": chain,
         }
-        provider_relation_data = _load_relation_data(relation.data[self.charm.app])
+        provider_relation_data = self._load_app_relation_data(relation)
         provider_certificates = provider_relation_data.get("certificates", [])
         certificates = copy.deepcopy(provider_certificates)
         if new_certificate in certificates:
@@ -964,7 +980,7 @@ def _remove_certificate(
             raise RuntimeError(
                 f"Relation {self.relationship_name} with relation id {relation_id} does not exist"
             )
-        provider_relation_data = _load_relation_data(relation.data[self.charm.app])
+        provider_relation_data = self._load_app_relation_data(relation)
         provider_certificates = provider_relation_data.get("certificates", [])
         certificates = copy.deepcopy(provider_certificates)
         for certificate_dict in certificates:
@@ -999,7 +1015,7 @@ def revoke_all_certificates(self) -> None:
         This method is meant to be used when the Root CA has changed.
         """
         for relation in self.model.relations[self.relationship_name]:
-            provider_relation_data = _load_relation_data(relation.data[self.charm.app])
+            provider_relation_data = self._load_app_relation_data(relation)
             provider_certificates = copy.deepcopy(provider_relation_data.get("certificates", []))
             for certificate in provider_certificates:
                 certificate["revoked"] = True
@@ -1081,7 +1097,7 @@ def get_issued_certificates(
             else self.model.relations.get(self.relationship_name, [])
         )
         for relation in relations:
-            provider_relation_data = _load_relation_data(relation.data[self.charm.app])
+            provider_relation_data = self._load_app_relation_data(relation)
             provider_certificates = provider_relation_data.get("certificates", [])
 
             certificates[relation.app.name] = []  # type: ignore[union-attr]
@@ -1111,9 +1127,12 @@ def _on_relation_changed(self, event: RelationChangedEvent) -> None:
         Returns:
             None
         """
-        assert event.unit is not None
+        if event.unit is None:
+            return
+        if not self.model.unit.is_leader():
+            return
         requirer_relation_data = _load_relation_data(event.relation.data[event.unit])
-        provider_relation_data = _load_relation_data(event.relation.data[self.charm.app])
+        provider_relation_data = self._load_app_relation_data(event.relation)
         if not self._relation_data_is_valid(requirer_relation_data):
             logger.debug("Relation data did not pass JSON Schema validation")
             return
@@ -1152,7 +1171,7 @@ def _revoke_certificates_for_which_no_csr_exists(self, relation_id: int) -> None
         )
         if not certificates_relation:
             raise RuntimeError(f"Relation {self.relationship_name} does not exist")
-        provider_relation_data = _load_relation_data(certificates_relation.data[self.charm.app])
+        provider_relation_data = self._load_app_relation_data(certificates_relation)
         list_of_csrs: List[str] = []
         for unit in certificates_relation.units:
             requirer_relation_data = _load_relation_data(certificates_relation.data[unit])

tests/integration/test_integration.py

@@ -8,6 +8,7 @@
 
 import pytest
 import yaml
+from pytest_operator.plugin import OpsTest
 
 logger = logging.getLogger(__name__)
 
@@ -19,7 +20,7 @@
 
 @pytest.fixture(scope="module")
 @pytest.mark.abort_on_fail
-async def build_and_deploy(ops_test):
+async def build_and_deploy(ops_test: OpsTest):
     """Build the charm-under-test and deploy it."""
     charm = await ops_test.build_charm(".")
     await ops_test.model.deploy(
@@ -37,7 +38,7 @@ async def build_and_deploy(ops_test):
 
 @pytest.mark.abort_on_fail
 async def test_given_charm_is_built_when_deployed_then_status_is_active(
-    ops_test,
+    ops_test: OpsTest,
     build_and_deploy,
 ):
     await ops_test.model.wait_for_idle(
@@ -48,7 +49,7 @@ async def test_given_charm_is_built_when_deployed_then_status_is_active(
 
 
 async def test_given_tls_requirer_is_deployed_and_related_then_certificate_is_created_and_passed_correctly(  # noqa: E501
-    ops_test,
+    ops_test: OpsTest,
     build_and_deploy,
 ):
     await ops_test.model.add_relation(
@@ -66,6 +67,16 @@ async def test_given_tls_requirer_is_deployed_and_related_then_certificate_is_cr
     assert action_output["csr"] is not None
 
 
+async def test_given_charm_scaled_then_charm_does_not_crash(
+    ops_test: OpsTest,
+    build_and_deploy,
+):
+    await ops_test.model.applications[APP_NAME].scale(2)
+    await ops_test.model.wait_for_idle(apps=[APP_NAME], timeout=1000, wait_for_exact_units=2)
+    await ops_test.model.applications[APP_NAME].scale(1)
+    await ops_test.model.wait_for_idle(apps=[APP_NAME], timeout=1000, wait_for_exact_units=1)
+
+
 async def run_get_certificate_action(ops_test) -> dict:
     """Runs `get-certificate` on the `tls-requirer-requirer/0` unit.
 

tests/unit/test_charm.py

@@ -17,11 +17,11 @@ class TestCharm(unittest.TestCase):
     def setUp(self):
         self.harness = ops.testing.Harness(SelfSignedCertificatesCharm)
         self.addCleanup(self.harness.cleanup)
+        self.harness.set_leader(is_leader=True)
         self.harness.begin()
 
     def test_given_invalid_config_when_config_changed_then_status_is_blocked(self):
         key_values = {"ca-common-name": "", "certificate-validity": 100}
-        self.harness.set_leader(is_leader=True)
 
         self.harness.update_config(key_values=key_values)