Repository: canonical/operator

Issue:
Better error messages on permission errors with accessing relation databags
In non-peer relations, a unit cannot read or write its own application's databag, which results into authorization errors.
However, those errors surface in unexpected places which can confuse developers:

```
self.unit.is_leader() # False
self.relation.data # -->  ModelError("ERROR permission denied")
self.relation.data[relation.app]  # {}
self.relation.data[self.app]  # -->  ModelError("ERROR permission denied")
self.relation.data[relation.app]['foo'] = 'bar' # -->  ModelError("ERROR permission denied")
self.relation.data[relation.units[0]]['foo'] = 'bar' # -->  ModelError("ERROR permission denied") 
```

goal:
```
self.unit.is_leader() # False
self.relation.data # RelationData(local_app: <N/A>, local_unit: {...}, remote_app: {...}, remote_unit: {...}) 
self.relation.data[relation.app]  # {}
self.relation.data[self.app]  # -->  ModelError("ERROR permission denied: non-leader units cannot read or write their own application's databag.")
self.relation.data[relation.app]['foo'] = 'bar' # -->  ModelError("ERROR permission denied: cannot write databags of remote applications")
self.relation.data[relation.units[0]]['foo'] = 'bar' # -->  ModelError("ERROR permission denied: cannot write databags of remote units"") 
```



Discussion:
PietroPasotti: We need to discuss: harness testing model backend serves at the moment two purposes:
1) Mock the interactions from the charm's POV (restricted by the same rule as a production charm would be)
2) Offer the testing environment tools to read/write databags (unrestrictedly)

These two needs are now starting to clash because we want the harness model backend to behave as juju does when a charm tries to read/write the data, but to leave tests 'admin access' when they are doing their testing.

How do we solve this?

Proposal: 
relation objects facing the charm return 'locked' (mock) RelationDataContents objects,
relation objects facing the harness return 'unlocked' ones.

Therefore:
```python
harness.set_leader(False)
harness.get_relation('foo').data[harness.charm.app]['foo'] = 'bar'  # ok

harness.charm.model.get_relation('foo').data[harness.charm.app]['foo'] = 'bar'  # ERROR!
```

@pengale @rwcarlsen @jnsgruk @jameinel 
PietroPasotti: Robert recommended to check the global env var 'am I inside an event hook' and only if YES then raise an error, else keep it quiet and behave as if we're in 'admin mode'.

Will do that and think of any edge cases we might be hitting (if someone is 'manually' emitting events by calling handlers with mocked events for example...) and mitigating strategies

PietroPasotti: In conclusion:
- the harness' backend is now access-controlling databags as the real backend would.
- for that to be possible, while still allowing test code to have unrestricted read/write access to the databags, we rely on the _is_hook_running flag set on the backend.
- In order to allow test code to enable 'strict' checking mode even when one is not relying on the Harness' internal machinery (_reemit specifically), I promoted EventContext to a `Harness.backend.event_context` public method (also directly accessible as `Harness.event_context`).

I think this is reasonable because otherwise it'd be too hard to switch between the two harness operation modes (strict: do as if we're really inside a charm; and admin: we're mocking and testing stuff).
PietroPasotti: RE the failing test: there's an assertBackendCalls that breaks for funny reasons.
It's the backend calls generated by `repr(relation.data)`

in python > 3.5 we get:
```
    [('relation_get', 1, 'myapp/0', False),
     ('is_leader',),
     ('relation_get', 1, 'remoteapp1/0', False),
     ('relation_get', 1, 'remoteapp1', True)]
     ```

in python 3.5 for some reason we get instead:
```
    [('relation_get', 1, 'remoteapp1/0', False),
     ('relation_get', 1, 'remoteapp1', True),
     ('is_leader',),
     ('relation_get', 1, 'myapp/0', False)]
     ```

I think it's got something to do with dict sorting :D 

rwcarlsen: Is this ready for a re-review?
PietroPasotti: > Is this ready for a re-review?

Yes and no. I'm done getting rid of large chunks of the testing data wrappers, but I am left with one big bug which I'm not sure how to tackle best.

Issue is: we want to use `ops.Model` for as much as possible. `ops.Model.relations[idx].data` returns RelationDataContents, which, on `__setitem__`, calls `relation_set`. 
What we want, in order to avoid code duplication and imperfectly replicating model logic in testing code, is to keep that flow as intact as possible.
The issue is that patching `relation_set` in `TestingModelBackend` is not enough, because `relation_set` is plumbed to always be called from the perspective of "this" unit or application, so it won't allow us to set "remote" units/apps databags, as it does not even expose the necessary arguments. So from the perspective of the harness, when we intercept a relation-set call, we can only assume we're setting local databags, which in testing code is not always the case.

Possible solutions: 
- patch `RelationDataContents` and overwrite its `__getitem__/__setitem__` in such a way that `TestingModelBackend.relation_get/relation_set` are called with some extra arguments (namely, `RelationDataContents._entity`). 
   - pros: kind of easy
   - cons: we'd need to still duplicate  `__getitem__/__setitem__`, which contain exactly that validation logic which we were trying to avoid to duplicate.
 - from `TestingModelBackend.relation_get/relation_set`, use `inspect` to get a hold of the outer frame and grab the `_entity` which the relation data is about. (erm...)
PietroPasotti: > I'm not sure I understand. It is the case that it doesn't work at all (without one of your proposed solutions) or that it doesn't catch and emit errors we want? The test harness itself only represents the "world" from the view of a single charm/unit - so it seems like it should be fine. But maybe I'm not imagining things correct. A concrete example of what we want tests to be able to do and how this problem prevents that might help me.

Right, so it is ready for a re-review.
I chose for the following approach:

add a backend method `update_relation_data` which in 'real' backend does little more than call `relation_set`. The testing backend bypasses `relation_set` and goes directly directly to manipulate the 'raw' relation data.
It works, it feels clean-ish, but I'm wondering now what the purpose of relation-set in the `testingmodelbackend` should be. Should it go?
We get a bunch of broken tests because the calls being recorded have changed, as `relation_set` is now basically out of the picture in the harness. We'd get the same if we were to change the `relation_set` signature directly, so there's no avoiding that. 

Before I fix those, I'd like to get a green light from @pengale as well, and think harder of whether this is the best approach
rwcarlsen: I've run into those test issues before.  I think it's not ideal to diff internal implementation call stacks for testing :/
pengale: As a follow-up to my comment, I'm not sure that this is the correct solution, but it might be more natural to introduce something like:

```
with as_leader():
    relation.data[app][key] = val
```

You could extend this idea to do other things, like:

```
with as_unit_0():
    set_some_remote_rel_data
```

Or even:

```
with as_human():
    set_some_config_val
```

That way, your decorator only intrudes on the situation when you specifically need it. This approach isn't as friendly to existing tests, which might break if we implement another strict check that better matches production behavior. But it does make stuff a bit more natural to reason about. "For the moment, I'm behaving as the unit that is the leader."
PietroPasotti: I addressed the last two points by @rwcarlsen and left @pengale's comments unaddressed as I tend to agree with Robert's viewpoint.
It's not the goal of this PR to change the way user code is written, its intention was to enable testing code to be more strict, as an opt-in feature, while maintaining compatibility with existing test code.

As a result of this, our testing code had to be changed:
- model tests all need to be wrapped in the 'strict' context to ensure that the validation checks are enabled, because these tests check the framework from the POV of a charm
- testing tests needed little-to-no-changes (so far as the event context is concerned) because they check the framework from the POV of testing code (i.e. code where the access control checks are skipped).

I agree with @pengale that opt-in means: barely anyone will even notice it's there, and that ideally we'd make strict mode the default, but that is a breaking change and out of scope for now.
We could plan ahead a transition period with a warning (like for SIMULATE_CAN_CONNECT), and after a couple of major releases, make strict mode the default.
rwcarlsen: I'm not even sure that a "strict" mode is preferred as a default.  What is implemented here basically is strict when they (charmers) want it strict (e.g. when running charm code in hooks, etc.) and unrestricted when they don't want restrictions (e.g. when setting up a test, running assertions, etc.).
PietroPasotti: > I'm not even sure that a "strict" mode is preferred as a default. What is implemented here basically is strict when they (charmers) want it strict (e.g. when running charm code in hooks, etc.) and unrestricted when they don't want restrictions (e.g. when setting up a test, running assertions, etc.).

Right, maybe the best we can aim for is a good doc explaining how to enable this strict mode when you want to mock an event handler call or so

Patch:
diff --git a/ops/framework.py b/ops/framework.py
index abe4e2c..28a4c29 100755
--- a/ops/framework.py
+++ b/ops/framework.py
@@ -28,6 +28,7 @@ import sys
 import types
 import typing
 import weakref
+from contextlib import contextmanager
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -847,34 +848,33 @@ class Framework(Object):
         """
         self._reemit()
 
-    def _reemit(self, single_event_path: str = None):
-
-        class EventContext:
-            """Handles toggling the hook-is-running state in backends.
+    @contextmanager
+    def _event_context(self, event_name: str):
+        """Handles toggling the hook-is-running state in backends.
 
-            This allows e.g. harness logic to know if it is executing within a running hook context
-            or not.  It sets backend._hook_is_running equal to the name of the currently running
-            hook (e.g. "set-leader") and reverts back to the empty string when the hook execution
-            is completed.
-            """
+        This allows e.g. harness logic to know if it is executing within a running hook context
+        or not.  It sets backend._hook_is_running equal to the name of the currently running
+        hook (e.g. "set-leader") and reverts back to the empty string when the hook execution
+        is completed.
 
-            def __init__(self, framework: Framework, event_name: str):
-                self._event = event_name
-                backend = None
-                if framework.model is not None:
-                    backend = framework.model._backend  # noqa
-                self._backend = backend  # type: Optional[_ModelBackend]
+        Usage:
+            >>> with harness._event_context('db-relation-changed'):
+            >>>     print('Harness thinks it is running an event hook.')
+            >>> with harness._event_context(''):
+            >>>     print('harness thinks it is not running an event hook.')
+        """
+        backend = self.model._backend if self.model else None  # type: Optional[_ModelBackend]
 
-            def __enter__(self):
-                if self._backend:
-                    self._backend._hook_is_running = self._event
-                return self
+        if not backend:
+            yield  # context does nothing in this case
+            return
 
-            def __exit__(self, exception_type: 'Type[Exception]',
-                         exception: Exception, traceback: Any):
-                if self._backend:
-                    self._backend._hook_is_running = ''
+        old = backend._hook_is_running
+        backend._hook_is_running = event_name
+        yield
+        backend._hook_is_running = old
 
+    def _reemit(self, single_event_path: str = None):
         last_event_path = None
         deferred = True
         for event_path, observer_path, method_name in self._storage.notices(single_event_path):
@@ -902,7 +902,7 @@ class Framework(Object):
                 if custom_handler:
                     event_is_from_juju = isinstance(event, charm.HookEvent)
                     event_is_action = isinstance(event, charm.ActionEvent)
-                    with EventContext(self, event_handle.kind):
+                    with self._event_context(event_handle.kind):
                         if (
                             event_is_from_juju or event_is_action
                         ) and self._juju_debug_at.intersection({'all', 'hook'}):
diff --git a/ops/model.py b/ops/model.py
index 4743b7e..605df64 100644
--- a/ops/model.py
+++ b/ops/model.py
@@ -915,6 +915,13 @@ class RelationDataContent(LazyMapping, MutableMapping[str, str]):
         self._backend = backend
         self._is_app = isinstance(entity, Application)  # type: bool
 
+    @property
+    def _hook_is_running(self) -> bool:
+        # this flag controls whether the access we have to RelationDataContent
+        # is 'strict' aka the same as a deployed charm would have, or whether it is
+        # unrestricted, allowing test code to read/write databags at will.
+        return bool(self._backend._hook_is_running)  # pyright: reportPrivateUsage=false
+
     def _load(self) -> '_RelationDataContent_Raw':
         """Load the data from the current entity / relation."""
         try:
@@ -923,35 +930,110 @@ class RelationDataContent(LazyMapping, MutableMapping[str, str]):
             # Dead relations tell no tales (and have no data).
             return {}
 
-    def _is_mutable(self):
-        """Return if the data content can be modified."""
+    def _validate_read(self):
+        """Return if the data content can be read."""
+        # if we're not in production (we're testing): we skip access control rules
+        if not self._hook_is_running:
+            return
+
+        # Only remote units (and the leader unit) can read *this* app databag.
+
+        # is this an app databag?
+        if not self._is_app:
+            # all unit databags are publicly readable
+            return
+
+        # Am I leader?
+        if self._backend.is_leader():
+            # leaders have no read restrictions
+            return
+
+        # type guard; we should not be accessing relation data
+        # if the remote app does not exist.
+        app = self.relation.app
+        if app is None:
+            raise RelationDataAccessError(
+                "Remote application instance cannot be retrieved for {}.".format(
+                    self.relation
+                )
+            )
+
+        # is this a peer relation?
+        if app.name == self._entity.name:
+            # peer relation data is always publicly readable
+            return
+
+        # if we're here it means: this is not a peer relation,
+        # this is an app databag, and we don't have leadership.
+
+        # is this a LOCAL app databag?
+        if self._backend.app_name == self._entity.name:
+            # minions can't read local app databags
+            raise RelationDataAccessError(
+                "{} is not leader and cannot read its own application databag".format(
+                    self._backend.unit_name
+                )
+            )
+
+        return True
+
+    def _validate_write(self, key: str, value: str):
+        """Validate writing key:value to this databag.
+
+        1) that key: value is a valid str:str pair
+        2) that we have write access to this databag
+        """
+        # firstly, we validate WHAT we're trying to write.
+        # this is independent of whether we're in testing code or production.
+        if not isinstance(key, str):
+            raise RelationDataTypeError(
+                'relation data keys must be strings, not {}'.format(type(key)))
+        if not isinstance(value, str):
+            raise RelationDataTypeError(
+                'relation data values must be strings, not {}'.format(type(value)))
+
+        # if we're not in production (we're testing): we skip access control rules
+        if not self._hook_is_running:
+            return
+
+        # finally, we check whether we have permissions to write this databag
         if self._is_app:
             is_our_app = self._backend.app_name == self._entity.name  # type: bool
             if not is_our_app:
-                return False
+                raise RelationDataAccessError(
+                    "{} cannot write the data of remote application {}".format(
+                        self._backend.app_name, self._entity.name
+                    ))
             # Whether the application data bag is mutable or not depends on
             # whether this unit is a leader or not, but this is not guaranteed
             # to be always true during the same hook execution.
-            return self._backend.is_leader()
+            if self._backend.is_leader():
+                return  # all good
+            raise RelationDataAccessError(
+                "{} is not leader and cannot write application data.".format(
+                    self._backend.unit_name
+                )
+            )
         else:
-            is_our_unit = self._backend.unit_name == self._entity.name
-            if is_our_unit:
-                return True
-        return False
+            # we are attempting to write a unit databag
+            # is it OUR UNIT's?
+            if self._backend.unit_name != self._entity.name:
+                raise RelationDataAccessError(
+                    "{} cannot write databag of {}: not the same unit.".format(
+                        self._backend.unit_name, self._entity.name
+                    )
+                )
 
     def __setitem__(self, key: str, value: str):
-        if not self._is_mutable():
-            raise RelationDataError(
-                'cannot set relation data for {}'.format(self._entity.name))
-        if not isinstance(key, str):
-            raise RelationDataError(
-                'relation data keys must be strings, not {}'.format(type(key)))
-        if not isinstance(value, str):
-            raise RelationDataError(
-                'relation data values must be strings, not {}'.format(type(value)))
+        self._validate_write(key, value)
+        self._commit(key, value)
+        self._update(key, value)
 
-        self._backend.relation_set(self.relation.id, key, value, self._is_app)
+    def _commit(self, key: str, value: str):
+        self._backend.update_relation_data(self.relation.id, self._entity, key, value)
 
+    def _update(self, key: str, value: str):
+        """Cache key:value in our local lazy data."""
         # Don't load data unnecessarily if we're only updating.
         if self._lazy_data is not None:
             if value == '':
@@ -961,11 +1043,23 @@ class RelationDataContent(LazyMapping, MutableMapping[str, str]):
             else:
                 self._data[key] = value
 
+    def __getitem__(self, key: str) -> str:
+        self._validate_read()
+        return super().__getitem__(key)
+
     def __delitem__(self, key: str):
+        self._validate_write(key, '')
         # Match the behavior of Juju, which is that setting the value to an empty
         # string will remove the key entirely from the relation data.
         self.__setitem__(key, '')
 
+    def __repr__(self):
+        try:
+            self._validate_read()
+        except RelationDataAccessError:
+            return '<n/a>'
+        return super().__repr__()
+
 
 class ConfigData(LazyMapping):
     """Configuration data.
@@ -1922,11 +2016,25 @@ class TooManyRelatedAppsError(ModelError):
 
 
 class RelationDataError(ModelError):
-    """Raised by ``Relation.data[entity][key] = 'foo'`` if the data is invalid.
+    """Raised when a relation data read/write is invalid.
 
     This is raised if you're either trying to set a value to something that isn't a string,
     or if you are trying to set a value in a bucket that you don't have access to. (eg,
     another application/unit or setting your application data but you aren't the leader.)
+    Also raised when you attempt to read a databag you don't have access to
+    (i.e. a local app databag if you're not the leader).
+    """
+
+
+class RelationDataTypeError(RelationDataError):
+    """Raised by ``Relation.data[entity][key] = value`` if `key` or `value` are not strings."""
+
+
+class RelationDataAccessError(RelationDataError):
+    """Raised by ``Relation.data[entity][key] = value`` if you don't have access.
+
+    This typically means that you don't have permission to write read/write the databag,
+    but in some cases it is raised when attempting to read/write from a deceased remote entity.
     """
 
 
@@ -2368,6 +2476,10 @@ class _ModelBackend:
         # Planned units can be zero. We don't need to do error checking here.
         return len(app_state.get('units', []))
 
+    def update_relation_data(self, relation_id: int, _entity: 'UnitOrApplication',
+                             key: str, value: str):
+        self.relation_set(relation_id, key, value, isinstance(_entity, Application))
+
 
 class _ModelBackendValidator:
     """Provides facilities for validating inputs and formatting them for model backends."""


Test patch:
diff --git a/ops/testing.py b/ops/testing.py
index d32cf07..a01189f 100755
--- a/ops/testing.py
+++ b/ops/testing.py
@@ -46,9 +46,14 @@ import warnings
 from contextlib import contextmanager
 from io import BytesIO, StringIO
 from textwrap import dedent
+from typing import TYPE_CHECKING, Dict
 
 from ops import charm, framework, model, pebble, storage
 from ops._private import yaml
+from ops.model import RelationNotFoundError
+
+if TYPE_CHECKING:
+    from ops.model import UnitOrApplication
 
 # Toggles Container.can_connect simulation globally for all harness instances.
 # For this to work, it must be set *before* Harness instances are created.
@@ -113,7 +118,6 @@ class Harness(typing.Generic[CharmType]):
         self._charm_dir = 'no-disk-path'  # this may be updated by _create_meta
         self._meta = self._create_meta(meta, actions)
         self._unit_name = self._meta.name + '/0'
-        self._framework = None
         self._hooks_enabled = True
         self._relation_id_counter = 0
         config_ = self._get_config(config)
@@ -132,6 +136,57 @@ class Harness(typing.Generic[CharmType]):
                 'Please set ops.testing.SIMULATE_CAN_CONNECT=True.'
                 'See https://juju.is/docs/sdk/testing#heading--simulate-can-connect for details.')
 
+    def _event_context(self, event_name: str):
+        """Configures the Harness to behave as if an event hook were running.
+
+        This means that the Harness will perform strict access control of relation data.
+
+        Example usage:
+
+        # this is how we test that attempting to write a remote app's
+        # databag will raise RelationDataError.
+        >>> with harness._event_context('foo'):
+        >>>     with pytest.raises(ops.model.RelationDataError):
+        >>>         my_relation.data[remote_app]['foo'] = 'bar'
+
+        # this is how we test with 'realistic conditions' how an event handler behaves
+        # when we call it directly -- i.e. without going through harness.add_relation
+        >>> def test_foo():
+        >>>     class MyCharm:
+        >>>         ...
+        >>>         def event_handler(self, event):
+        >>>             # this is expected to raise an exception
+        >>>             event.relation.data[event.relation.app]['foo'] = 'bar'
+        >>>
+        >>>     harness = Harness(MyCharm)
+        >>>     event = MagicMock()
+        >>>     event.relation = harness.charm.model.relations[0]
+        >>>
+        >>>     with harness._event_context('my_relation_joined'):
+        >>>         with pytest.raises(ops.model.RelationDataError):
+        >>>             harness.charm.event_handler(event)
+
+
+        If event_name == '', conversely, the Harness will believe that no hook
+        is running, allowing you to temporarily have unrestricted access to read/write
+        a relation's databags even if you're inside an event handler.
+        >>> def test_foo():
+        >>>     class MyCharm:
+        >>>         ...
+        >>>         def event_handler(self, event):
+        >>>             # this is expected to raise an exception since we're not leader
+        >>>             event.relation.data[self.app]['foo'] = 'bar'
+        >>>
+        >>>     harness = Harness(MyCharm)
+        >>>     event = MagicMock()
+        >>>     event.relation = harness.charm.model.relations[0]
+        >>>
+        >>>     with harness._event_context('my_relation_joined'):
+        >>>         harness.charm.event_handler(event)
+
+        """
+        return self._framework._event_context(event_name)
+
     def set_can_connect(self, container: typing.Union[str, model.Container], val: bool):
         """Change the simulated can_connect status of a container's underlying pebble client.
 
@@ -276,7 +331,7 @@ class Harness(typing.Generic[CharmType]):
             # relation-joined for the same unit.
             # Juju only fires relation-changed (app) if there is data for the related application
             relation = self._model.get_relation(rel_name, rel_id)
-            if self._backend._relation_data[rel_id].get(app_name):
+            if self._backend._relation_data_raw[rel_id].get(app_name):
                 app = self._model.get_app(app_name)
                 self._charm.on[rel_name].relation_changed.emit(
                     relation, app, None)
@@ -561,24 +616,24 @@ class Harness(typing.Generic[CharmType]):
         Return:
             The relation_id created by this add_relation.
         """
-        rel_id = self._next_relation_id()
-        self._backend._relation_ids_map.setdefault(relation_name, []).append(rel_id)
-        self._backend._relation_names[rel_id] = relation_name
-        self._backend._relation_list_map[rel_id] = []
-        self._backend._relation_data[rel_id] = {
-            remote_app: _TestingRelationDataContents(),
-            self._backend.unit_name: _TestingRelationDataContents(),
-            self._backend.app_name: _TestingRelationDataContents(),
-        }
-        self._backend._relation_app_and_units[rel_id] = {
+        relation_id = self._next_relation_id()
+        self._backend._relation_ids_map.setdefault(relation_name, []).append(relation_id)
+        self._backend._relation_names[relation_id] = relation_name
+        self._backend._relation_list_map[relation_id] = []
+        self._backend._relation_data_raw[relation_id] = {
+            remote_app: {},
+            self._backend.unit_name: {},
+            self._backend.app_name: {}}
+
+        self._backend._relation_app_and_units[relation_id] = {
             "app": remote_app,
             "units": [],
         }
         # Reload the relation_ids list
         if self._model is not None:
             self._model.relations._invalidate(relation_name)
-        self._emit_relation_created(relation_name, rel_id, remote_app)
-        return rel_id
+        self._emit_relation_created(relation_name, relation_id, remote_app)
+        return relation_id
 
     def remove_relation(self, relation_id: int) -> None:
         """Remove a relation.
@@ -603,7 +658,7 @@ class Harness(typing.Generic[CharmType]):
             self._model.relations._invalidate(relation_name)
 
         self._backend._relation_app_and_units.pop(relation_id)
-        self._backend._relation_data.pop(relation_id)
+        self._backend._relation_data_raw.pop(relation_id)
         self._backend._relation_list_map.pop(relation_id)
         self._backend._relation_ids_map[relation_name].remove(relation_id)
         self._backend._relation_names.pop(relation_id)
@@ -650,16 +705,20 @@ class Harness(typing.Generic[CharmType]):
             None
         """
         self._backend._relation_list_map[relation_id].append(remote_unit_name)
-        self._backend._relation_data[relation_id][
-            remote_unit_name] = _TestingRelationDataContents()
-        # TODO: jam 2020-08-03 This is where we could assert that the unit name matches the
-        #  application name (eg you don't have a relation to 'foo' but add units of 'bar/0'
-        self._backend._relation_app_and_units[relation_id]["units"].append(remote_unit_name)
+        # we can write remote unit data iff we are not in a hook env
         relation_name = self._backend._relation_names[relation_id]
+        relation = self._model.get_relation(relation_name, relation_id)
+        self._backend._relation_data_raw[relation_id][remote_unit_name] = {}
+        if not remote_unit_name.startswith(relation.app.name):
+            raise ValueError(
+                'Remote unit name invalid: the remote application of {} is called {!r}; '
+                'the remote unit name should be {}/<some-number>, not {!r}.'
+                ''.format(relation_name, relation.app.name,
+                          relation.app.name, remote_unit_name))
+        self._backend._relation_app_and_units[relation_id]["units"].append(remote_unit_name)
         # Make sure that the Model reloads the relation_list for this relation_id, as well as
         # reloading the relation data for this unit.
         remote_unit = self._model.get_unit(remote_unit_name)
-        relation = self._model.get_relation(relation_name, relation_id)
         unit_cache = relation.data.get(remote_unit, None)
         if unit_cache is not None:
             unit_cache._invalidate()
@@ -709,7 +768,7 @@ class Harness(typing.Generic[CharmType]):
         # remove the relation data for the departed unit now that the event has happened
         self._backend._relation_list_map[relation_id].remove(remote_unit_name)
         self._backend._relation_app_and_units[relation_id]["units"].remove(remote_unit_name)
-        self._backend._relation_data[relation_id].pop(remote_unit_name)
+        self._backend._relation_data_raw[relation_id].pop(remote_unit_name)
         self.model._relations._invalidate(relation_name=relation.name)
 
         if unit_cache is not None:
@@ -747,7 +806,8 @@ class Harness(typing.Generic[CharmType]):
         """
         if hasattr(app_or_unit, 'name'):
             app_or_unit = app_or_unit.name
-        return self._backend._relation_data[relation_id].get(app_or_unit, None)
+        # bypass access control by going directly to raw
+        return self._backend._relation_data_raw[relation_id].get(app_or_unit, None)
 
     def get_pod_spec(self) -> (typing.Mapping, typing.Mapping):
         """Return the content of the pod spec as last set by the charm.
@@ -854,21 +914,21 @@ class Harness(typing.Generic[CharmType]):
             # Note, this won't cause the data to be loaded if it wasn't already.
             rel_data._invalidate()
 
-        new_values = self._backend._relation_data[relation_id][app_or_unit].copy()
-        assert isinstance(new_values, _TestingRelationDataContents), new_values
-        values_have_changed = False
-        for k, v in key_values.items():
-            if v == '':
-                if new_values.pop(k, None) != v:
-                    values_have_changed = True
-            else:
-                if k not in new_values or new_values[k] != v:
-                    new_values[k] = v
-                    values_have_changed = True
-
-        # Update the relation data in any case to avoid spurious references
-        # by an test to an updated value to be invalidated by a lack of assignment
-        self._backend._relation_data[relation_id][app_or_unit] = new_values
+        old_values = self._backend._relation_data_raw[relation_id][app_or_unit].copy()
+        assert isinstance(old_values, dict), old_values
+
+        databag = self.model.relations._get_unique(relation.name, relation_id).data[entity]
+        # ensure that WE as harness can temporarily write the databag
+        with self._event_context(''):
+            values_have_changed = False
+            for k, v in key_values.items():
+                if v == '':
+                    if databag.pop(k, None) != v:
+                        values_have_changed = True
+                else:
+                    if k not in databag or databag[k] != v:
+                        databag[k] = v  # this triggers relation-set
+                        values_have_changed = True
 
         if not values_have_changed:
             # Do not issue a relation changed event if the data bags have not changed
@@ -1179,7 +1239,8 @@ class _TestingModelBackend:
         self._relation_ids_map = {}  # relation name to [relation_ids,...]
         self._relation_names = {}  # reverse map from relation_id to relation_name
         self._relation_list_map = {}  # relation_id: [unit_name,...]
-        self._relation_data = {}  # {relation_id: {name: data}}
+        # {relation_id: {name: Dict[str: str]}}
+        self._relation_data_raw = {}  # type: Dict[int, Dict[str, Dict[str, str]]]
         # {relation_id: {"app": app_name, "units": ["app/0",...]}
         self._relation_app_and_units = {}
         self._config = _TestingConfig(config)
@@ -1282,9 +1343,18 @@ class _TestingModelBackend:
                 'remote-side relation data cannot be accessed during a relation-broken event')
         if is_app and '/' in member_name:
             member_name = member_name.split('/')[0]
-        if relation_id not in self._relation_data:
+        if relation_id not in self._relation_data_raw:
             raise model.RelationNotFoundError()
-        return self._relation_data[relation_id][member_name].copy()
+        return self._relation_data_raw[relation_id][member_name]
+
+    def update_relation_data(self, relation_id: int, _entity: 'UnitOrApplication',
+                             key: str, value: str):
+        # this is where the 'real' backend would call relation-set.
+        raw_data = self._relation_data_raw[relation_id][_entity.name]
+        if value == '':
+            raw_data.pop(key, None)
+        else:
+            raw_data[key] = value
 
     def relation_set(self, relation_id: int, key: str, value: str, is_app: bool):
         if not isinstance(is_app, bool):
@@ -1295,7 +1365,10 @@ class _TestingModelBackend:
             raise RuntimeError(
                 'remote-side relation data cannot be accessed during a relation-broken event')
 
-        relation = self._relation_data[relation_id]
+        if relation_id not in self._relation_data_raw:
+            raise RelationNotFoundError(relation_id)
+
+        relation = self._relation_data_raw[relation_id]
         if is_app:
             bucket_key = self.app_name
         else:
diff --git a/test/test_model.py b/test/test_model.py
index 0d38eed..495859f 100755
--- a/test/test_model.py
+++ b/test/test_model.py
@@ -18,6 +18,7 @@ import ipaddress
 import json
 import os
 import pathlib
+import sys
 import tempfile
 import unittest
 from collections import OrderedDict
@@ -36,7 +37,6 @@ from ops.pebble import APIError, FileInfo, FileType, ServiceInfo
 
 
 class TestModel(unittest.TestCase):
-
     def setUp(self):
         self.harness = ops.testing.Harness(ops.charm.CharmBase, meta='''
             name: myapp
@@ -63,6 +63,7 @@ class TestModel(unittest.TestCase):
         ''')
         self.addCleanup(self.harness.cleanup)
         self.relation_id_db0 = self.harness.add_relation('db0', 'db')
+        self.harness._get_backend_calls(reset=True)
         self.model = self.harness.model
 
     def test_model_attributes(self):
@@ -182,24 +183,41 @@ class TestModel(unittest.TestCase):
         self.harness.add_relation_unit(relation_id, 'remoteapp1/0')
 
         with self.assertRaises(model.RelationDataError):
-            self.harness.update_relation_data(
-                relation_id,
-                'remoteapp1/0',
-                {42: 'remoteapp1-0'})
+            with self.harness._event_context('foo_event'):
+                self.harness.update_relation_data(
+                    relation_id,
+                    'remoteapp1/0',
+                    {42: 'remoteapp1-0'})
 
         with self.assertRaises(model.RelationDataError):
+            with self.harness._event_context('foo_event'):
+                self.harness.update_relation_data(
+                    relation_id,
+                    'remoteapp1/0',
+                    {'foo': 42})
+
+    def test_get_app_relation_data(self):
+        self.harness.begin()
+        relation_id = self.harness.add_relation('db1', 'remote')
+        self.harness.add_relation_unit(relation_id, 'remote/0')
+        local_app = self.harness.model.app.name
+        with self.harness._event_context('foo_event'):
             self.harness.update_relation_data(
                 relation_id,
-                'remoteapp1/0',
-                {'foo': 42})
+                local_app,
+                {'foo': 'bar'})
+            assert self.harness.get_relation_data(
+                relation_id, self.harness.model.app) == self.harness.get_relation_data(
+                relation_id, local_app) == {'foo': 'bar'}
 
     def test_unit_relation_data(self):
         relation_id = self.harness.add_relation('db1', 'remoteapp1')
         self.harness.add_relation_unit(relation_id, 'remoteapp1/0')
-        self.harness.update_relation_data(
-            relation_id,
-            'remoteapp1/0',
-            {'host': 'remoteapp1-0'})
+        with self.harness._event_context('foo_event'):
+            self.harness.update_relation_data(
+                relation_id,
+                'remoteapp1/0',
+                {'host': 'remoteapp1-0'})
         self.model.relations._invalidate('db1')
         self.resetBackendCalls()
 
@@ -219,7 +237,9 @@ class TestModel(unittest.TestCase):
 
     def test_remote_app_relation_data(self):
         relation_id = self.harness.add_relation('db1', 'remoteapp1')
-        self.harness.update_relation_data(relation_id, 'remoteapp1', {'secret': 'cafedeadbeef'})
+        with self.harness._event_context('foo_event'):
+            self.harness.update_relation_data(relation_id, 'remoteapp1',
+                                              {'secret': 'cafedeadbeef'})
         self.harness.add_relation_unit(relation_id, 'remoteapp1/0')
         self.harness.add_relation_unit(relation_id, 'remoteapp1/1')
         self.resetBackendCalls()
@@ -243,9 +263,12 @@ class TestModel(unittest.TestCase):
 
     def test_relation_data_modify_remote(self):
         relation_id = self.harness.add_relation('db1', 'remoteapp1')
-        self.harness.update_relation_data(relation_id, 'remoteapp1', {'secret': 'cafedeadbeef'})
-        self.harness.add_relation_unit(relation_id, 'remoteapp1/0')
-        self.harness.update_relation_data(relation_id, 'remoteapp1/0', {'host': 'remoteapp1/0'})
+        with self.harness._event_context('foo_event'):
+            self.harness.update_relation_data(relation_id, 'remoteapp1',
+                                              {'secret': 'cafedeadbeef'})
+            self.harness.add_relation_unit(relation_id, 'remoteapp1/0')
+            self.harness.update_relation_data(relation_id, 'remoteapp1/0',
+                                              {'host': 'remoteapp1/0'})
         self.model.relations._invalidate('db1')
         self.resetBackendCalls()
 
@@ -256,8 +279,9 @@ class TestModel(unittest.TestCase):
         self.assertIn('host', rel_db1.data[remoteapp1_0])
         self.assertEqual(repr(rel_db1.data[remoteapp1_0]), "{'host': 'remoteapp1/0'}")
 
-        with self.assertRaises(ops.model.RelationDataError):
-            rel_db1.data[remoteapp1_0]['foo'] = 'bar'
+        with self.harness._event_context('foo_event'):
+            with self.assertRaises(ops.model.RelationDataError):
+                rel_db1.data[remoteapp1_0]['foo'] = 'bar'
         self.assertNotIn('foo', rel_db1.data[remoteapp1_0])
 
         self.assertBackendCalls([
@@ -267,29 +291,35 @@ class TestModel(unittest.TestCase):
         ])
 
         # this will fire more backend calls
+        with self.harness._event_context('foo_event'):
+            data_repr = repr(rel_db1.data)
         # the CountEqual and weird (and brittle) splitting is to accommodate python 3.5
         # TODO: switch to assertEqual when we drop 3.5
         self.assertCountEqual(
-            repr(rel_db1.data)[1:-1].split(', '),
+            data_repr[1:-1].split(', '),
             ["<ops.model.Unit myapp/0>: {}",
-             "<ops.model.Application myapp>: {}",
+             "<ops.model.Application myapp>: <n/a>",
              "<ops.model.Unit remoteapp1/0>: {'host': 'remoteapp1/0'}",
              "<ops.model.Application remoteapp1>: {'secret': 'cafedeadbeef'}"])
 
     def test_relation_data_modify_our(self):
         relation_id = self.harness.add_relation('db1', 'remoteapp1')
+
         self.harness.update_relation_data(relation_id, 'myapp/0', {'host': 'nothing'})
         self.resetBackendCalls()
-
-        rel_db1 = self.model.get_relation('db1')
-        # Force memory cache to be loaded.
-        self.assertIn('host', rel_db1.data[self.model.unit])
-        rel_db1.data[self.model.unit]['host'] = 'bar'
-        self.assertEqual(rel_db1.data[self.model.unit]['host'], 'bar')
+        with self.harness._event_context('foo_event'):
+            rel_db1 = self.model.get_relation('db1')
+            # update_relation_data will also trigger relation-get, so we
+            # invalidate the cache to ensure it will be reloaded
+            rel_db1.data[self.model.unit]._invalidate()
+            # Force memory cache to be loaded.
+            self.assertIn('host', rel_db1.data[self.model.unit])
+            rel_db1.data[self.model.unit]['host'] = 'bar'
+            self.assertEqual(rel_db1.data[self.model.unit]['host'], 'bar')
 
         self.assertBackendCalls([
             ('relation_get', relation_id, 'myapp/0', False),
-            ('relation_set', relation_id, 'host', 'bar', False),
+            ('update_relation_data', relation_id, self.model.unit, 'host', 'bar'),
         ])
 
     def test_app_relation_data_modify_local_as_leader(self):
@@ -308,13 +338,11 @@ class TestModel(unittest.TestCase):
 
         self.assertEqual(rel_db1.data[local_app]['password'], 'foo')
 
-        self.assertBackendCalls([
-            ('relation_ids', 'db1'),
-            ('relation_list', relation_id),
-            ('relation_get', relation_id, 'myapp', True),
-            ('is_leader',),
-            ('relation_set', relation_id, 'password', 'foo', True),
-        ])
+        self.assertBackendCalls(
+            [('relation_ids', 'db1'),
+             ('relation_list', 1),
+             ('relation_get', 1, 'myapp', True),
+             ('update_relation_data', 1, self.model.app, 'password', 'foo')])
 
     def test_app_relation_data_modify_local_as_minion(self):
         relation_id = self.harness.add_relation('db1', 'remoteapp1')
@@ -328,19 +356,40 @@ class TestModel(unittest.TestCase):
         rel_db1 = self.model.get_relation('db1')
         self.assertEqual(rel_db1.data[local_app], {'password': 'deadbeefcafe'})
 
-        with self.assertRaises(ops.model.RelationDataError):
-            rel_db1.data[local_app]['password'] = 'foobar'
-
-        self.assertBackendCalls([
-            ('relation_ids', 'db1'),
-            ('relation_list', relation_id),
-            ('relation_get', relation_id, 'myapp', True),
-            ('is_leader',),
-        ])
+        with self.harness._event_context('foo_event'):
+            # if we were inside an event context, we'd get:
+            with self.assertRaises(ops.model.RelationDataError):
+                rel_db1.data[local_app]['password'] = 'foobar'
+
+        self.assertBackendCalls([('relation_ids', 'db1'),
+                                 ('relation_list', 1),
+                                 ('relation_get', 1, 'myapp', True),
+                                 ('is_leader',)])
+
+    def test_relation_data_access_peer_leader(self):
+        r_id = self.harness.add_relation('db2', 'myapp')
+        self.harness.add_relation_unit(r_id, 'myapp/1')  # peer!
+        self.harness.update_relation_data(r_id, 'myapp', {'foo': 'bar'})
+        with self.harness._event_context('foo_event'):
+            # leaders can read
+            self.harness.set_leader(True)
+            relation = self.harness.model.get_relation('db2')
+            self.assertEqual(relation.data[relation.app]['foo'], 'bar')
+
+    def test_relation_data_access_peer_minion(self):
+        r_id = self.harness.add_relation('db2', 'myapp')
+        self.harness.add_relation_unit(r_id, 'myapp/1')  # peer!
+        self.harness.update_relation_data(r_id, 'myapp', {'foo': 'bar'})
+        with self.harness._event_context('foo_event'):
+            # nonleaders can read
+            self.harness.set_leader(False)
+            relation = self.harness.model.get_relation('db2')
+            self.assertEqual(relation.data[relation.app]['foo'], 'bar')
 
     def test_relation_data_del_key(self):
         relation_id = self.harness.add_relation('db1', 'remoteapp1')
-        self.harness.update_relation_data(relation_id, 'myapp/0', {'host': 'bar'})
+        with self.harness._event_context('foo_event'):
+            self.harness.update_relation_data(relation_id, 'myapp/0', {'host': 'bar'})
         self.harness.add_relation_unit(relation_id, 'remoteapp1/0')
         self.resetBackendCalls()
 
@@ -355,32 +404,37 @@ class TestModel(unittest.TestCase):
             ('relation_ids', 'db1'),
             ('relation_list', relation_id),
             ('relation_get', relation_id, 'myapp/0', False),
-            ('relation_set', relation_id, 'host', '', False),
+            ('update_relation_data', relation_id, self.model.unit, 'host', ''),
         ])
 
     def test_relation_data_del_missing_key(self):
         relation_id = self.harness.add_relation('db1', 'remoteapp1')
-        self.harness.update_relation_data(relation_id, 'myapp/0', {'host': 'bar'})
+        with self.harness._event_context('foo_event'):
+            self.harness.update_relation_data(relation_id, 'myapp/0', {'host': 'bar'})
         self.harness.add_relation_unit(relation_id, 'remoteapp1/0')
         self.resetBackendCalls()
 
         rel_db1 = self.model.get_relation('db1')
         # Force memory cache to be loaded.
         self.assertIn('host', rel_db1.data[self.model.unit])
-        rel_db1.data[self.model.unit]['port'] = ''   # Same as a delete, should not fail.
+        with self.harness._event_context('foo_event'):
+            rel_db1.data[self.model.unit]['port'] = ''   # Same as a delete, should not fail.
         self.assertNotIn('port', rel_db1.data[self.model.unit])
-        self.assertEqual({'host': 'bar'}, self.harness.get_relation_data(relation_id, 'myapp/0'))
+        with self.harness._event_context('foo_event'):
+            self.assertEqual({'host': 'bar'},
+                             self.harness.get_relation_data(relation_id, 'myapp/0'))
 
         self.assertBackendCalls([
             ('relation_ids', 'db1'),
             ('relation_list', relation_id),
             ('relation_get', relation_id, 'myapp/0', False),
-            ('relation_set', relation_id, 'port', '', False),
+            ('update_relation_data', relation_id, self.model.unit, 'port', ''),
         ])
 
     def test_relation_set_fail(self):
         relation_id = self.harness.add_relation('db1', 'remoteapp1')
-        self.harness.update_relation_data(relation_id, 'myapp/0', {'host': 'myapp-0'})
+        with self.harness._event_context('foo_event'):
+            self.harness.update_relation_data(relation_id, 'myapp/0', {'host': 'myapp-0'})
         self.harness.add_relation_unit(relation_id, 'remoteapp1/0')
         self.resetBackendCalls()
 
@@ -390,27 +444,29 @@ class TestModel(unittest.TestCase):
         #       but there was nothing illegal about the data that was being set,
         #       for us to properly test the side effects of relation-set failing.
 
-        def broken_relation_set(relation_id, key, value, is_app):
-            backend._calls.append(('relation_set', relation_id, key, value, is_app))
+        def broken_update_relation_data(relation_id, entity, key, value):
+            backend._calls.append(('update_relation_data', relation_id, entity, key, value))
             raise ops.model.ModelError()
-        backend.relation_set = broken_relation_set
+        backend.update_relation_data = broken_update_relation_data
 
         rel_db1 = self.model.get_relation('db1')
         # Force memory cache to be loaded.
         self.assertIn('host', rel_db1.data[self.model.unit])
-        with self.assertRaises(ops.model.ModelError):
-            rel_db1.data[self.model.unit]['host'] = 'bar'
-        self.assertEqual(rel_db1.data[self.model.unit]['host'], 'myapp-0')
-        with self.assertRaises(ops.model.ModelError):
-            del rel_db1.data[self.model.unit]['host']
-        self.assertIn('host', rel_db1.data[self.model.unit])
+
+        with self.harness._event_context('foo_event'):
+            with self.assertRaises(ops.model.ModelError):
+                rel_db1.data[self.model.unit]['host'] = 'bar'
+            self.assertEqual(rel_db1.data[self.model.unit]['host'], 'myapp-0')
+            with self.assertRaises(ops.model.ModelError):
+                del rel_db1.data[self.model.unit]['host']
+            self.assertIn('host', rel_db1.data[self.model.unit])
 
         self.assertBackendCalls([
             ('relation_ids', 'db1'),
             ('relation_list', relation_id),
             ('relation_get', relation_id, 'myapp/0', False),
-            ('relation_set', relation_id, 'host', 'bar', False),
-            ('relation_set', relation_id, 'host', '', False),
+            ('update_relation_data', relation_id, self.model.unit, 'host', 'bar'),
+            ('update_relation_data', relation_id, self.model.unit, 'host', ''),
         ])
 
     def test_relation_data_type_check(self):
@@ -431,7 +487,8 @@ class TestModel(unittest.TestCase):
                 (None, None)
         ):
             with self.assertRaises(ops.model.RelationDataError):
-                rel_db1.data[self.model.unit][key] = value
+                with self.harness.framework._event_context('foo_event'):
+                    rel_db1.data[self.model.unit][key] = value
 
         # No data has actually been changed
         self.assertEqual(dict(rel_db1.data[self.model.unit]), {'host': 'myapp-0'})
@@ -442,6 +499,104 @@ class TestModel(unittest.TestCase):
             ('relation_get', relation_id, 'myapp/0', False),
         ])
 
+    def test_relation_local_app_data_readability_leader(self):
+        relation_id = self.harness.add_relation('db1', 'remoteapp1')
+        self.harness.update_relation_data(relation_id, 'remoteapp1',
+                                          {'secret': 'cafedeadbeef'})
+        self.harness.update_relation_data(relation_id, 'myapp',
+                                          {'local': 'data'})
+
+        self.harness.add_relation_unit(relation_id, 'remoteapp1/0')
+        self.harness.update_relation_data(relation_id, 'remoteapp1/0',
+                                          {'host': 'remoteapp1/0'})
+        self.model.relations._invalidate('db1')
+        self.resetBackendCalls()
+
+        rel_db1 = self.model.get_relation('db1')
+        self.harness.begin()
+        self.harness.set_leader(True)
+        self.resetBackendCalls()
+
+        local_app = self.harness.charm.app
+        self.resetBackendCalls()
+
+        # addressing the object is OK
+        rel_db1.data[local_app]
+
+        self.assertBackendCalls([])
+
+        with self.harness._event_context('foo_event'):
+            self.resetBackendCalls()
+
+            self.assertEqual(rel_db1.data[local_app]['local'], 'data')
+
+            self.assertBackendCalls([('is_leader',),
+                                     ('relation_get', 1, 'myapp', True)])
+
+            self.resetBackendCalls()
+
+            self.assertEqual(repr(rel_db1.data[local_app]), repr({'local': 'data'}))
+
+            # we don't get the data, because we're lazy
+            self.assertBackendCalls([('is_leader',)])
+
+            # as well as relation data repr() in general:
+            self.assertIsInstance(repr(rel_db1.data), str)
+
+    def test_relation_local_app_data_readability_follower(self):
+        relation_id = self.harness.add_relation('db1', 'remoteapp1')
+        with self.harness._event_context('foo_event'):
+            self.harness.update_relation_data(relation_id, 'remoteapp1',
+                                              {'secret': 'cafedeadbeef'})
+            self.harness.update_relation_data(relation_id, 'myapp',
+                                              {'local': 'data'})
+
+            self.harness.add_relation_unit(relation_id, 'remoteapp1/0')
+            self.harness.update_relation_data(relation_id, 'remoteapp1/0',
+                                              {'host': 'remoteapp1/0'})
+        self.model.relations._invalidate('db1')
+        self.resetBackendCalls()
+
+        rel_db1 = self.model.get_relation('db1')
+        self.harness.begin()
+        self.harness.set_leader(False)
+
+        local_app = self.harness.charm.app
+        # addressing the object is OK
+        rel_db1.data[local_app]
+        # nonleader units cannot read their local app databag
+        # attempting to read it is not
+        with self.harness._event_context('foo_event'):
+            self.resetBackendCalls()
+
+            with self.assertRaises(ops.model.RelationDataError):
+                # 'local' is there, but still:
+                rel_db1.data[local_app]['local']
+
+            # we didn't even get to relation-get
+            self.assertBackendCalls([('is_leader', )])
+
+            # we can't see it but repr() works
+            self.assertEqual(repr(rel_db1.data[local_app]), '<n/a>')
+            self.assertBackendCalls([('is_leader', )])
+
+            # as well as relation data repr() in general:
+            self.assertIsInstance(repr(rel_db1.data), str)
+
+            expected_backend_calls = [
+                ('relation_get', 1, 'myapp/0', False),
+                ('is_leader',),
+                ('relation_get', 1, 'remoteapp1/0', False),
+                ('is_leader',),
+                ('relation_get', 1, 'remoteapp1', True)]
+            # in < 3.5 dicts are unsorted
+            major, minor, *_ = sys.version_info
+            if (major, minor) > (3, 5):
+                self.assertBackendCalls(expected_backend_calls)
+            else:
+                backend_calls = set(self.harness._get_backend_calls())
+                self.assertEqual(backend_calls, set(expected_backend_calls))
+
     def test_relation_no_units(self):
         self.harness.add_relation('db1', 'remoteapp1')
         rel = self.model.get_relation('db1')
@@ -454,6 +609,7 @@ class TestModel(unittest.TestCase):
         ])
 
     def test_config(self):
+        self.harness._get_backend_calls(reset=True)
         self.harness.update_config({'foo': 'foo', 'bar': 1, 'qux': True})
         self.assertEqual(self.model.config, {
             'foo': 'foo',
@@ -588,6 +744,7 @@ class TestModel(unittest.TestCase):
         self.assertEqual(ops.model.ActiveStatus().message, '')
 
     def test_local_set_valid_unit_status(self):
+        self.harness._get_backend_calls(reset=True)
         test_cases = [(
             'active',
             ops.model.ActiveStatus('Green'),
diff --git a/test/test_testing.py b/test/test_testing.py
index 248eb3f..983a81b 100644
--- a/test/test_testing.py
+++ b/test/test_testing.py
@@ -22,9 +22,11 @@ import shutil
 import sys
 import tempfile
 import textwrap
+import typing
 import unittest
 import uuid
 from io import BytesIO, StringIO
+from unittest.mock import MagicMock
 
 import pytest
 import yaml
@@ -39,7 +41,7 @@ from ops.charm import (
     StorageAttachedEvent,
     StorageDetachingEvent,
 )
-from ops.framework import Object
+from ops.framework import Framework, Object
 from ops.model import (
     ActiveStatus,
     Application,
@@ -493,8 +495,7 @@ class TestHarness(unittest.TestCase):
         self.assertEqual(len(harness.charm.model.get_relation('db').units), 0)
         self.assertFalse(rel_unit in harness.charm.model.get_relation('db').data)
         # Check relation departed was raised with correct data
-        self.assertEqual(harness.charm.get_changes()[0],
-                         {'name': 'relation-departed',
+        self.assertEqual({'name': 'relation-departed',
                           'relation': 'db',
                           'data': {'app': 'postgresql',
                                    'unit': 'postgresql/0',
@@ -503,7 +504,8 @@ class TestHarness(unittest.TestCase):
                                    'relation_data': {'test-app/0': {},
                                                      'test-app': {},
                                                      'postgresql/0': {'foo': 'bar'},
-                                                     'postgresql': {}}}})
+                                                     'postgresql': {}}}},
+                         harness.charm.get_changes()[0])
 
     def test_removing_relation_removes_remote_app_data(self):
         # language=YAML
@@ -528,8 +530,9 @@ class TestHarness(unittest.TestCase):
         harness.remove_relation(rel_id)
         # Check relation and app data are removed
         self.assertEqual(backend.relation_ids('db'), [])
-        self.assertRaises(RelationNotFoundError, backend.relation_get,
-                          rel_id, remote_app, is_app=True)
+        with harness._event_context('foo'):
+            self.assertRaises(RelationNotFoundError, backend.relation_get,
+                              rel_id, remote_app, is_app=True)
 
     def test_removing_relation_refreshes_charm_model(self):
         # language=YAML
@@ -1149,8 +1152,9 @@ class TestHarness(unittest.TestCase):
         rel_id = harness.add_relation('db', 'postgresql')
         harness.add_relation_unit(rel_id, 'postgresql/0')
         rel = harness.charm.model.get_relation('db')
-        with self.assertRaises(ModelError):
-            rel.data[harness.charm.app]['foo'] = 'bar'
+        with harness._event_context('foo'):
+            with self.assertRaises(ModelError):
+                rel.data[harness.charm.app]['foo'] = 'bar'
         # The data has not actually been changed
         self.assertEqual(harness.get_relation_data(rel_id, 'test-charm'), {})
         harness.set_leader(True)
@@ -1754,6 +1758,68 @@ class TestHarness(unittest.TestCase):
         self.addCleanup(harness.cleanup)
         self.assertEqual(list(harness.framework.meta.actions), ['test-action'])
 
+    def test_event_context(self):
+        class MyCharm(CharmBase):
+            def event_handler(self, evt):
+                evt.relation.data[evt.relation.app]['foo'] = 'bar'
+
+        harness = Harness(MyCharm, meta='''
+            name: test-charm
+            requires:
+                db:
+                    interface: pgsql
+            ''')
+        harness.begin()
+        rel_id = harness.add_relation('db', 'postgresql')
+        rel = harness.charm.model.get_relation('db', rel_id)
+
+        event = MagicMock()
+        event.relation = rel
+
+        with harness._event_context('my_relation_joined'):
+            with self.assertRaises(ops.model.RelationDataError):
+                harness.charm.event_handler(event)
+
+    def test_event_context_inverse(self):
+        class MyCharm(CharmBase):
+            def __init__(self, framework: Framework,
+                         key: typing.Optional = None):
+                super().__init__(framework, key)
+                self.framework.observe(self.on.db_relation_joined,
+                                       self._join_db)
+
+            def _join_db(self, event):
+                # do things with APIs we cannot easily mock
+                raise NotImplementedError
+
+        harness = Harness(MyCharm, meta='''
+            name: test-charm
+            requires:
+                db:
+                    interface: pgsql
+            ''')
+        harness.begin()
+
+        def mock_join_db(event):
+            # the harness thinks we're inside a db_relation_joined hook
+            # but we want to mock the remote data here:
+            with harness._event_context(''):
+                # pretend for a moment we're not in a hook context,
+                # so the harness will let us:
+                print(event.relation.app)
+                event.relation.data[harness.charm.app]['foo'] = 'bar'
+
+        harness.charm._join_db = mock_join_db
+        rel_id = harness.add_relation('db', 'remote')
+        harness.add_relation_unit(rel_id, 'remote/0')
+        rel = harness.charm.model.get_relation('db', rel_id)
+        self.assertEqual({'foo': 'bar'},
+                         harness.get_relation_data(rel_id, 'test-charm'))
+
+        # now we're outside of the hook context:
+        assert not harness._backend._hook_is_running
+        assert rel.data[harness.charm.app]['foo'] == 'bar'
+
     def test_relation_set_deletes(self):
         harness = Harness(CharmBase, meta='''
             name: test-charm
@@ -1809,29 +1875,45 @@ class TestHarness(unittest.TestCase):
         # No calls to the backend yet
         self.assertEqual(harness._get_backend_calls(), [])
         rel_id = harness.add_relation('db', 'postgresql')
-        # update_relation_data ensures the cached data for the relation is wiped
-        harness.update_relation_data(rel_id, 'test-charm/0', {'foo': 'bar'})
+
         self.assertEqual(
-            harness._get_backend_calls(reset=True), [
+            [
                 ('relation_ids', 'db'),
                 ('relation_list', rel_id),
                 ('relation_remote_app_name', 0),
-            ])
+            ],
+            harness._get_backend_calls())
+
+        # update_relation_data ensures the cached data for the relation is wiped
+        harness.update_relation_data(rel_id, 'test-charm/0', {'foo': 'bar'})
+        test_charm_unit = harness.model.get_unit('test-charm/0')
+        self.assertEqual(
+            [
+                ('relation_get', 0, 'test-charm/0', False),
+                ('update_relation_data', 0, test_charm_unit, 'foo', 'bar')
+            ],
+            harness._get_backend_calls(reset=True))
         # add_relation_unit resets the relation_list, but doesn't trigger backend calls
         harness.add_relation_unit(rel_id, 'postgresql/0')
         self.assertEqual([], harness._get_backend_calls(reset=False))
         # however, update_relation_data does, because we are preparing relation-changed
         harness.update_relation_data(rel_id, 'postgresql/0', {'foo': 'bar'})
+        pgql_unit = harness.model.get_unit('postgresql/0')
+
         self.assertEqual(
             harness._get_backend_calls(reset=False), [
                 ('relation_ids', 'db'),
                 ('relation_list', rel_id),
+                ('relation_get', 0, 'postgresql/0', False),
+                ('update_relation_data', 0, pgql_unit, 'foo', 'bar')
             ])
         # If we check again, they are still there, but now we reset it
         self.assertEqual(
             harness._get_backend_calls(reset=True), [
                 ('relation_ids', 'db'),
                 ('relation_list', rel_id),
+                ('relation_get', 0, 'postgresql/0', False),
+                ('update_relation_data', 0, pgql_unit, 'foo', 'bar')
             ])
         # And the calls are gone
         self.assertEqual(harness._get_backend_calls(), [])
@@ -2281,7 +2363,6 @@ class TestHarness(unittest.TestCase):
         harness.update_relation_data(rel_id, 'postgresql', {'app': 'data'})
         harness.begin_with_initial_hooks()
         self.assertEqual(
-            harness.charm.changes,
             [
                 {'name': 'install'},
                 {'name': 'relation-created',
@@ -2315,7 +2396,8 @@ class TestHarness(unittest.TestCase):
                      'unit': 'postgresql/0',
                      'app': 'postgresql',
                  }},
-            ])
+            ],
+            harness.charm.changes)
 
     def test_begin_with_initial_hooks_with_multiple_units(self):
         class CharmWithDB(RelationEventCharm):
@@ -2886,6 +2968,9 @@ class TestTestingModelBackend(unittest.TestCase):
     def test_relation_remote_app_name(self):
         harness = Harness(CharmBase, meta='''
             name: test-charm
+            requires:
+               db:
+                 interface: foo
             ''')
         self.addCleanup(harness.cleanup)
         backend = harness._backend