Repository: canonical/operator
Pull request #734 (2022-04-04 18:38:54)

harness: support mocking storage mounts
This mocks storage mounts (e.g. storage from metadata.yaml) with using a disk-backed tmpdir.  It includes support for pebble file-ops bridging between workload and charm container - i.e. pebble file ops can "see" files in a workload storage mount that were put there directly via a (shared) storage mount in a charm container.

jnsgruk: Failing test and query on '/' in storage names aside, this now looks much cleaner. Thanks for the iteration!

Patch:
diff --git a/ops/charm.py b/ops/charm.py
index 6c94566..ba85b17 100755
--- a/ops/charm.py
+++ b/ops/charm.py
@@ -479,7 +479,7 @@ class StorageEvent(HookEvent):
 
         if isinstance(self.storage, model.Storage):
             snapshot["storage_name"] = self.storage.name
-            snapshot["storage_id"] = self.storage.id
+            snapshot["storage_index"] = self.storage.index
             snapshot["storage_location"] = str(self.storage.location)
         return snapshot
 
@@ -489,13 +489,16 @@ class StorageEvent(HookEvent):
         Not meant to be called by charm code.
         """
         storage_name = snapshot.get("storage_name")
-        storage_id = snapshot.get("storage_id")
+        storage_index = snapshot.get("storage_index")
         storage_location = snapshot.get("storage_location")
 
-        if storage_name and storage_id is not None:
-            self.storage = next(
-                (s for s in self.framework.model.storages[storage_name] if s.id == storage_id),
-                None,)
+        if storage_name and storage_index is not None:
+            storages = self.framework.model.storages[storage_name]
+            self.storage = next((s for s in storages if s.index == storage_index), None,)
+            if self.storage is None:
+                msg = 'failed loading storage (name={!r}, index={!r}) from snapshot' \
+                    .format(storage_name, storage_index)
+                raise RuntimeError(msg)
             self.storage.location = storage_location
 
 
diff --git a/ops/model.py b/ops/model.py
index 388b18a..4fa9f44 100644
--- a/ops/model.py
+++ b/ops/model.py
@@ -1011,8 +1011,8 @@ class StorageMapping(Mapping):
         storage_list = self._storage_map[storage_name]
         if storage_list is None:
             storage_list = self._storage_map[storage_name] = []
-            for storage_id in self._backend.storage_list(storage_name):
-                storage_list.append(Storage(storage_name, storage_id, self._backend))
+            for storage_index in self._backend.storage_list(storage_name):
+                storage_list.append(Storage(storage_name, storage_index, self._backend))
         return storage_list
 
     def request(self, storage_name: str, count: int = 1):
@@ -1039,20 +1039,36 @@ class Storage:
 
     Attributes:
         name: Simple string name of the storage
-        id: The provider id for storage
+        id: The index number for storage
     """
 
-    def __init__(self, storage_name, storage_id, backend):
+    def __init__(self, storage_name: str, storage_index: int, backend):
         self.name = storage_name
-        self.id = storage_id
+        self._index = storage_index
         self._backend = backend
         self._location = None
 
+    @property
+    def index(self) -> int:
+        """The index associated with the storage (usually 0 for singular storage)."""
+        return self._index
+
+    @property
+    def id(self) -> int:
+        """DEPRECATED (use ".index"): The index associated with the storage."""
+        logger.warning("model.Storage.id is being replaced - please use model.Storage.index")
+        return self.index
+
+    @property
+    def full_id(self) -> str:
+        """Returns the canonical storage name and id/index based identifier."""
+        return '{}/{}'.format(self.name, self._index)
+
     @property
     def location(self) -> Path:
         """Return the location of the storage."""
         if self._location is None:
-            raw = self._backend.storage_get('{}/{}'.format(self.name, self.id), "location")
+            raw = self._backend.storage_get(self.full_id, "location")
             self._location = Path(raw)
         return self._location
 


Test patch:
diff --git a/ops/testing.py b/ops/testing.py
index ca6ffde..bf6eda3 100755
--- a/ops/testing.py
+++ b/ops/testing.py
@@ -35,6 +35,7 @@ Global Variables:
 import datetime
 import fnmatch
 import inspect
+import os
 import pathlib
 import random
 import signal
@@ -208,7 +209,8 @@ class Harness(typing.Generic[CharmType]):
             harness = Harness(MyCharm)
             # Do initial setup here
             # Add storage if needed before begin_with_initial_hooks() is called
-            storage_id = harness.add_storage('data', count=1)
+            storage_ids = harness.add_storage('data', count=1)[0]
+            storage_id = storage_id[0] # we only added one storage instance
             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'})
@@ -467,11 +469,13 @@ class Harness(typing.Generic[CharmType]):
         # Model._storages won't return Storage objects for subsequently-added storage.
         self._model._storages._invalidate(storage_name)
 
-        if self.charm is not None and self._hooks_enabled:
-            for storage_index in storage_indices:
-                self.charm.on[storage_name].storage_attached.emit(
-                    model.Storage(storage_name, storage_index, self._backend))
-        return ["{}/{}".format(storage_name, storage_index) for storage_index in storage_indices]
+        ids = []
+        for storage_index in storage_indices:
+            s = model.Storage(storage_name, storage_index, self._backend)
+            ids.append(s.full_id)
+            if self.charm is not None and self._hooks_enabled:
+                self.charm.on[storage_name].storage_attached.emit(s)
+        return ids
 
     def detach_storage(self, storage_id: str) -> None:
         """Detach a storage device.
@@ -481,7 +485,7 @@ class Harness(typing.Generic[CharmType]):
         and is presently marked as attached.
 
         Args:
-            storage_id: The full storage ID of th e storage unit being detached, including the
+            storage_id: The full storage ID of the storage unit being detached, including the
                 storage key, e.g. my-storage/0.
         """
         if self.charm is None:
@@ -511,11 +515,11 @@ class Harness(typing.Generic[CharmType]):
                 model.Storage(storage_name, storage_index, self._backend))
 
     def remove_storage(self, storage_id: str) -> None:
-        """Attach a storage device.
+        """Detach a storage device.
 
         The intent of this function is to simulate a "juju remove-storage" call.
         It will trigger a storage-detaching hook if the storage unit in question exists
-        and is presently marked as detached.  Additionally, it will remove the storage
+        and is presently marked as attached.  Then it will remove the storage
         unit from the testing backend.
 
         Args:
@@ -756,8 +760,7 @@ class Harness(typing.Generic[CharmType]):
             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)
+        client = self._backend._pebble_clients.get(container_name)
         if client is None:
             raise KeyError('no known pebble client for container "{}"'.format(container_name))
         return client.get_plan()
@@ -1078,6 +1081,8 @@ class _TestingModelBackend:
         self.app_name = self.unit_name.split('/')[0]
         self.model_name = None
         self.model_uuid = 'f2c1b2a6-e006-11eb-ba80-0242ac130004'
+
+        self._harness_tmp_dir = tempfile.TemporaryDirectory(prefix='ops-harness-')
         self._calls = []
         self._meta = meta
         self._relation_ids_map = {}  # relation name to [relation_ids,...]
@@ -1099,10 +1104,10 @@ class _TestingModelBackend:
         # <ID1>: device id that is key for given storage_name
         # Initialize the _storage_list with values present on metadata.yaml
         self._storage_list = {k: {} for k in self._meta.storages}
+
         self._storage_detached = {k: set() for k in self._meta.storages}
-        self._storage_id_counter = 0
-        # {socket_path : _TestingPebbleClient}
-        # socket_path = '/charm/containers/{container_name}/pebble.socket'
+        self._storage_index_counter = 0
+        # {container_name : _TestingPebbleClient}
         self._pebble_clients = {}  # type: {str: _TestingPebbleClient}
         self._pebble_clients_can_connect = {}  # type: {_TestingPebbleClient: bool}
         self._planned_units = None
@@ -1252,15 +1257,18 @@ class _TestingModelBackend:
             raise model.ModelError(
                 'ERROR invalid value "{}/{}" for option -s: storage not found'.format(name, index))
 
-    def storage_add(self, name: str, count: int = 1):
+    def storage_add(self, name: str, count: int = 1) -> typing.List[int]:
+        if '/' in name:
+            raise model.ModelError('storage name cannot contain "/"')
+
         if name not in self._storage_list:
             self._storage_list[name] = {}
         result = []
         for i in range(count):
-            index = self._storage_id_counter
-            self._storage_id_counter += 1
+            index = self._storage_index_counter
+            self._storage_index_counter += 1
             self._storage_list[name][index] = {
-                "location": "/{}/{}".format(name, index)
+                'location': os.path.join(self._harness_tmp_dir.name, name, str(index)),
             }
             result.append(index)
         return result
@@ -1270,6 +1278,10 @@ class _TestingModelBackend:
         # detachment of a storage unit.  This is not present in ops.model._ModelBackend.
         name, index = storage_id.split('/', 1)
         index = int(index)
+
+        for client in self._pebble_clients.values():
+            client._fs.remove_mount(name)
+
         if self._storage_is_attached(name, index):
             self._storage_detached[name].add(index)
 
@@ -1278,6 +1290,14 @@ class _TestingModelBackend:
         # re-attachment of a storage unit.  This is not present in
         # ops.model._ModelBackend.
         name, index = storage_id.split('/', 1)
+
+        for container, client in self._pebble_clients.items():
+            for _, mount in self._meta.containers[container].mounts.items():
+                if mount.storage != name:
+                    continue
+                for index, store in self._storage_list[mount.storage].items():
+                    client._fs.add_mount(mount.storage, mount.location, store['location'])
+
         index = int(index)
         if not self._storage_is_attached(name, index):
             self._storage_detached[name].remove(index)
@@ -1321,11 +1341,19 @@ class _TestingModelBackend:
     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)
+    def get_pebble(self, socket_path: str) -> 'pebble.Client':
+        container = socket_path.split('/')[3]  # /charm/containers/<container_name>/pebble.socket
+        client = self._pebble_clients.get(container, None)
         if client is None:
             client = _TestingPebbleClient(self)
-            self._pebble_clients[socket_path] = client
+            self._pebble_clients[container] = client
+
+            # we need to know which container a new pebble client belongs to
+            # so we can figure out which storage mounts must be simulated on
+            # this pebble client's mock file systems when storage is
+            # attached/detached later.
+            self._pebble_clients[container] = client
+
         self._pebble_clients_can_connect[client] = not SIMULATE_CAN_CONNECT
         return client
 
@@ -1367,7 +1395,7 @@ class _TestingPebbleClient:
         self._layers = {}
         # Has a service been started/stopped?
         self._service_status = {}
-        self._fs = _MockFilesystem()
+        self._fs = _TestingFilesystem()
         self._backend = backend
 
     def _check_connection(self):
@@ -1752,14 +1780,155 @@ ChangeError: cannot perform the following tasks:
 
 
 class NonAbsolutePathError(Exception):
-    """Error raised by _MockFilesystem.
+    """Error raised by _TestingFilesystem.
 
     This error is raised when an absolute path is required but the code instead encountered a
     relative path.
     """
 
 
-class _MockFilesystem:
+class _TestingStorageMount:
+    """Simulates a filesystem backend for storage mounts."""
+
+    def __init__(self, location: pathlib.PurePosixPath, src: pathlib.Path):
+        """Creates a new simulated storage mount.
+
+        Args:
+            location: The path within simulated filesystem at which this storage will be mounted.
+            src: The temporary on-disk location where the simulated storage will live.
+        """
+        self._src = src
+        self._location = location
+
+        src.mkdir(exist_ok=True, parents=True)
+
+    def contains(self, path: typing.Union[str, pathlib.PurePosixPath]) -> bool:
+        """Returns true whether path resides within this simulated storage mount's location."""
+        try:
+            pathlib.PurePosixPath(path).relative_to(self._location)
+            return True
+        except Exception:
+            return False
+
+    def check_contains(self, path: typing.Union[str,
+                       pathlib.PurePosixPath]) -> pathlib.PurePosixPath:
+        """Raises if path does not reside within this simulated storage mount's location."""
+        if not self.contains(path):
+            msg = 'the provided path "{!s}" does not reside within the mount location "{!s}"' \
+                .format(path, self._location)
+            raise RuntimeError(msg)
+        return pathlib.PurePosixPath(path)
+
+    def _srcpath(self, path: pathlib.PurePosixPath) -> pathlib.Path:
+        """Returns the disk-backed path where the simulated path will actually be stored."""
+        suffix = path.relative_to(self._location)
+        return self._src / suffix
+
+    def create_dir(
+            self,
+            path: pathlib.PurePosixPath,
+            make_parents: bool = False,
+            **kwargs) -> '_Directory':
+        if not pathlib.PurePosixPath(path).is_absolute():
+            raise NonAbsolutePathError(str(path))
+        path = self.check_contains(path)
+        srcpath = self._srcpath(path)
+
+        if srcpath.exists() and srcpath.is_dir() and make_parents:
+            return _Directory(path, **kwargs)  # nothing to do
+        if srcpath.exists():
+            raise FileExistsError(str(path))
+
+        dirname = srcpath.parent
+        if not dirname.exists():
+            if not make_parents:
+                raise FileNotFoundError(str(path.parent))
+            dirname.mkdir(parents=True, exist_ok=True)
+        srcpath.mkdir(exist_ok=True)
+        return _Directory(path, **kwargs)
+
+    def create_file(
+            self,
+            path: pathlib.PurePosixPath,
+            data: typing.Union[bytes, str, typing.BinaryIO, typing.TextIO],
+            encoding: str = 'utf-8',
+            make_dirs: bool = False,
+            **kwargs
+    ) -> '_File':
+        path = self.check_contains(path)
+        srcpath = self._srcpath(path)
+
+        dirname = srcpath.parent
+        if not dirname.exists():
+            if not make_dirs:
+                raise FileNotFoundError(str(path.parent))
+            dirname.mkdir(parents=True, exist_ok=True)
+
+        if isinstance(data, str):
+            data = data.encode(encoding=encoding)
+        elif not isinstance(data, bytes):
+            data = data.getvalue()
+            if isinstance(data, str):
+                data = data.encode()
+
+        with srcpath.open('wb') as f:
+            f.write(data)
+
+        return _File(path, data, encoding=encoding, **kwargs)
+
+    def list_dir(self, path: pathlib.PurePosixPath) -> typing.List['_File']:
+        path = self.check_contains(path)
+        srcpath = self._srcpath(path)
+
+        results = []
+        if not srcpath.exists():
+            raise FileNotFoundError(str(path))
+        if not srcpath.is_dir():
+            raise NotADirectoryError(str(path))
+        for fpath in srcpath.iterdir():
+            mountpath = path / fpath.name
+            if fpath.is_dir():
+                results.append(_Directory(mountpath))
+            elif fpath.is_file():
+                with fpath.open('rb') as f:
+                    results.append(_File(mountpath, f.read()))
+            else:
+                raise RuntimeError('unsupported file type at path {}'.format(fpath))
+        return results
+
+    def open(
+            self,
+            path: typing.Union[str, pathlib.PurePosixPath],
+            encoding: typing.Optional[str] = 'utf-8',
+    ) -> typing.Union[typing.BinaryIO, typing.TextIO]:
+        path = self.check_contains(path)
+
+        file = self.get_path(path)
+        if isinstance(file, _Directory):
+            raise IsADirectoryError(str(file.path))
+        return file.open(encoding=encoding)
+
+    def get_path(self, path: typing.Union[str, pathlib.PurePosixPath]
+                 ) -> typing.Union['_Directory', '_File']:
+        path = self.check_contains(path)
+        srcpath = self._srcpath(path)
+        if srcpath.is_dir():
+            return _Directory(path)
+        if not srcpath.exists():
+            raise FileNotFoundError(str(path))
+        with srcpath.open('rb') as f:
+            return _File(path, f.read())
+
+    def delete_path(self, path: typing.Union[str, pathlib.PurePosixPath]) -> None:
+        path = self.check_contains(path)
+        srcpath = self._srcpath(path)
+        if srcpath.exists():
+            srcpath.unlink()
+        else:
+            raise FileNotFoundError(str(path))
+
+
+class _TestingFilesystem:
     r"""An in-memory mock of a pebble-controlled container's filesystem.
 
     For now, the filesystem is assumed to be a POSIX-style filesytem; Windows-style directories
@@ -1768,10 +1937,22 @@ class _MockFilesystem:
 
     def __init__(self):
         self.root = _Directory(pathlib.PurePosixPath('/'))
+        self._mounts = {}
+
+    def add_mount(self, name, mount_path, backing_src_path):
+        self._mounts[name] = _TestingStorageMount(
+            pathlib.PurePosixPath(mount_path), pathlib.Path(backing_src_path))
+
+    def remove_mount(self, name):
+        if name in self._mounts:
+            del self._mounts[name]
 
     def create_dir(self, path: str, make_parents: bool = False, **kwargs) -> '_Directory':
         if not path.startswith('/'):
             raise NonAbsolutePathError(path)
+        for mount in self._mounts.values():
+            if mount.contains(path):
+                return mount.create_dir(path, make_parents, **kwargs)
         current_dir = self.root
         tokens = pathlib.PurePosixPath(path).parts[1:]
         for token in tokens[:-1]:
@@ -1813,6 +1994,9 @@ class _MockFilesystem:
     ) -> '_File':
         if not path.startswith('/'):
             raise NonAbsolutePathError(path)
+        for mount in self._mounts.values():
+            if mount.contains(path):
+                return mount.create_file(path, data, encoding, make_dirs, **kwargs)
         path_obj = pathlib.PurePosixPath(path)
         try:
             dir_ = self.get_path(path_obj.parent)
@@ -1832,6 +2016,9 @@ class _MockFilesystem:
         return dir_.create_file(path_obj.name, data, encoding=encoding, **kwargs)
 
     def list_dir(self, path) -> typing.List['_File']:
+        for mount in self._mounts.values():
+            if mount.contains(path):
+                return mount.list_dir(path)
         current_dir = self.root
         tokens = pathlib.PurePosixPath(path).parts[1:]
         for token in tokens:
@@ -1852,6 +2039,9 @@ class _MockFilesystem:
             path: typing.Union[str, pathlib.PurePosixPath],
             encoding: typing.Optional[str] = 'utf-8',
     ) -> typing.Union[typing.BinaryIO, typing.TextIO]:
+        for mount in self._mounts.values():
+            if mount.contains(path):
+                return mount.open(path, encoding)
         path = pathlib.PurePosixPath(path)
         file = self.get_path(path)  # warning: no check re: directories
         if isinstance(file, _Directory):
@@ -1860,6 +2050,9 @@ class _MockFilesystem:
 
     def get_path(self, path: typing.Union[str, pathlib.PurePosixPath]) \
             -> typing.Union['_Directory', '_File']:
+        for mount in self._mounts.values():
+            if mount.contains(path):
+                return mount.get_path(path)
         path = pathlib.PurePosixPath(path)
         tokens = path.parts[1:]
         current_object = self.root
@@ -1872,6 +2065,9 @@ class _MockFilesystem:
         return current_object
 
     def delete_path(self, path: typing.Union[str, pathlib.PurePosixPath]) -> None:
+        for mount in self._mounts.values():
+            if mount.contains(path):
+                return mount.delete_path(path)
         path = pathlib.PurePosixPath(path)
         parent_dir = self.get_path(path.parent)
         del parent_dir[path.name]
diff --git a/test/test_testing.py b/test/test_testing.py
index 36356ea..265f272 100644
--- a/test/test_testing.py
+++ b/test/test_testing.py
@@ -51,8 +51,9 @@ from ops.testing import (
     Harness,
     NonAbsolutePathError,
     _Directory,
-    _MockFilesystem,
+    _TestingFilesystem,
     _TestingPebbleClient,
+    _TestingStorageMount,
 )
 
 is_linux = platform.system() == 'Linux'
@@ -122,7 +123,16 @@ class TestHarness(unittest.TestCase):
         self.assertEqual(backend.relation_get(rel_id, 'test-app', is_app=True), {})
         self.assertEqual(backend.relation_get(rel_id, 'test-app/0', is_app=False), {})
 
-    def test_can_connect(self):
+    def test_can_connect_legacy(self):
+        # This tests the old behavior where we weren't simulating can_connect status of containers
+        # like it runs in juju.
+        tmp = ops.testing.SIMULATE_CAN_CONNECT
+        ops.testing.SIMULATE_CAN_CONNECT = False
+
+        def reset_can_connect():
+            ops.testing.SIMULATE_CAN_CONNECT = tmp
+        self.addCleanup(reset_can_connect)
+
         harness = Harness(CharmBase, meta='''
             name: test-app
             containers:
@@ -138,7 +148,13 @@ class TestHarness(unittest.TestCase):
         self.assertTrue(c.can_connect())
 
     def test_simulate_can_connect(self):
+        tmp = ops.testing.SIMULATE_CAN_CONNECT
         ops.testing.SIMULATE_CAN_CONNECT = True
+
+        def reset_can_connect():
+            ops.testing.SIMULATE_CAN_CONNECT = tmp
+        self.addCleanup(reset_can_connect)
+
         harness = Harness(CharmBase, meta='''
             name: test-app
             containers:
@@ -162,7 +178,6 @@ class TestHarness(unittest.TestCase):
         harness.container_pebble_ready('foo')
         self.assertTrue(c.can_connect())
         c.get_plan()
-        ops.testing.SIMULATE_CAN_CONNECT = False
 
     def test_add_relation_and_unit(self):
         harness = Harness(CharmBase, meta='''
@@ -1285,7 +1300,8 @@ class TestHarness(unittest.TestCase):
         stor_ids = harness.add_storage("test", count=3)
         self.assertSetEqual(set(self._extract_storage_index(stor_id) for stor_id in stor_ids),
                             set(harness._backend.storage_list("test")))
-        self.assertEqual("/test/0", harness._backend.storage_get("test/0", "location"))
+        want = str(pathlib.PurePath('test', '0'))
+        self.assertEqual(want, harness._backend.storage_get("test/0", "location")[-6:])
 
         harness.begin_with_initial_hooks()
         self.assertEqual(len(harness.charm.observed_events), 3)
@@ -1325,7 +1341,8 @@ class TestHarness(unittest.TestCase):
         stor_id = harness.add_storage("test")[0]
         self.assertIn(self._extract_storage_index(stor_id),
                       harness._backend.storage_list("test"))
-        self.assertEqual("/test/0", harness._backend.storage_get(stor_id, "location"))
+        want = str(pathlib.PurePath('test', '0'))
+        self.assertEqual(want, harness._backend.storage_get(stor_id, "location")[-6:])
 
         harness.begin_with_initial_hooks()
         self.assertEqual(len(harness.charm.observed_events), 1)
@@ -1339,9 +1356,11 @@ class TestHarness(unittest.TestCase):
 
         added_indices = {self._extract_storage_index(stor_id) for stor_id in stor_ids}
         self.assertTrue(added_indices.issubset(set(harness._backend.storage_list("test"))))
-        self.assertEqual("/test/1", harness._backend.storage_get("test/1", "location"))
-        self.assertEqual("/test/2", harness._backend.storage_get("test/2", "location"))
-        self.assertEqual("/test/3", harness._backend.storage_get("test/3", "location"))
+
+        for i in ['1', '2', '3']:
+            storage_name = 'test/' + i
+            want = str(pathlib.PurePath('test', i))
+            self.assertEqual(want, harness._backend.storage_get(storage_name, "location")[-6:])
         self.assertEqual(len(harness.charm.observed_events), 4)
         for i in range(1, 4):
             self.assertTrue(isinstance(harness.charm.observed_events[i], StorageAttachedEvent))
@@ -1456,7 +1475,8 @@ class TestHarness(unittest.TestCase):
         # Verify backend functions return appropriate values.
         # Real backend would return info only for actively attached storage units.
         self.assertIn(self._extract_storage_index(stor_id), harness._backend.storage_list("test"))
-        self.assertEqual("/test/0", harness._backend.storage_get("test/0", "location"))
+        want = str(pathlib.PurePath('test', '0'))
+        self.assertEqual(want, harness._backend.storage_get("test/0", "location")[-6:])
 
         # Retry attach
         # Since already detached, no more hooks should fire
@@ -3728,17 +3748,14 @@ class _PebbleStorageAPIsTestMixin:
     #   nuance.
 
 
-class TestMockFilesystem(unittest.TestCase):
-    def setUp(self):
-        self.fs = _MockFilesystem()
-
+class GenericTestingFilesystemTests:
     def test_listdir_root_on_empty_os(self):
         self.assertEqual(self.fs.list_dir('/'), [])
 
     def test_listdir_on_nonexistent_dir(self):
         with self.assertRaises(FileNotFoundError) as cm:
             self.fs.list_dir('/etc')
-        self.assertEqual(cm.exception.args[0], '/etc')
+        self.assertTrue('/etc' in cm.exception.args[0])
 
     def test_listdir(self):
         self.fs.create_dir('/opt')
@@ -3756,7 +3773,7 @@ class TestMockFilesystem(unittest.TestCase):
         self.fs.create_file('/file', 'data')
         with self.assertRaises(NotADirectoryError) as cm:
             self.fs.list_dir('/file')
-        self.assertEqual(cm.exception.args[0], '/file')
+        self.assertTrue('/file' in cm.exception.args[0])
 
     def test_makedir(self):
         d = self.fs.create_dir('/etc')
@@ -3770,7 +3787,7 @@ class TestMockFilesystem(unittest.TestCase):
         self.fs.create_dir('/etc')
         with self.assertRaises(FileExistsError) as cm:
             self.fs.create_dir('/etc')
-        self.assertEqual(cm.exception.args[0], '/etc')
+        self.assertTrue('/etc' in cm.exception.args[0])
 
     def test_makedir_succeeds_if_already_exists_when_make_parents_true(self):
         d1 = self.fs.create_dir('/etc')
@@ -3781,7 +3798,7 @@ class TestMockFilesystem(unittest.TestCase):
     def test_makedir_fails_if_parent_dir_doesnt_exist(self):
         with self.assertRaises(FileNotFoundError) as cm:
             self.fs.create_dir('/etc/init.d')
-        self.assertEqual(cm.exception.args[0], '/etc')
+        self.assertTrue('/etc' in cm.exception.args[0])
 
     def test_make_and_list_directory(self):
         self.fs.create_dir('/etc')
@@ -3802,7 +3819,7 @@ class TestMockFilesystem(unittest.TestCase):
     def test_create_file_fails_if_parent_dir_doesnt_exist(self):
         with self.assertRaises(FileNotFoundError) as cm:
             self.fs.create_file('/etc/passwd', "foo")
-        self.assertEqual(cm.exception.args[0], '/etc')
+        self.assertTrue('/etc' in cm.exception.args[0])
 
     def test_create_file_succeeds_if_parent_dir_doesnt_exist_when_make_dirs_true(self):
         self.fs.create_file('/test/subdir/testfile', "foo", make_dirs=True)
@@ -3861,7 +3878,7 @@ class TestMockFilesystem(unittest.TestCase):
         # Deleting deleted files should fail as well
         with self.assertRaises(FileNotFoundError) as cm:
             self.fs.delete_path('/test')
-        self.assertEqual(cm.exception.args[0], '/test')
+        self.assertTrue('/test' in cm.exception.args[0])
 
     def test_create_dir_with_extra_args(self):
         d = self.fs.create_dir('/dir1')
@@ -3896,7 +3913,7 @@ class TestMockFilesystem(unittest.TestCase):
         self.fs.create_dir('/etc/init.d', make_parents=True)
 
         # By path
-        o = self.fs.get_path(pathlib.Path('/etc/init.d'))
+        o = self.fs.get_path(pathlib.PurePosixPath('/etc/init.d'))
         self.assertIsInstance(o, _Directory)
         self.assertEqual(o.path, pathlib.PurePosixPath('/etc/init.d'))
 
@@ -3911,7 +3928,29 @@ class TestMockFilesystem(unittest.TestCase):
         # gives a closer semantic feeling, in my opinion.
         with self.assertRaises(FileNotFoundError) as cm:
             self.fs.get_path('/nonexistent_file')
-        self.assertEqual(cm.exception.args[0], '/nonexistent_file')
+        self.assertTrue('/nonexistent_file' in cm.exception.args[0])
+
+
+class TestTestingFilesystem(GenericTestingFilesystemTests, unittest.TestCase):
+    def setUp(self):
+        self.fs = _TestingFilesystem()
+
+    def test_storage_mount(self):
+        tmpdir = tempfile.TemporaryDirectory()
+        self.fs.add_mount('foo', '/foo', tmpdir.name)
+        self.fs.create_file('/foo/bar/baz.txt', 'quux', make_dirs=True)
+
+        tmppath = os.path.join(tmpdir.name, 'bar/baz.txt')
+        self.assertTrue(os.path.exists(tmppath))
+        with open(tmppath) as f:
+            self.assertEqual(f.read(), 'quux')
+
+
+class TestTestingStorageMount(GenericTestingFilesystemTests, unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.fs = _TestingStorageMount('/', pathlib.Path(self.tmp.name))
 
 
 class TestPebbleStorageAPIsUsingMocks(
@@ -3924,6 +3963,73 @@ class TestPebbleStorageAPIsUsingMocks(
         if self.prefix:
             self.client.make_dir(self.prefix, make_parents=True)
 
+    @unittest.skipUnless(is_linux, 'Pebble runs on Linux')
+    def test_container_storage_mounts(self):
+        harness = Harness(CharmBase, meta='''
+            name: test-app
+            containers:
+                c1:
+                    mounts:
+                        - storage: store1
+                          location: /mounts/foo
+                c2:
+                    mounts:
+                        - storage: store2
+                          location: /mounts/foo
+                c3:
+                    mounts:
+                        - storage: store1
+                          location: /mounts/bar
+            storage:
+                store1:
+                    type: filesystem
+                store2:
+                    type: filesystem
+            ''')
+        self.addCleanup(harness.cleanup)
+
+        store_id = harness.add_storage('store1')[0]
+        harness.attach_storage(store_id)
+
+        harness.begin()
+
+        # push file to c1 storage mount, check that we can see it in charm container storage path.
+        c1 = harness.model.unit.get_container('c1')
+        c1_fname = 'foo.txt'
+        c1_fpath = os.path.join('/mounts/foo', c1_fname)
+        c1.push(c1_fpath, '42')
+        self.assertTrue(c1.exists(c1_fpath))
+        fpath = os.path.join(str(harness.model.storages['store1'][0].location), 'foo.txt')
+        with open(fpath) as f:
+            self.assertEqual('42', f.read())
+
+        # check that the file is not visible in c2 which has a different storage mount
+        c2 = harness.model.unit.get_container('c2')
+        c2_fpath = os.path.join('/mounts/foo', c1_fname)
+        self.assertFalse(c2.exists(c2_fpath))
+
+        # check that the file is visible in c3 which has the same storage mount
+        c3 = harness.model.unit.get_container('c3')
+        c3_fpath = os.path.join('/mounts/bar', c1_fname)
+        self.assertTrue(c3.exists(c3_fpath))
+        with c3.pull(c3_fpath) as f:
+            self.assertEqual('42', f.read())
+
+        # test all other container file ops
+        with c1.pull(c1_fpath) as f:
+            self.assertEqual('42', f.read())
+        files = c1.list_files(c1_fpath)
+        self.assertEqual([c1_fpath], [fi.path for fi in files])
+        c1.remove_path(c1_fpath)
+        self.assertFalse(c1.exists(c1_fpath))
+
+        # test detaching storage
+        c1.push(c1_fpath, '42')
+        self.assertTrue(c1.exists(c1_fpath))
+        store1_id = harness.model.storages['store1'][0].full_id
+        harness.remove_storage(store1_id)
+        self.assertFalse(c1.exists(c1_fpath))
+
     def test_push_with_ownership(self):
         # Note: To simplify implementation, ownership is simply stored as-is with no verification.
         data = 'data'