mirror of
https://github.com/MAGICGrants/truenas-apps.git
synced 2026-01-09 20:47:58 -05:00
33a4a12ae8
* lib: use pre-built image * lib: use the `tmpfs` key instead of the `volumes` for tmpfs type (#1708) * tmpfs * update hash and add tests
76 lines
2.8 KiB
Python
76 lines
2.8 KiB
Python
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from container import Container
|
|
from render import Render
|
|
from storage import IxStorage
|
|
|
|
try:
|
|
from .error import RenderError
|
|
from .validations import valid_fs_path_or_raise, valid_octal_mode_or_raise
|
|
except ImportError:
|
|
from error import RenderError
|
|
from validations import valid_fs_path_or_raise, valid_octal_mode_or_raise
|
|
|
|
|
|
class Tmpfs:
|
|
|
|
def __init__(self, render_instance: "Render", container_instance: "Container"):
|
|
self._render_instance = render_instance
|
|
self._container_instance = container_instance
|
|
self._tmpfs: dict = {}
|
|
|
|
def add(self, mount_path: str, config: "IxStorage"):
|
|
mount_path = valid_fs_path_or_raise(mount_path)
|
|
if self.is_defined(mount_path):
|
|
raise RenderError(f"Tmpfs mount path [{mount_path}] already added")
|
|
|
|
if self._container_instance.storage.is_defined(mount_path):
|
|
raise RenderError(f"Tmpfs mount path [{mount_path}] already used for another volume mount")
|
|
|
|
mount_config = config.get("tmpfs_config", {})
|
|
size = mount_config.get("size", None)
|
|
mode = mount_config.get("mode", None)
|
|
uid = mount_config.get("uid", None)
|
|
gid = mount_config.get("gid", None)
|
|
|
|
if size is not None:
|
|
if not isinstance(size, int):
|
|
raise RenderError(f"Expected [size] to be an integer for [tmpfs] type, got [{size}]")
|
|
if not size > 0:
|
|
raise RenderError(f"Expected [size] to be greater than 0 for [tmpfs] type, got [{size}]")
|
|
# Convert Mebibytes to Bytes
|
|
size = size * 1024 * 1024
|
|
|
|
if mode is not None:
|
|
mode = valid_octal_mode_or_raise(mode)
|
|
|
|
if uid is not None and not isinstance(uid, int):
|
|
raise RenderError(f"Expected [uid] to be an integer for [tmpfs] type, got [{uid}]")
|
|
|
|
if gid is not None and not isinstance(gid, int):
|
|
raise RenderError(f"Expected [gid] to be an integer for [tmpfs] type, got [{gid}]")
|
|
|
|
self._tmpfs[mount_path] = {}
|
|
if size is not None:
|
|
self._tmpfs[mount_path]["size"] = str(size)
|
|
if mode is not None:
|
|
self._tmpfs[mount_path]["mode"] = str(mode)
|
|
if uid is not None:
|
|
self._tmpfs[mount_path]["uid"] = str(uid)
|
|
if gid is not None:
|
|
self._tmpfs[mount_path]["gid"] = str(gid)
|
|
|
|
def is_defined(self, mount_path: str):
|
|
return mount_path in self._tmpfs
|
|
|
|
def has_tmpfs(self):
|
|
return bool(self._tmpfs)
|
|
|
|
def render(self):
|
|
result = []
|
|
for mount_path, config in self._tmpfs.items():
|
|
opts = sorted([f"{k}={v}" for k, v in config.items()])
|
|
result.append(f"{mount_path}:{','.join(opts)}" if opts else mount_path)
|
|
return sorted(result)
|