# Copyright 2021 Canonical Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Infrastructure to build unittests for Charms using the Operator Framework."""

import datetime
import inspect
import pathlib
import random
import tempfile
import typing
from contextlib import contextmanager
from textwrap import dedent

from ops import (
    charm,
    framework,
    model,
    pebble,
    storage,
)
from ops._private import yaml


# OptionalYAML is something like metadata.yaml or actions.yaml. You can
# pass in a file-like object or the string directly.
OptionalYAML = typing.Optional[typing.Union[str, typing.TextIO]]


# noinspection PyProtectedMember
class Harness:
    """This class represents a way to build up the model that will drive a test suite.

    The model that is created is from the viewpoint of the charm that you are testing.

    Example::

        harness = Harness(MyCharm)
        # Do initial setup here
        relation_id = harness.add_relation('db', 'postgresql')
        # Now instantiate the charm to see events as the model changes
        harness.begin()
        harness.add_relation_unit(relation_id, 'postgresql/0')
        harness.update_relation_data(relation_id, 'postgresql/0', {'key': 'val'})
        # Check that charm has properly handled the relation_joined event for postgresql/0
        self.assertEqual(harness.charm. ...)

    Args:
        charm_cls: The Charm class that you'll be testing.
        meta: charm.CharmBase is a A string or file-like object containing the contents of
            metadata.yaml. If not supplied, we will look for a 'metadata.yaml' file in the
            parent directory of the Charm, and if not found fall back to a trivial
            'name: test-charm' metadata.
        actions: A string or file-like object containing the contents of
            actions.yaml. If not supplied, we will look for a 'actions.yaml' file in the
            parent directory of the Charm.
        config: A string or file-like object containing the contents of
            config.yaml. If not supplied, we will look for a 'config.yaml' file in the
            parent directory of the Charm.
    """

    def __init__(
            self,
            charm_cls: typing.Type[charm.CharmBase],
            *,
            meta: OptionalYAML = None,
            actions: OptionalYAML = None,
            config: OptionalYAML = None):
        self._charm_cls = charm_cls
        self._charm = None
        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
        self._backend = _TestingModelBackend(self._unit_name, self._meta)
        self._model = model.Model(self._meta, self._backend)
        self._storage = storage.SQLiteStorage(':memory:')
        self._oci_resources = {}
        self._framework = framework.Framework(
            self._storage, self._charm_dir, self._meta, self._model)
        self._update_config(key_values=self._load_config_defaults(config))

    @property
    def charm(self) -> charm.CharmBase:
        """Return the instance of the charm class that was passed to __init__.

        Note that the Charm is not instantiated until you have called
        :meth:`.begin()`.
        """
        return self._charm

    @property
    def model(self) -> model.Model:
        """Return the :class:`~ops.model.Model` that is being driven by this Harness."""
        return self._model

    @property
    def framework(self) -> framework.Framework:
        """Return the Framework that is being driven by this Harness."""
        return self._framework

    def begin(self) -> None:
        """Instantiate the Charm and start handling events.

        Before calling :meth:`begin`, there is no Charm instance, so changes to the Model won't
        emit events. You must call :meth:`.begin` before :attr:`.charm` is valid.
        """
        if self._charm is not None:
            raise RuntimeError('cannot call the begin method on the harness more than once')

        # The Framework adds attributes to class objects for events, etc. As such, we can't re-use
        # the original class against multiple Frameworks. So create a locally defined class
        # and register it.
        # TODO: jam 2020-03-16 We are looking to changes this to Instance attributes instead of
        #       Class attributes which should clean up this ugliness. The API can stay the same
        class TestEvents(self._charm_cls.on.__class__):
            pass

        TestEvents.__name__ = self._charm_cls.on.__class__.__name__

        class TestCharm(self._charm_cls):
            on = TestEvents()

        # Note: jam 2020-03-01 This is so that errors in testing say MyCharm has no attribute foo,
        # rather than TestCharm has no attribute foo.
        TestCharm.__name__ = self._charm_cls.__name__
        self._charm = TestCharm(self._framework)

    def begin_with_initial_hooks(self) -> None:
        """Called when you want the Harness to fire the same hooks that Juju would fire at startup.

        This triggers install, relation-created, config-changed, start, and any relation-joined
        hooks. Based on what relations have been defined before you called begin().
        Note that all of these are fired before returning control to the test suite, so if you
        want to introspect what happens at each step, you need to fire them directly
        (eg Charm.on.install.emit()).

        To use this with all the normal hooks, you should instantiate the harness, setup any
        relations that you want active when the charm starts, and then call this method.

        Example::

            harness = Harness(MyCharm)
            # Do initial setup here
            relation_id = harness.add_relation('db', 'postgresql')
            harness.add_relation_unit(relation_id, 'postgresql/0')
            harness.update_relation_data(relation_id, 'postgresql/0', {'key': 'val'})
            harness.set_leader(True)
            harness.update_config({'initial': 'config'})
            harness.begin_with_initial_hooks()
            # This will cause
            # install, db-relation-created('postgresql'), leader-elected, config-changed, start
            # db-relation-joined('postrgesql/0'), db-relation-changed('postgresql/0')
            # To be fired.
        """
        self.begin()
        # TODO: jam 2020-08-03 This should also handle storage-attached hooks once we have support
        #  for dealing with storage.
        self._charm.on.install.emit()
        # Juju itself iterates what relation to fire based on a map[int]relation, so it doesn't
        # guarantee a stable ordering between relation events. It *does* give a stable ordering
        # of joined units for a given relation.
        items = list(self._meta.relations.items())
        random.shuffle(items)
        this_app_name = self._meta.name
        for relname, rel_meta in items:
            if rel_meta.role == charm.RelationRole.peer:
                # If the user has directly added a relation, leave it be, but otherwise ensure
                # that peer relations are always established at before leader-elected.
                rel_ids = self._backend._relation_ids_map.get(relname)
                if rel_ids is None:
                    self.add_relation(relname, self._meta.name)
                else:
                    random.shuffle(rel_ids)
                    for rel_id in rel_ids:
                        self._emit_relation_created(relname, rel_id, this_app_name)
            else:
                rel_ids = self._backend._relation_ids_map.get(relname, [])
                random.shuffle(rel_ids)
                for rel_id in rel_ids:
                    app_name = self._backend._relation_app_and_units[rel_id]["app"]
                    self._emit_relation_created(relname, rel_id, app_name)
        if self._backend._is_leader:
            self._charm.on.leader_elected.emit()
        else:
            self._charm.on.leader_settings_changed.emit()
        self._charm.on.config_changed.emit()
        self._charm.on.start.emit()
        all_ids = list(self._backend._relation_names.items())
        random.shuffle(all_ids)
        for rel_id, rel_name in all_ids:
            rel_app_and_units = self._backend._relation_app_and_units[rel_id]
            app_name = rel_app_and_units["app"]
            # Note: Juju *does* fire relation events for a given relation in the sorted order of
            # the unit names. It also always fires relation-changed immediately after
            # 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):
                app = self._model.get_app(app_name)
                self._charm.on[rel_name].relation_changed.emit(
                    relation, app, None)
            for unit_name in sorted(rel_app_and_units["units"]):
                remote_unit = self._model.get_unit(unit_name)
                self._charm.on[rel_name].relation_joined.emit(
                    relation, remote_unit.app, remote_unit)
                self._charm.on[rel_name].relation_changed.emit(
                    relation, remote_unit.app, remote_unit)

    def cleanup(self) -> None:
        """Called by your test infrastructure to cleanup any temporary directories/files/etc.

        Currently this only needs to be called if you test with resources. But it is reasonable
        to always include a `testcase.addCleanup(harness.cleanup)` just in case.
        """
        self._backend._cleanup()

    def _create_meta(self, charm_metadata, action_metadata):
        """Create a CharmMeta object.

        Handle the cases where a user doesn't supply explicit metadata snippets.
        """
        filename = inspect.getfile(self._charm_cls)
        charm_dir = pathlib.Path(filename).parents[1]

        if charm_metadata is None:
            metadata_path = charm_dir / 'metadata.yaml'
            if metadata_path.is_file():
                charm_metadata = metadata_path.read_text()
                self._charm_dir = charm_dir
            else:
                # The simplest of metadata that the framework can support
                charm_metadata = 'name: test-charm'
        elif isinstance(charm_metadata, str):
            charm_metadata = dedent(charm_metadata)

        if action_metadata is None:
            actions_path = charm_dir / 'actions.yaml'
            if actions_path.is_file():
                action_metadata = actions_path.read_text()
                self._charm_dir = charm_dir
        elif isinstance(action_metadata, str):
            action_metadata = dedent(action_metadata)

        return charm.CharmMeta.from_yaml(charm_metadata, action_metadata)

    def _load_config_defaults(self, charm_config):
        """Load default values from config.yaml.

        Handle the case where a user doesn't supply explicit config snippets.
        """
        filename = inspect.getfile(self._charm_cls)
        charm_dir = pathlib.Path(filename).parents[1]

        if charm_config is None:
            config_path = charm_dir / 'config.yaml'
            if config_path.is_file():
                charm_config = config_path.read_text()
                self._charm_dir = charm_dir
            else:
                # The simplest of config that the framework can support
                charm_config = '{}'
        elif isinstance(charm_config, str):
            charm_config = dedent(charm_config)
        charm_config = yaml.safe_load(charm_config)
        charm_config = charm_config.get('options', {})
        return {key: value['default'] for key, value in charm_config.items()
                if 'default' in value}

    def add_oci_resource(self, resource_name: str,
                         contents: typing.Mapping[str, str] = None) -> None:
        """Add oci resources to the backend.

        This will register an oci resource and create a temporary file for processing metadata
        about the resource. A default set of values will be used for all the file contents
        unless a specific contents dict is provided.

        Args:
            resource_name: Name of the resource to add custom contents to.
            contents: Optional custom dict to write for the named resource.
        """
        if not contents:
            contents = {'registrypath': 'registrypath',
                        'username': 'username',
                        'password': 'password',
                        }
        if resource_name not in self._meta.resources.keys():
            raise RuntimeError('Resource {} is not a defined resources'.format(resource_name))
        if self._meta.resources[resource_name].type != "oci-image":
            raise RuntimeError('Resource {} is not an OCI Image'.format(resource_name))

        as_yaml = yaml.safe_dump(contents)
        self._backend._resources_map[resource_name] = ('contents.yaml', as_yaml)

    def add_resource(self, resource_name: str, content: typing.AnyStr) -> None:
        """Add content for a resource to the backend.

        This will register the content, so that a call to `Model.resources.fetch(resource_name)`
        will return a path to a file containing that content.

        Args:
            resource_name: The name of the resource being added
            content: Either string or bytes content, which will be the content of the filename
                returned by resource-get. If contents is a string, it will be encoded in utf-8
        """
        if resource_name not in self._meta.resources.keys():
            raise RuntimeError('Resource {} is not a defined resources'.format(resource_name))
        record = self._meta.resources[resource_name]
        if record.type != "file":
            raise RuntimeError(
                'Resource {} is not a file, but actually {}'.format(resource_name, record.type))
        filename = record.filename
        if filename is None:
            filename = resource_name

        self._backend._resources_map[resource_name] = (filename, content)

    def populate_oci_resources(self) -> None:
        """Populate all OCI resources."""
        for name, data in self._meta.resources.items():
            if data.type == "oci-image":
                self.add_oci_resource(name)

    def disable_hooks(self) -> None:
        """Stop emitting hook events when the model changes.

        This can be used by developers to stop changes to the model from emitting events that
        the charm will react to. Call :meth:`.enable_hooks`
        to re-enable them.
        """
        self._hooks_enabled = False

    def enable_hooks(self) -> None:
        """Re-enable hook events from charm.on when the model is changed.

        By default hook events are enabled once you call :meth:`.begin`,
        but if you have used :meth:`.disable_hooks`, this can be used to
        enable them again.
        """
        self._hooks_enabled = True

    @contextmanager
    def hooks_disabled(self):
        """A context manager to run code with hooks disabled.

        Example::

            with harness.hooks_disabled():
                # things in here don't fire events
                harness.set_leader(True)
                harness.update_config(unset=['foo', 'bar'])
            # things here will again fire events
        """
        if self._hooks_enabled:
            self.disable_hooks()
            try:
                yield None
            finally:
                self.enable_hooks()
        else:
            yield None

    def _next_relation_id(self):
        rel_id = self._relation_id_counter
        self._relation_id_counter += 1
        return rel_id

    def add_relation(self, relation_name: str, remote_app: str) -> int:
        """Declare that there is a new relation between this app and `remote_app`.

        Args:
            relation_name: The relation on Charm that is being related to
            remote_app: The name of the application that is being related to

        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: {},
            self._backend.unit_name: {},
            self._backend.app_name: {},
        }
        self._backend._relation_app_and_units[rel_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

    def _emit_relation_created(self, relation_name: str, relation_id: int,
                               remote_app: str) -> None:
        """Trigger relation-created for a given relation with a given remote application."""
        if self._charm is None or not self._hooks_enabled:
            return
        if self._charm is None or not self._hooks_enabled:
            return
        relation = self._model.get_relation(relation_name, relation_id)
        app = self._model.get_app(remote_app)
        self._charm.on[relation_name].relation_created.emit(
            relation, app)

    def add_relation_unit(self, relation_id: int, remote_unit_name: str) -> None:
        """Add a new unit to a relation.

        Example::

          rel_id = harness.add_relation('db', 'postgresql')
          harness.add_relation_unit(rel_id, 'postgresql/0')

        This will trigger a `relation_joined` event. This would naturally be
        followed by a `relation_changed` event, which you can trigger with
        :meth:`.update_relation_data`. This separation is artificial in the
        sense that Juju will always fire the two, but is intended to make
        testing relations and their data bags slightly more natural.

        Args:
            relation_id: The integer relation identifier (as returned by add_relation).
            remote_unit_name: A string representing the remote unit that is being added.

        Return:
            None
        """
        self._backend._relation_list_map[relation_id].append(remote_unit_name)
        self._backend._relation_data[relation_id][remote_unit_name] = {}
        # 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)
        relation_name = self._backend._relation_names[relation_id]
        # Make sure that the Model reloads the relation_list for this relation_id, as well as
        # reloading the relation data for this unit.
        if self._model is not None:
            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()
            self._model.relations._invalidate(relation_name)
        if self._charm is None or not self._hooks_enabled:
            return
        self._charm.on[relation_name].relation_joined.emit(
            relation, remote_unit.app, remote_unit)

    def get_relation_data(self, relation_id: int, app_or_unit: str) -> typing.Mapping:
        """Get the relation data bucket for a single app or unit in a given relation.

        This ignores all of the safety checks of who can and can't see data in relations (eg,
        non-leaders can't read their own application's relation data because there are no events
        that keep that data up-to-date for the unit).

        Args:
            relation_id: The relation whose content we want to look at.
            app_or_unit: The name of the application or unit whose data we want to read
        Return:
            A dict containing the relation data for `app_or_unit` or None.

        Raises:
            KeyError: if relation_id doesn't exist
        """
        return self._backend._relation_data[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.

        This returns both the pod spec and any k8s_resources that were supplied.
        See the signature of Model.pod.set_spec
        """
        return self._backend._pod_spec

    def get_container_pebble_plan(
            self, container_name: str
    ) -> pebble.Plan:
        """Return the current Plan that pebble is executing for the given container.

        Args:
            container_name: The simple name of the associated container
        Return:
            The pebble.Plan for this container. You can use :meth:`pebble.Plan.to_yaml` to get
            a string form for the content. Will raise KeyError if no pebble client exists
            for that container name. (should only happen if container is not present in
            metadata.yaml)
        """
        socket_path = '/charm/containers/{}/pebble.socket'.format(container_name)
        client = self._backend._pebble_clients.get(socket_path)
        if client is None:
            raise KeyError('no known pebble client for container "{}"'.format(container_name))
        return client.get_plan()

    def get_workload_version(self) -> str:
        """Read the workload version that was set by the unit."""
        return self._backend._workload_version

    def set_model_name(self, name: str) -> None:
        """Set the name of the Model that this is representing.

        This cannot be called once begin() has been called. But it lets you set the value that
        will be returned by Model.name.
        """
        if self._charm is not None:
            raise RuntimeError('cannot set the Model name after begin()')
        self._backend.model_name = name

    def update_relation_data(
            self,
            relation_id: int,
            app_or_unit: str,
            key_values: typing.Mapping,
    ) -> None:
        """Update the relation data for a given unit or application in a given relation.

        This also triggers the `relation_changed` event for this relation_id.

        Args:
            relation_id: The integer relation_id representing this relation.
            app_or_unit: The unit or application name that is being updated.
                This can be the local or remote application.
            key_values: Each key/value will be updated in the relation data.
        """
        relation_name = self._backend._relation_names[relation_id]
        relation = self._model.get_relation(relation_name, relation_id)
        if '/' in app_or_unit:
            entity = self._model.get_unit(app_or_unit)
        else:
            entity = self._model.get_app(app_or_unit)
        rel_data = relation.data.get(entity, None)
        if rel_data is not None:
            # rel_data may have cached now-stale data, so _invalidate() it.
            # 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()
        for k, v in key_values.items():
            if v == '':
                new_values.pop(k, None)
            else:
                new_values[k] = v
        self._backend._relation_data[relation_id][app_or_unit] = new_values

        if app_or_unit == self._model.unit.name:
            # No events for our own unit
            return
        if app_or_unit == self._model.app.name:
            # updating our own app only generates an event if it is a peer relation and we
            # aren't the leader
            is_peer = self._meta.relations[relation_name].role.is_peer()
            if not is_peer:
                return
            if self._model.unit.is_leader():
                return
        self._emit_relation_changed(relation_id, app_or_unit)

    def _emit_relation_changed(self, relation_id, app_or_unit):
        if self._charm is None or not self._hooks_enabled:
            return
        rel_name = self._backend._relation_names[relation_id]
        relation = self.model.get_relation(rel_name, relation_id)
        if '/' in app_or_unit:
            app_name = app_or_unit.split('/')[0]
            unit_name = app_or_unit
            app = self.model.get_app(app_name)
            unit = self.model.get_unit(unit_name)
            args = (relation, app, unit)
        else:
            app_name = app_or_unit
            app = self.model.get_app(app_name)
            args = (relation, app)
        self._charm.on[rel_name].relation_changed.emit(*args)

    def _update_config(
            self,
            key_values: typing.Mapping[str, str] = None,
            unset: typing.Iterable[str] = (),
    ) -> None:
        """Update the config as seen by the charm.

        This will *not* trigger a `config_changed` event, and is intended for internal use.

        Note that the `key_values` mapping will only add or update configuration items.
        To remove existing ones, see the `unset` parameter.

        Args:
            key_values: A Mapping of key:value pairs to update in config.
            unset: An iterable of keys to remove from Config. (Note that this does
                not currently reset the config values to the default defined in config.yaml.)
        """
        # NOTE: jam 2020-03-01 Note that this sort of works "by accident". Config
        # is a LazyMapping, but its _load returns a dict and this method mutates
        # the dict that Config is caching. Arguably we should be doing some sort
        # of charm.framework.model.config._invalidate()
        config = self._backend._config
        if key_values is not None:
            for key, value in key_values.items():
                config[key] = value
        for key in unset:
            config.pop(key, None)

    def update_config(
            self,
            key_values: typing.Mapping[str, str] = None,
            unset: typing.Iterable[str] = (),
    ) -> None:
        """Update the config as seen by the charm.

        This will trigger a `config_changed` event.

        Note that the `key_values` mapping will only add or update configuration items.
        To remove existing ones, see the `unset` parameter.

        Args:
            key_values: A Mapping of key:value pairs to update in config.
            unset: An iterable of keys to remove from Config. (Note that this does
                not currently reset the config values to the default defined in config.yaml.)
        """
        self._update_config(key_values, unset)
        if self._charm is None or not self._hooks_enabled:
            return
        self._charm.on.config_changed.emit()

    def set_leader(self, is_leader: bool = True) -> None:
        """Set whether this unit is the leader or not.

        If this charm becomes a leader then `leader_elected` will be triggered.

        Args:
            is_leader: True/False as to whether this unit is the leader.
        """
        was_leader = self._backend._is_leader
        self._backend._is_leader = is_leader
        # Note: jam 2020-03-01 currently is_leader is cached at the ModelBackend level, not in
        # the Model objects, so this automatically gets noticed.
        if is_leader and not was_leader and self._charm is not None and self._hooks_enabled:
            self._charm.on.leader_elected.emit()

    def _get_backend_calls(self, reset: bool = True) -> list:
        """Return the calls that we have made to the TestingModelBackend.

        This is useful mostly for testing the framework itself, so that we can assert that we
        do/don't trigger extra calls.

        Args:
            reset: If True, reset the calls list back to empty, if false, the call list is
                preserved.

        Return:
            ``[(call1, args...), (call2, args...)]``
        """
        calls = self._backend._calls.copy()
        if reset:
            self._backend._calls.clear()
        return calls


def _record_calls(cls):
    """Replace methods on cls with methods that record that they have been called.

    Iterate all attributes of cls, and for public methods, replace them with a wrapped method
    that records the method called along with the arguments and keyword arguments.
    """
    for meth_name, orig_method in cls.__dict__.items():
        if meth_name.startswith('_'):
            continue

        def decorator(orig_method):
            def wrapped(self, *args, **kwargs):
                full_args = (orig_method.__name__,) + args
                if kwargs:
                    full_args = full_args + (kwargs,)
                self._calls.append(full_args)
                return orig_method(self, *args, **kwargs)
            return wrapped

        setattr(cls, meth_name, decorator(orig_method))
    return cls


def _copy_docstrings(source_cls):
    """Copy the docstrings from source_cls to target_cls.

    Use this as:
      @_copy_docstrings(source_class)
      class TargetClass:

    And for any public method that exists on both classes, it will copy the
    __doc__ for that method.
    """
    def decorator(target_cls):
        for meth_name, orig_method in target_cls.__dict__.items():
            if meth_name.startswith('_'):
                continue
            source_method = source_cls.__dict__.get(meth_name)
            if source_method is not None and source_method.__doc__:
                target_cls.__dict__[meth_name].__doc__ = source_method.__doc__
        return target_cls
    return decorator


class _ResourceEntry:
    """Tracks the contents of a Resource."""

    def __init__(self, resource_name):
        self.name = resource_name


@_copy_docstrings(model._ModelBackend)
@_record_calls
class _TestingModelBackend:
    """This conforms to the interface for ModelBackend but provides canned data.

    DO NOT use this class directly, it is used by `Harness`_ to drive the model.
    `Harness`_ is responsible for maintaining the internal consistency of the values here,
    as the only public methods of this type are for implementing ModelBackend.
    """

    def __init__(self, unit_name, meta):
        self.unit_name = unit_name
        self.app_name = self.unit_name.split('/')[0]
        self.model_name = None
        self._calls = []
        self._meta = meta
        self._is_leader = None
        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: {"app": app_name, "units": ["app/0",...]}
        self._relation_app_and_units = {}
        self._config = {}
        self._is_leader = False
        self._resources_map = {}  # {resource_name: resource_content}
        self._pod_spec = None
        self._app_status = {'status': 'unknown', 'message': ''}
        self._unit_status = {'status': 'maintenance', 'message': ''}
        self._workload_version = None
        self._resource_dir = None
        # {socket_path : _TestingPebbleClient}
        # socket_path = '/charm/containers/{container_name}/pebble.socket'
        self._pebble_clients = {}  # type: {str: _TestingPebbleClient}

    def _cleanup(self):
        if self._resource_dir is not None:
            self._resource_dir.cleanup()
            self._resource_dir = None

    def _get_resource_dir(self) -> pathlib.Path:
        if self._resource_dir is None:
            # In actual Juju, the resource path for a charm's resource is
            # $AGENT_DIR/resources/$RESOURCE_NAME/$RESOURCE_FILENAME
            # However, charms shouldn't depend on this.
            self._resource_dir = tempfile.TemporaryDirectory(prefix='tmp-ops-test-resource-')
        return pathlib.Path(self._resource_dir.name)

    def relation_ids(self, relation_name):
        try:
            return self._relation_ids_map[relation_name]
        except KeyError as e:
            if relation_name not in self._meta.relations:
                raise model.ModelError('{} is not a known relation'.format(relation_name)) from e
            return []

    def relation_list(self, relation_id):
        try:
            return self._relation_list_map[relation_id]
        except KeyError as e:
            raise model.RelationNotFoundError from e

    def relation_remote_app_name(self, relation_id: int) -> typing.Optional[str]:
        if relation_id not in self._relation_app_and_units:
            # Non-existent or dead relation
            return None
        return self._relation_app_and_units[relation_id]['app']

    def relation_get(self, relation_id, member_name, is_app):
        if is_app and '/' in member_name:
            member_name = member_name.split('/')[0]
        if relation_id not in self._relation_data:
            raise model.RelationNotFoundError()
        return self._relation_data[relation_id][member_name].copy()

    def relation_set(self, relation_id, key, value, is_app):
        relation = self._relation_data[relation_id]
        if is_app:
            bucket_key = self.app_name
        else:
            bucket_key = self.unit_name
        if bucket_key not in relation:
            relation[bucket_key] = {}
        bucket = relation[bucket_key]
        if value == '':
            bucket.pop(key, None)
        else:
            bucket[key] = value

    def config_get(self):
        return self._config

    def is_leader(self):
        return self._is_leader

    def application_version_set(self, version):
        self._workload_version = version

    def resource_get(self, resource_name):
        if resource_name not in self._resources_map:
            raise model.ModelError(
                "ERROR could not download resource: HTTP request failed: "
                "Get https://.../units/unit-{}/resources/{}: resource#{}/{} not found".format(
                    self.unit_name.replace('/', '-'), resource_name, self.app_name, resource_name
                ))
        filename, contents = self._resources_map[resource_name]
        resource_dir = self._get_resource_dir()
        resource_filename = resource_dir / resource_name / filename
        if not resource_filename.exists():
            if isinstance(contents, bytes):
                mode = 'wb'
            else:
                mode = 'wt'
            resource_filename.parent.mkdir(exist_ok=True)
            with resource_filename.open(mode=mode) as resource_file:
                resource_file.write(contents)
        return resource_filename

    def pod_spec_set(self, spec, k8s_resources):
        self._pod_spec = (spec, k8s_resources)

    def status_get(self, *, is_app=False):
        if is_app:
            return self._app_status
        else:
            return self._unit_status

    def status_set(self, status, message='', *, is_app=False):
        if is_app:
            self._app_status = {'status': status, 'message': message}
        else:
            self._unit_status = {'status': status, 'message': message}

    def storage_list(self, name):
        raise NotImplementedError(self.storage_list)

    def storage_get(self, storage_name_id, attribute):
        raise NotImplementedError(self.storage_get)

    def storage_add(self, name, count=1):
        raise NotImplementedError(self.storage_add)

    def action_get(self):
        raise NotImplementedError(self.action_get)

    def action_set(self, results):
        raise NotImplementedError(self.action_set)

    def action_log(self, message):
        raise NotImplementedError(self.action_log)

    def action_fail(self, message=''):
        raise NotImplementedError(self.action_fail)

    def network_get(self, endpoint_name, relation_id=None):
        raise NotImplementedError(self.network_get)

    def add_metrics(self, metrics, labels=None):
        raise NotImplementedError(self.add_metrics)

    def juju_log(self, level, msg):
        raise NotImplementedError(self.juju_log)

    def get_pebble(self, socket_path: str):
        client = self._pebble_clients.get(socket_path, None)
        if client is None:
            client = _TestingPebbleClient(self)
            self._pebble_clients[socket_path] = client
        return client


@_copy_docstrings(pebble.Client)
class _TestingPebbleClient:
    """This conforms to the interface for pebble.Client but provides canned data.

    DO NOT use this class directly, it is used by `Harness`_ to run interactions with Pebble.
    `Harness`_ is responsible for maintaining the internal consistency of the values here,
    as the only public methods of this type are for implementing Client.
    """

    def __init__(self, backend: _TestingModelBackend):
        self._backend = _TestingModelBackend
        self._layers = {}
        # Has a service been started/stopped?
        self._service_status = {}

    def get_system_info(self) -> pebble.SystemInfo:
        raise NotImplementedError(self.get_system_info)

    def get_warnings(
            self, select: pebble.WarningState = pebble.WarningState.PENDING,
    ) -> typing.List['pebble.Warning']:
        raise NotImplementedError(self.get_warnings)

    def ack_warnings(self, timestamp: datetime.datetime) -> int:
        raise NotImplementedError(self.ack_warnings)

    def get_changes(
            self, select: pebble.ChangeState = pebble.ChangeState.IN_PROGRESS, service: str = None,
    ) -> typing.List[pebble.Change]:
        raise NotImplementedError(self.get_changes)

    def get_change(self, change_id: pebble.ChangeID) -> pebble.Change:
        raise NotImplementedError(self.get_change)

    def abort_change(self, change_id: pebble.ChangeID) -> pebble.Change:
        raise NotImplementedError(self.abort_change)

    def autostart_services(self, timeout: float = 30.0, delay: float = 0.1) -> pebble.ChangeID:
        for name, service in self._render_services().items():
            # TODO: jam 2021-04-20 This feels awkward that Service.startup might be a string or
            #  might be an enum. Probably should make Service.startup a property rather than an
            #  attribute.
            if service.startup == '':
                startup = pebble.ServiceStartup.DISABLED
            else:
                startup = pebble.ServiceStartup(service.startup)
            if startup == pebble.ServiceStartup.ENABLED:
                self._service_status[name] = pebble.ServiceStatus.ACTIVE

    def start_services(
            self, services: typing.List[str], timeout: float = 30.0, delay: float = 0.1,
    ) -> pebble.ChangeID:
        # A common mistake is to pass just the name of a service, rather than a list of services,
        # so trap that so it is caught quickly.
        if isinstance(services, str):
            raise TypeError('start_services should take a list of names, not just "{}"'.format(
                services))

        # Note: jam 2021-04-20 We don't implement ChangeID, but the default caller of this is
        # Container.start() which currently ignores the return value
        known_services = self._render_services()
        # Names appear to be validated before any are activated, so do two passes
        for name in services:
            if name not in known_services:
                # TODO: jam 2021-04-20 This needs a better error type
                raise RuntimeError('400 Bad Request: service "{}" does not exist'.format(name))
            current = self._service_status.get(name, pebble.ServiceStatus.INACTIVE)
            if current == pebble.ServiceStatus.ACTIVE:
                # TODO: jam 2021-04-20 I believe pebble actually validates all the service names
                #  can be started before starting any, and gives a list of things that couldn't
                #  be done, but this is good enough for now
                raise pebble.ChangeError('''\
cannot perform the following tasks:
- Start service "{}" (service "{}" was previously started)
'''.format(name, name), change=1234)  # the change id is not respected
        for name in services:
            # If you try to start a service which is started, you get a ChangeError:
            # $ PYTHONPATH=. python3 ./test/pebble_cli.py start serv
            # ChangeError: cannot perform the following tasks:
            # - Start service "serv" (service "serv" was previously started)
            self._service_status[name] = pebble.ServiceStatus.ACTIVE

    def stop_services(
            self, services: typing.List[str], timeout: float = 30.0, delay: float = 0.1,
    ) -> pebble.ChangeID:
        # handle a common mistake of passing just a name rather than a list of names
        if isinstance(services, str):
            raise TypeError('stop_services should take a list of names, not just "{}"'.format(
                services))
        # TODO: handle invalid names
        # Note: jam 2021-04-20 We don't implement ChangeID, but the default caller of this is
        # Container.stop() which currently ignores the return value
        known_services = self._render_services()
        for name in services:
            if name not in known_services:
                # TODO: jam 2021-04-20 This needs a better error type
                #  400 Bad Request: service "bal" does not exist
                raise RuntimeError('400 Bad Request: service "{}" does not exist'.format(name))
            current = self._service_status.get(name, pebble.ServiceStatus.INACTIVE)
            if current != pebble.ServiceStatus.ACTIVE:
                # TODO: jam 2021-04-20 I believe pebble actually validates all the service names
                #  can be started before starting any, and gives a list of things that couldn't
                #  be done, but this is good enough for now
                raise pebble.ChangeError('''\
ChangeError: cannot perform the following tasks:
- Stop service "{}" (service "{}" is not active)
'''.format(name, name), change=1234)  # the change id is not respected
        for name in services:
            self._service_status[name] = pebble.ServiceStatus.INACTIVE

    def wait_change(
            self, change_id: pebble.ChangeID, timeout: float = 30.0, delay: float = 0.1,
    ) -> pebble.Change:
        raise NotImplementedError(self.wait_change)

    def add_layer(
            self, label: str, layer: typing.Union[str, dict, pebble.Layer], *,
            combine: bool = False):
        # I wish we could combine some of this helpful object corralling with the actual backend,
        # rather than having to re-implement it. Maybe we could subclass
        if not isinstance(label, str):
            raise TypeError('label must be a str, not {}'.format(type(label).__name__))

        if isinstance(layer, (str, dict)):
            layer_obj = pebble.Layer(layer)
        elif isinstance(layer, pebble.Layer):
            layer_obj = layer
        else:
            raise TypeError('layer must be str, dict, or pebble.Layer, not {}'.format(
                type(layer).__name__))
        if label in self._layers:
            # TODO: jam 2021-04-19 These should not be RuntimeErrors but should be proper error
            #  types. https://github.com/canonical/operator/issues/514
            if not combine:
                raise RuntimeError('400 Bad Request: layer "{}" already exists'.format(label))
            layer = self._layers[label]
            for name, service in layer_obj.services.items():
                # 'override' is actually single quoted in the real error, but
                # it shouldn't be, hopefully that gets cleaned up.
                if not service.override:
                    raise RuntimeError('500 Internal Server Error: layer "{}" must define'
                                       '"override" for service "{}"'.format(label, name))
                if service.override not in ('merge', 'replace'):
                    raise RuntimeError('500 Internal Server Error: layer "{}" has invalid '
                                       '"override" value on service "{}"'.format(label, name))
                if service.override != 'replace':
                    raise RuntimeError(
                        'override: "{}" unsupported for layer "{}" service "{}"'.format(
                            service.override, label, name))
                layer.services[name] = service
        else:
            self._layers[label] = layer_obj

    def _render_services(self) -> typing.Mapping[str, pebble.Service]:
        services = {}
        for key in sorted(self._layers.keys()):
            layer = self._layers[key]
            for name, service in layer.services.items():
                # TODO: (jam) 2021-04-07 have a way to merge existing services
                services[name] = service
        return services

    def get_plan(self) -> pebble.Plan:
        plan = pebble.Plan('{}')
        services = self._render_services()
        if not services:
            return plan
        for name in sorted(services.keys()):
            plan.services[name] = services[name]
        return plan

    def get_services(self, names: typing.List[str] = None) -> typing.List[pebble.ServiceInfo]:
        if isinstance(names, str):
            raise TypeError('start_services should take a list of names, not just "{}"'.format(
                names))
        services = self._render_services()
        infos = []
        if names is None:
            names = sorted(services.keys())
        for name in sorted(names):
            try:
                service = services[name]
            except KeyError:
                # in pebble, it just returns "nothing matched" if there are 0 matches,
                # but it ignores services it doesn't recognize
                continue
            status = self._service_status.get(name, pebble.ServiceStatus.INACTIVE)
            if service.startup == '':
                startup = pebble.ServiceStartup.DISABLED
            else:
                startup = pebble.ServiceStartup(service.startup)
            info = pebble.ServiceInfo(name,
                                      startup=startup,
                                      current=pebble.ServiceStatus(status))
            infos.append(info)
        return infos

    def pull(self, path: str, *, encoding: str = 'utf-8') -> typing.Union[typing.BinaryIO,
                                                                          typing.TextIO]:
        raise NotImplementedError(self.pull)

    def push(
            self, path: str, source: typing.Union[bytes, str, typing.BinaryIO, typing.TextIO], *,
            encoding: str = 'utf-8', make_dirs: bool = False, permissions: int = None,
            user_id: int = None, user: str = None, group_id: int = None, group: str = None):
        raise NotImplementedError(self.push)

    def list_files(self, path: str, *, pattern: str = None,
                   itself: bool = False) -> typing.List[pebble.FileInfo]:
        raise NotImplementedError(self.list_files)

    def make_dir(
            self, path: str, *, make_parents: bool = False, permissions: int = None,
            user_id: int = None, user: str = None, group_id: int = None, group: str = None):
        raise NotImplementedError(self.make_dir)

    def remove_path(self, path: str, *, recursive: bool = False):
        raise NotImplementedError(self.remove_path)
