Merge branch 'fix/web-app-weblate/swarm-static-mountpoint' into feature/store-i18n

tests/unit/python/utils/storage/test_nfs.py

@@ -1,15 +1,42 @@
 import unittest
 
+from utils.cache.files import PROJECT_ROOT, read_text
+from utils.cache.yaml import load_yaml_str
+from utils.roles.mapping import ROLE_FILE_META_VOLUMES
 from utils.storage.nfs import (
     client_src,
     fstype,
     mount_opts,
+    nested_tmpfs_mountpoints,
     state_path,
     swarm_nfs_backed,
     volume_dir,
 )
 
 
+def _role_volumes(application_id):
+    """Return a role's ``meta/volumes.yml`` mapping.
+
+    Args:
+        application_id: the role whose file is read.
+    """
+    path = PROJECT_ROOT / "roles" / application_id / ROLE_FILE_META_VOLUMES
+    return load_yaml_str(read_text(str(path)))
+
+
+def _every_volume_on_nfs(volumes_meta):
+    """Return the NFS directory of every volume that does not opt out of NFS.
+
+    Args:
+        volumes_meta: a ``meta/volumes.yml`` mapping.
+    """
+    return {
+        name: volume_dir(name, entry)
+        for name, entry in volumes_meta.items()
+        if entry.get("type", "volume") == "volume" and entry.get("nfs") is not False
+    }
+
+
 class TestNfsHelpers(unittest.TestCase):
     def test_state_path(self):
         self.assertEqual(
@@ -155,5 +182,86 @@ class TestSwarmNfsBacked(unittest.TestCase):
         )
 
 
+class TestNestedTmpfsMountpoints(unittest.TestCase):
+    def test_weblate_static_mountpoint_carries_the_tmpfs_mode(self):
+        volumes = _role_volumes("web-app-weblate")
+        self.assertEqual(
+            nested_tmpfs_mountpoints(volumes, _every_volume_on_nfs(volumes)),
+            [{"path": "weblate_cache/static", "mode": 0o1777}],
+            "the static tmpfs of weblate, beat and worker sits inside the "
+            "weblate_cache volume; its mountpoint directory must exist with mode "
+            "1777 before the stack deploys, or uid 1000 cannot collectstatic",
+        )
+
+    def test_matrix_mautrix_workdir_is_not_nested_in_its_own_services_volume(self):
+        volumes = _role_volumes("web-app-matrix")
+        self.assertEqual(
+            nested_tmpfs_mountpoints(volumes, _every_volume_on_nfs(volumes)),
+            [],
+            "the mautrix tmpfs at /data shares its target with synapse_data, but "
+            "only synapse mounts that volume",
+        )
+
+    def test_a_sibling_path_is_not_nested(self):
+        volumes = {
+            "cache": {
+                "type": "volume",
+                "mounts": [{"service": "app", "target": "/app/cache"}],
+            },
+            "scratch": {
+                "type": "tmpfs",
+                "mounts": [{"service": "app", "target": "/app/cachex"}],
+            },
+        }
+        self.assertEqual(nested_tmpfs_mountpoints(volumes, {"cache": "app_cache"}), [])
+
+    def test_the_innermost_volume_holds_the_mountpoint(self):
+        volumes = {
+            "app": {
+                "type": "volume",
+                "mounts": [{"service": "app", "target": "/app"}],
+            },
+            "cache": {
+                "type": "volume",
+                "mounts": [{"service": "app", "target": "/app/cache/"}],
+            },
+            "static": {
+                "type": "tmpfs",
+                "mounts": [
+                    {"service": "app", "target": "/app/cache/static", "mode": "0700"}
+                ],
+            },
+        }
+        nfs_dirs = {"app": "app_app", "cache": "app_cache"}
+        self.assertEqual(
+            nested_tmpfs_mountpoints(volumes, nfs_dirs),
+            [{"path": "app_cache/static", "mode": 0o700}],
+        )
+
+    def test_a_nested_tmpfs_without_mode_fails_loudly(self):
+        volumes = {
+            "cache": {
+                "type": "volume",
+                "mounts": [{"service": "app", "target": "/app/cache"}],
+            },
+            "static": {
+                "type": "tmpfs",
+                "mounts": [{"service": "app", "target": "/app/cache/static"}],
+            },
+        }
+        with self.assertRaisesRegex(ValueError, "declares no 'mode'"):
+            nested_tmpfs_mountpoints(volumes, {"cache": "app_cache"})
+
+    def test_every_role_declares_a_mode_on_its_nested_tmpfs_mounts(self):
+        for path in sorted(
+            (PROJECT_ROOT / "roles").glob(f"*/{ROLE_FILE_META_VOLUMES}")
+        ):
+            volumes = load_yaml_str(read_text(str(path)))
+            if not isinstance(volumes, dict):
+                continue
+            with self.subTest(role=path.parent.parent.name):
+                nested_tmpfs_mountpoints(volumes, _every_volume_on_nfs(volumes))
+
+
 if __name__ == "__main__":
     unittest.main()