mirror of
https://github.com/home-assistant/supervisor.git
synced 2026-08-18 02:43:18 +01:00
4081fd5556
* Check for local data before adding or updating a mount Adding a mount whose media/share directory already holds local data (e.g. Frigate recordings written before network storage was set up, or after systemd tore down the bind mount, see #7013) only failed at the bind-mount step: the data mount was already mounted and the mount registered in memory but never persisted, leaving a half-created mount until the next Supervisor restart. Updating an existing mount in that state even unmounted the working data mount first, just to fail on the non-empty bind target afterwards. Check the mount's target directories for local data upfront and fail the add/update before anything is touched. Paths that are already mount points are skipped since they get unmounted before reuse. If mounting still fails halfway (e.g. the bind mount unit fails to start), roll back by unmounting the new units instead of keeping the half-created mount in the list. Also pass the missing mount name argument to the debug log statement in create_mount. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Make mount target errors translatable Raise the mount target error conditions as dedicated exceptions with an error_key, message template and extra fields following the established pattern, so the frontend can localize them and show the mount name and path separately: - MountTargetNotEmptyError (mount_target_not_empty_error), raised both by the upfront local data check and the mount-time non-empty check - MountTargetNotDirectoryError (mount_target_not_directory_error) for a target path that exists but is not a directory Both remain subclasses of MountInvalidError so existing exception handling keeps working. Frontends without the translation strings fall back to the English message as before. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Raise not-a-directory error from upfront local data check The upfront check collapsed a target path that exists but is not a directory into MountTargetNotEmptyError. Distinguish the two conditions like the mount-time check does and raise MountTargetNotDirectoryError for the non-directory case. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
926 lines
31 KiB
Python
926 lines
31 KiB
Python
"""Tests for mount manager."""
|
|
|
|
import errno
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from unittest.util import unorderable_list_difference
|
|
|
|
from dbus_fast import DBusError, ErrorType, Variant
|
|
from dbus_fast.aio.message_bus import MessageBus
|
|
import pytest
|
|
|
|
from supervisor.coresys import CoreSys
|
|
from supervisor.dbus.const import UnitActiveState
|
|
from supervisor.exceptions import (
|
|
MountActivationError,
|
|
MountError,
|
|
MountJobError,
|
|
MountNotFound,
|
|
MountTargetNotDirectoryError,
|
|
MountTargetNotEmptyError,
|
|
)
|
|
from supervisor.mounts.manager import MountManager
|
|
from supervisor.mounts.mount import BindMount, Mount
|
|
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
|
|
from supervisor.resolution.data import Issue, Suggestion
|
|
|
|
from tests.common import mock_dbus_services
|
|
from tests.dbus_service_mocks.base import DBusServiceMock
|
|
from tests.dbus_service_mocks.systemd import Systemd as SystemdService
|
|
from tests.dbus_service_mocks.systemd_unit import SystemdUnit as SystemdUnitService
|
|
|
|
ERROR_NO_UNIT = DBusError("org.freedesktop.systemd1.NoSuchUnit", "error")
|
|
BACKUP_TEST_DATA = {
|
|
"name": "backup_test",
|
|
"type": "cifs",
|
|
"usage": "backup",
|
|
"server": "backup.local",
|
|
"share": "backups",
|
|
}
|
|
MEDIA_TEST_DATA = {
|
|
"name": "media_test",
|
|
"type": "nfs",
|
|
"usage": "media",
|
|
"server": "media.local",
|
|
"path": "/media",
|
|
}
|
|
SHARE_TEST_DATA = {
|
|
"name": "share_test",
|
|
"type": "nfs",
|
|
"usage": "share",
|
|
"server": "share.local",
|
|
"path": "/share",
|
|
}
|
|
|
|
|
|
@pytest.fixture(name="mount")
|
|
async def fixture_mount(
|
|
coresys: CoreSys, tmp_supervisor_data, path_extern, mount_propagation, mock_is_mount
|
|
) -> Mount:
|
|
"""Add an initial mount and load mounts."""
|
|
mount = Mount.from_dict(coresys, MEDIA_TEST_DATA)
|
|
coresys.mounts._mounts = {"media_test": mount} # pylint: disable=protected-access
|
|
await coresys.mounts.load()
|
|
return mount
|
|
|
|
|
|
async def test_load(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
mount_propagation,
|
|
):
|
|
"""Test mount manager loading."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
|
|
backup_test = Mount.from_dict(coresys, BACKUP_TEST_DATA)
|
|
media_test = Mount.from_dict(coresys, MEDIA_TEST_DATA)
|
|
# pylint: disable=protected-access
|
|
coresys.mounts._mounts = {
|
|
"backup_test": backup_test,
|
|
"media_test": media_test,
|
|
}
|
|
# pylint: enable=protected-access
|
|
assert coresys.mounts.backup_mounts == [backup_test]
|
|
assert coresys.mounts.media_mounts == [media_test]
|
|
|
|
assert backup_test.state is None
|
|
assert media_test.state is None
|
|
assert not backup_test.local_where.exists()
|
|
assert not media_test.local_where.exists()
|
|
assert not any(coresys.config.path_media.iterdir())
|
|
|
|
systemd_service.response_get_unit = {
|
|
"mnt-data-supervisor-mounts-backup_test.mount": [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
],
|
|
"mnt-data-supervisor-mounts-media_test.mount": [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
],
|
|
"mnt-data-supervisor-media-media_test.mount": [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
],
|
|
}
|
|
await coresys.mounts.load()
|
|
|
|
assert backup_test.state == UnitActiveState.ACTIVE
|
|
assert media_test.state == UnitActiveState.ACTIVE
|
|
assert backup_test.local_where.is_dir()
|
|
assert media_test.local_where.is_dir()
|
|
assert (coresys.config.path_media / "media_test").is_dir()
|
|
|
|
assert unorderable_list_difference(
|
|
systemd_service.StartTransientUnit.calls,
|
|
[
|
|
(
|
|
"mnt-data-supervisor-mounts-backup_test.mount",
|
|
"fail",
|
|
[
|
|
(
|
|
"Options",
|
|
Variant(
|
|
"s", "noserverino,soft,echo_interval=10,retrans=0,guest"
|
|
),
|
|
),
|
|
("Type", Variant("s", "cifs")),
|
|
("Description", Variant("s", "Supervisor cifs mount: backup_test")),
|
|
("What", Variant("s", "//backup.local/backups")),
|
|
("TimeoutUSec", Variant("t", 35000000)),
|
|
],
|
|
[],
|
|
),
|
|
(
|
|
"mnt-data-supervisor-mounts-media_test.mount",
|
|
"fail",
|
|
[
|
|
("Options", Variant("s", "softerr,timeo=100,retrans=2")),
|
|
("Type", Variant("s", "nfs")),
|
|
("Description", Variant("s", "Supervisor nfs mount: media_test")),
|
|
("What", Variant("s", "media.local:/media")),
|
|
("TimeoutUSec", Variant("t", 35000000)),
|
|
],
|
|
[],
|
|
),
|
|
(
|
|
"mnt-data-supervisor-media-media_test.mount",
|
|
"fail",
|
|
[
|
|
("Options", Variant("s", "bind")),
|
|
(
|
|
"Description",
|
|
Variant("s", "Supervisor bind mount: bind_media_test"),
|
|
),
|
|
("What", Variant("s", "/mnt/data/supervisor/mounts/media_test")),
|
|
("TimeoutUSec", Variant("t", 35000000)),
|
|
],
|
|
[],
|
|
),
|
|
],
|
|
) == ([], [])
|
|
|
|
|
|
async def test_load_share_mount(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
mount_propagation,
|
|
):
|
|
"""Test mount manager loading with share mount."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
|
|
share_test = Mount.from_dict(coresys, SHARE_TEST_DATA)
|
|
# pylint: disable=protected-access
|
|
coresys.mounts._mounts = {
|
|
"share_test": share_test,
|
|
}
|
|
# pylint: enable=protected-access
|
|
assert coresys.mounts.share_mounts == [share_test]
|
|
|
|
assert share_test.state is None
|
|
assert not share_test.local_where.exists()
|
|
assert not any(coresys.config.path_share.iterdir())
|
|
|
|
systemd_service.response_get_unit = {
|
|
"mnt-data-supervisor-mounts-share_test.mount": [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
],
|
|
"mnt-data-supervisor-share-share_test.mount": [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
],
|
|
}
|
|
await coresys.mounts.load()
|
|
|
|
assert share_test.state == UnitActiveState.ACTIVE
|
|
assert share_test.local_where.is_dir()
|
|
assert (coresys.config.path_share / "share_test").is_dir()
|
|
|
|
assert systemd_service.StartTransientUnit.calls == [
|
|
(
|
|
"mnt-data-supervisor-mounts-share_test.mount",
|
|
"fail",
|
|
[
|
|
("Options", Variant("s", "softerr,timeo=100,retrans=2")),
|
|
("Type", Variant("s", "nfs")),
|
|
("Description", Variant("s", "Supervisor nfs mount: share_test")),
|
|
("What", Variant("s", "share.local:/share")),
|
|
("TimeoutUSec", Variant("t", 35000000)),
|
|
],
|
|
[],
|
|
),
|
|
(
|
|
"mnt-data-supervisor-share-share_test.mount",
|
|
"fail",
|
|
[
|
|
("Options", Variant("s", "bind")),
|
|
("Description", Variant("s", "Supervisor bind mount: bind_share_test")),
|
|
("What", Variant("s", "/mnt/data/supervisor/mounts/share_test")),
|
|
("TimeoutUSec", Variant("t", 35000000)),
|
|
],
|
|
[],
|
|
),
|
|
]
|
|
|
|
|
|
async def test_mount_failed_during_load(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
dbus_session_bus: MessageBus,
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
mount_propagation,
|
|
):
|
|
"""Test mount failed during load."""
|
|
await mock_dbus_services(
|
|
{"systemd_unit": "/org/freedesktop/systemd1/unit/tmp_test"}, dbus_session_bus
|
|
)
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_unit_service: SystemdUnitService = all_dbus_services["systemd_unit"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
|
|
backup_test = Mount.from_dict(coresys, BACKUP_TEST_DATA)
|
|
media_test = Mount.from_dict(coresys, MEDIA_TEST_DATA)
|
|
# pylint: disable=protected-access
|
|
coresys.mounts._mounts = {
|
|
"backup_test": backup_test,
|
|
"media_test": media_test,
|
|
}
|
|
# pylint: enable=protected-access
|
|
|
|
assert backup_test.state is None
|
|
assert media_test.state is None
|
|
assert not backup_test.local_where.exists()
|
|
assert not media_test.local_where.exists()
|
|
assert not any(coresys.config.path_emergency.iterdir())
|
|
assert not any(coresys.config.path_media.iterdir())
|
|
|
|
assert coresys.resolution.issues == []
|
|
assert coresys.resolution.suggestions == []
|
|
|
|
systemd_service.response_get_unit = {
|
|
"mnt-data-supervisor-mounts-backup_test.mount": [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
],
|
|
"mnt-data-supervisor-mounts-media_test.mount": [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
],
|
|
"mnt-data-supervisor-media-media_test.mount": [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_test",
|
|
],
|
|
}
|
|
systemd_unit_service.active_state = "failed"
|
|
await coresys.mounts.load()
|
|
|
|
assert backup_test.state == UnitActiveState.FAILED
|
|
assert media_test.state == UnitActiveState.FAILED
|
|
assert backup_test.local_where.is_dir()
|
|
assert media_test.local_where.is_dir()
|
|
assert (coresys.config.path_media / "media_test").is_dir()
|
|
emergency_dir = coresys.config.path_emergency / "media_test"
|
|
assert emergency_dir.is_dir()
|
|
assert os.access(emergency_dir, os.R_OK)
|
|
assert not os.access(emergency_dir, os.W_OK)
|
|
|
|
assert (
|
|
Issue(IssueType.MOUNT_FAILED, ContextType.MOUNT, reference="backup_test")
|
|
in coresys.resolution.issues
|
|
)
|
|
assert (
|
|
Suggestion(
|
|
SuggestionType.EXECUTE_RELOAD, ContextType.MOUNT, reference="backup_test"
|
|
)
|
|
in coresys.resolution.suggestions
|
|
)
|
|
assert (
|
|
Suggestion(
|
|
SuggestionType.EXECUTE_REMOVE, ContextType.MOUNT, reference="backup_test"
|
|
)
|
|
in coresys.resolution.suggestions
|
|
)
|
|
assert (
|
|
Issue(IssueType.MOUNT_FAILED, ContextType.MOUNT, reference="media_test")
|
|
in coresys.resolution.issues
|
|
)
|
|
assert (
|
|
Suggestion(
|
|
SuggestionType.EXECUTE_RELOAD, ContextType.MOUNT, reference="media_test"
|
|
)
|
|
in coresys.resolution.suggestions
|
|
)
|
|
assert (
|
|
Suggestion(
|
|
SuggestionType.EXECUTE_REMOVE, ContextType.MOUNT, reference="media_test"
|
|
)
|
|
in coresys.resolution.suggestions
|
|
)
|
|
|
|
assert len(systemd_service.StartTransientUnit.calls) == 3
|
|
assert systemd_service.StartTransientUnit.calls[2] == (
|
|
"mnt-data-supervisor-media-media_test.mount",
|
|
"fail",
|
|
[
|
|
("Options", Variant("s", "ro,bind")),
|
|
(
|
|
"Description",
|
|
Variant("s", "Supervisor bind mount: emergency_media_test"),
|
|
),
|
|
("What", Variant("s", "/mnt/data/supervisor/emergency/media_test")),
|
|
("TimeoutUSec", Variant("t", 35000000)),
|
|
],
|
|
[],
|
|
)
|
|
|
|
|
|
async def test_create_mount(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
mount_propagation,
|
|
mock_is_mount,
|
|
):
|
|
"""Test creating a mount."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
|
|
await coresys.mounts.load()
|
|
|
|
mount = Mount.from_dict(coresys, MEDIA_TEST_DATA)
|
|
|
|
assert mount.state is None
|
|
assert mount not in coresys.mounts
|
|
assert "media_test" not in coresys.mounts
|
|
assert not mount.local_where.exists()
|
|
assert not any(coresys.config.path_media.iterdir())
|
|
|
|
# Create the mount
|
|
systemd_service.response_get_unit = [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
]
|
|
await coresys.mounts.create_mount(mount)
|
|
|
|
assert mount.state == UnitActiveState.ACTIVE
|
|
assert mount in coresys.mounts
|
|
assert "media_test" in coresys.mounts
|
|
assert mount.local_where.exists()
|
|
assert (coresys.config.path_media / "media_test").exists()
|
|
|
|
assert [call[0] for call in systemd_service.StartTransientUnit.calls] == [
|
|
"mnt-data-supervisor-mounts-media_test.mount",
|
|
"mnt-data-supervisor-media-media_test.mount",
|
|
]
|
|
|
|
|
|
async def test_update_mount(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
mount: Mount,
|
|
):
|
|
"""Test updating a mount."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_unit_service: SystemdUnitService = all_dbus_services["systemd_unit"]
|
|
systemd_service.mock_systemd_unit = systemd_unit_service
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
systemd_service.StopUnit.calls.clear()
|
|
|
|
# Update the mount. Should be unmounted then remounted
|
|
mount_new = Mount.from_dict(coresys, MEDIA_TEST_DATA)
|
|
assert mount.state == UnitActiveState.ACTIVE
|
|
assert mount_new.state is None
|
|
|
|
systemd_service.response_get_unit = [
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
]
|
|
await coresys.mounts.create_mount(mount_new)
|
|
|
|
assert mount.state is None
|
|
assert mount_new.state == UnitActiveState.ACTIVE
|
|
|
|
assert [call[0] for call in systemd_service.StartTransientUnit.calls] == [
|
|
"mnt-data-supervisor-mounts-media_test.mount",
|
|
"mnt-data-supervisor-media-media_test.mount",
|
|
]
|
|
assert [call[0] for call in systemd_service.StopUnit.calls] == [
|
|
"mnt-data-supervisor-media-media_test.mount",
|
|
"mnt-data-supervisor-mounts-media_test.mount",
|
|
]
|
|
|
|
|
|
async def test_reload_mount_healthy_skips_systemd(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
mount: Mount,
|
|
):
|
|
"""A healthy mount (active + probe passes) skips the systemd reload but rebinds."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.ReloadOrRestartUnit.calls.clear()
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
systemd_service.StopUnit.calls.clear()
|
|
|
|
systemd_service.response_get_unit = [
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
]
|
|
|
|
await coresys.mounts.reload_mount(mount.name)
|
|
|
|
assert systemd_service.ReloadOrRestartUnit.calls == []
|
|
assert [call[0] for call in systemd_service.StopUnit.calls] == [
|
|
"mnt-data-supervisor-media-media_test.mount"
|
|
]
|
|
assert [call[0] for call in systemd_service.StartTransientUnit.calls] == [
|
|
"mnt-data-supervisor-media-media_test.mount"
|
|
]
|
|
|
|
|
|
async def test_reload_mount_recreates_bind_mount_removed_by_systemd(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
mount: Mount,
|
|
):
|
|
"""Reload re-creates a bind mount that systemd tore down with the data mount.
|
|
|
|
systemd's implicit Requires= dependency from the bind unit to the data
|
|
mount unit means a stop/restart of a failed data mount unmounts the bind
|
|
mount as well, while the BoundMount bookkeeping still says it is bound.
|
|
"""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
systemd_service.StopUnit.calls.clear()
|
|
|
|
# Bind mount unit is gone: unmount is skipped, load mounts it again
|
|
systemd_service.response_get_unit = [
|
|
ERROR_NO_UNIT,
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
]
|
|
|
|
assert not coresys.mounts.bound_mounts[0].emergency
|
|
await coresys.mounts.reload_mount(mount.name)
|
|
|
|
assert systemd_service.StopUnit.calls == []
|
|
assert [call[0] for call in systemd_service.StartTransientUnit.calls] == [
|
|
"mnt-data-supervisor-media-media_test.mount"
|
|
]
|
|
|
|
|
|
async def test_reload_mount_probe_failure_triggers_systemd_reload(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
mount: Mount,
|
|
):
|
|
"""A failed probe drives the systemd reload."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.ReloadOrRestartUnit.calls.clear()
|
|
|
|
systemd_service.response_get_unit = [
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
]
|
|
with (
|
|
patch(
|
|
"supervisor.mounts.mount._probe_network_mount",
|
|
side_effect=OSError(errno.EHOSTDOWN, "Host is down"),
|
|
),
|
|
pytest.raises(MountActivationError),
|
|
):
|
|
await coresys.mounts.reload_mount(mount.name)
|
|
|
|
assert len(systemd_service.ReloadOrRestartUnit.calls) == 1
|
|
assert (
|
|
systemd_service.ReloadOrRestartUnit.calls[0][0]
|
|
== "mnt-data-supervisor-mounts-media_test.mount"
|
|
)
|
|
|
|
|
|
async def test_remove_mount(
|
|
coresys: CoreSys, all_dbus_services: dict[str, DBusServiceMock], mount: Mount
|
|
):
|
|
"""Test removing a mount."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_unit_service: SystemdUnitService = all_dbus_services["systemd_unit"]
|
|
systemd_unit_service.active_state = ["active", "inactive", "active", "inactive"]
|
|
systemd_service.StopUnit.calls.clear()
|
|
|
|
# Remove the mount
|
|
assert mount == await coresys.mounts.remove_mount(mount.name)
|
|
|
|
assert mount.state is None
|
|
assert mount not in coresys.mounts
|
|
|
|
assert [call[0] for call in systemd_service.StopUnit.calls] == [
|
|
"mnt-data-supervisor-media-media_test.mount",
|
|
"mnt-data-supervisor-mounts-media_test.mount",
|
|
]
|
|
|
|
|
|
async def test_remove_reload_mount_missing(coresys: CoreSys, mount_propagation):
|
|
"""Test removing or reloading a non existent mount errors."""
|
|
await coresys.mounts.load()
|
|
|
|
with pytest.raises(MountNotFound):
|
|
await coresys.mounts.remove_mount("does_not_exist")
|
|
|
|
with pytest.raises(MountNotFound):
|
|
await coresys.mounts.reload_mount("does_not_exist")
|
|
|
|
|
|
async def test_save_data(
|
|
coresys: CoreSys,
|
|
tmp_supervisor_data: Path,
|
|
path_extern,
|
|
mount_propagation,
|
|
mock_is_mount,
|
|
):
|
|
"""Test saving mount config data."""
|
|
# Replace mount manager with one that doesn't have save_data mocked
|
|
coresys._mounts = await MountManager(coresys).load_config() # pylint: disable=protected-access
|
|
|
|
path = tmp_supervisor_data / "mounts.json"
|
|
assert not path.exists()
|
|
|
|
await coresys.mounts.load()
|
|
await coresys.mounts.create_mount(
|
|
Mount.from_dict(
|
|
coresys,
|
|
{
|
|
"name": "auth_test",
|
|
"type": "cifs",
|
|
"usage": "backup",
|
|
"server": "backup.local",
|
|
"share": "backups",
|
|
"username": "admin",
|
|
"password": "password",
|
|
},
|
|
)
|
|
)
|
|
await coresys.mounts.save_data()
|
|
|
|
assert path.exists()
|
|
with path.open() as file:
|
|
config = json.load(file)
|
|
assert config["mounts"] == [
|
|
{
|
|
"version": None,
|
|
"name": "auth_test",
|
|
"type": "cifs",
|
|
"usage": "backup",
|
|
"server": "backup.local",
|
|
"share": "backups",
|
|
"username": "admin",
|
|
"password": "password",
|
|
"read_only": False,
|
|
}
|
|
]
|
|
|
|
|
|
async def test_create_mount_blocked_by_existing_local_data(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
mount_propagation,
|
|
mock_is_mount,
|
|
):
|
|
"""Test creating a media mount fails fast if the media directory has local data."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
|
|
await coresys.mounts.load()
|
|
|
|
media_dir = coresys.config.path_media / "media_test"
|
|
media_dir.mkdir()
|
|
(media_dir / "recording.mp4").touch()
|
|
|
|
with pytest.raises(MountTargetNotEmptyError):
|
|
await coresys.mounts.create_mount(Mount.from_dict(coresys, MEDIA_TEST_DATA))
|
|
|
|
assert "media_test" not in coresys.mounts
|
|
assert systemd_service.StartTransientUnit.calls == []
|
|
|
|
|
|
async def test_create_mount_blocked_by_non_directory_target(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
mount_propagation,
|
|
mock_is_mount,
|
|
):
|
|
"""Test creating a media mount fails fast if the media target is not a directory."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
|
|
await coresys.mounts.load()
|
|
|
|
(coresys.config.path_media / "media_test").touch()
|
|
|
|
with pytest.raises(MountTargetNotDirectoryError):
|
|
await coresys.mounts.create_mount(Mount.from_dict(coresys, MEDIA_TEST_DATA))
|
|
|
|
assert "media_test" not in coresys.mounts
|
|
assert systemd_service.StartTransientUnit.calls == []
|
|
|
|
|
|
async def test_update_mount_blocked_by_existing_local_data(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
mount: Mount,
|
|
):
|
|
"""Test updating a mount fails fast on local data without touching the mount.
|
|
|
|
Simulates the state after systemd tore down the bind mount and an add-on
|
|
wrote into the bare media directory: the update must not unmount the data
|
|
mount just to fail on the non-empty bind target afterwards.
|
|
"""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StopUnit.calls.clear()
|
|
|
|
media_dir = coresys.config.path_media / "media_test"
|
|
media_dir.mkdir(exist_ok=True)
|
|
(media_dir / "recording.mp4").touch()
|
|
|
|
with pytest.raises(MountTargetNotEmptyError):
|
|
await coresys.mounts.create_mount(Mount.from_dict(coresys, MEDIA_TEST_DATA))
|
|
|
|
assert mount == coresys.mounts.get("media_test")
|
|
assert mount.state == UnitActiveState.ACTIVE
|
|
assert systemd_service.StopUnit.calls == []
|
|
|
|
|
|
async def test_create_mount_bind_failure_rolls_back(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
mount_propagation,
|
|
mock_is_mount,
|
|
):
|
|
"""Test a bind mount failure during create unmounts the new data mount."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
systemd_service.StopUnit.calls.clear()
|
|
|
|
await coresys.mounts.load()
|
|
|
|
systemd_service.response_get_unit = {
|
|
"mnt-data-supervisor-mounts-media_test.mount": [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
],
|
|
"mnt-data-supervisor-media-media_test.mount": [ERROR_NO_UNIT],
|
|
}
|
|
|
|
with (
|
|
patch.object(
|
|
BindMount, "load", side_effect=MountError("Test bind mount failure")
|
|
),
|
|
pytest.raises(MountError),
|
|
):
|
|
await coresys.mounts.create_mount(Mount.from_dict(coresys, MEDIA_TEST_DATA))
|
|
|
|
assert "media_test" not in coresys.mounts
|
|
assert coresys.mounts.bound_mounts == []
|
|
assert [call[0] for call in systemd_service.StartTransientUnit.calls] == [
|
|
"mnt-data-supervisor-mounts-media_test.mount"
|
|
]
|
|
assert [call[0] for call in systemd_service.StopUnit.calls] == [
|
|
"mnt-data-supervisor-mounts-media_test.mount"
|
|
]
|
|
|
|
|
|
async def test_create_mount_start_unit_failure(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
mount_propagation,
|
|
):
|
|
"""Test failure to start mount unit does not add mount to the list."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
systemd_service.ResetFailedUnit.calls.clear()
|
|
systemd_service.StopUnit.calls.clear()
|
|
|
|
systemd_service.response_get_unit = ERROR_NO_UNIT
|
|
systemd_service.response_start_transient_unit = DBusError(ErrorType.FAILED, "fail")
|
|
|
|
await coresys.mounts.load()
|
|
|
|
mount = Mount.from_dict(coresys, BACKUP_TEST_DATA)
|
|
|
|
with pytest.raises(MountError):
|
|
await coresys.mounts.create_mount(mount)
|
|
|
|
assert mount.state is None
|
|
assert mount not in coresys.mounts
|
|
|
|
assert len(systemd_service.StartTransientUnit.calls) == 1
|
|
assert not systemd_service.ResetFailedUnit.calls
|
|
assert not systemd_service.StopUnit.calls
|
|
|
|
|
|
async def test_create_mount_activation_failure(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
mount_propagation,
|
|
):
|
|
"""Test activation failure during create mount does not add mount to the list and unmounts new mount."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_unit_service: SystemdUnitService = all_dbus_services["systemd_unit"]
|
|
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
systemd_service.ResetFailedUnit.calls.clear()
|
|
systemd_service.StopUnit.calls.clear()
|
|
|
|
systemd_service.response_get_unit = [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
]
|
|
systemd_unit_service.active_state = ["failed", "failed", "failed"]
|
|
|
|
await coresys.mounts.load()
|
|
|
|
mount = Mount.from_dict(coresys, BACKUP_TEST_DATA)
|
|
|
|
with pytest.raises(MountActivationError):
|
|
await coresys.mounts.create_mount(mount)
|
|
|
|
assert mount.state is None
|
|
assert mount not in coresys.mounts
|
|
|
|
assert len(systemd_service.StartTransientUnit.calls) == 1
|
|
assert len(systemd_service.ResetFailedUnit.calls) == 1
|
|
assert not systemd_service.StopUnit.calls
|
|
|
|
|
|
async def test_reload_mounts(
|
|
coresys: CoreSys, all_dbus_services: dict[str, DBusServiceMock], mount: Mount
|
|
):
|
|
"""Test reloading mounts."""
|
|
systemd_unit_service: SystemdUnitService = all_dbus_services["systemd_unit"]
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.ReloadOrRestartUnit.calls.clear()
|
|
|
|
assert mount.state == UnitActiveState.ACTIVE
|
|
assert mount.failed_issue not in coresys.resolution.issues
|
|
|
|
systemd_unit_service.active_state = "failed"
|
|
await coresys.mounts.reload()
|
|
|
|
assert mount.state == UnitActiveState.FAILED
|
|
assert mount.failed_issue in coresys.resolution.issues
|
|
assert len(coresys.resolution.suggestions_for_issue(mount.failed_issue)) == 2
|
|
assert len(systemd_service.ReloadOrRestartUnit.calls) == 1
|
|
|
|
# This should now remove the issue from the list
|
|
systemd_unit_service.active_state = "active"
|
|
await coresys.mounts.reload()
|
|
|
|
assert mount.state == UnitActiveState.ACTIVE
|
|
assert mount.failed_issue not in coresys.resolution.issues
|
|
assert not coresys.resolution.suggestions_for_issue(mount.failed_issue)
|
|
|
|
|
|
async def test_reload_mounts_attempts_initial_mount(
|
|
coresys: CoreSys, all_dbus_services: dict[str, DBusServiceMock], mount: Mount
|
|
):
|
|
"""Test reloading mounts attempts initial mount."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
systemd_service.response_get_unit = [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
ERROR_NO_UNIT,
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
]
|
|
systemd_service.response_reload_or_restart_unit = ERROR_NO_UNIT
|
|
|
|
await coresys.mounts.reload()
|
|
|
|
assert systemd_service.StartTransientUnit.calls == [
|
|
(
|
|
"mnt-data-supervisor-mounts-media_test.mount",
|
|
"fail",
|
|
[
|
|
("Options", Variant("s", "softerr,timeo=100,retrans=2")),
|
|
("Type", Variant("s", "nfs")),
|
|
("Description", Variant("s", "Supervisor nfs mount: media_test")),
|
|
("What", Variant("s", "media.local:/media")),
|
|
("TimeoutUSec", Variant("t", 35000000)),
|
|
],
|
|
[],
|
|
),
|
|
(
|
|
"mnt-data-supervisor-media-media_test.mount",
|
|
"fail",
|
|
[
|
|
("Options", Variant("s", "bind")),
|
|
("Description", Variant("s", "Supervisor bind mount: bind_media_test")),
|
|
("What", Variant("s", "/mnt/data/supervisor/mounts/media_test")),
|
|
("TimeoutUSec", Variant("t", 35000000)),
|
|
],
|
|
[],
|
|
),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("os_available", ["9.5"], indirect=True)
|
|
async def test_mounting_not_supported(
|
|
coresys: CoreSys,
|
|
caplog: pytest.LogCaptureFixture,
|
|
os_available,
|
|
):
|
|
"""Test mounting not supported on system."""
|
|
caplog.clear()
|
|
|
|
await coresys.mounts.load()
|
|
assert not caplog.text
|
|
|
|
mount = Mount.from_dict(coresys, MEDIA_TEST_DATA)
|
|
coresys.mounts._mounts = {"media_test": mount} # pylint: disable=protected-access
|
|
|
|
# Only tell the user about an issue here if they actually have mounts we couldn't load
|
|
# This is an edge case but users can downgrade OS so its possible
|
|
await coresys.mounts.load()
|
|
assert "Cannot load configured mounts" in caplog.text
|
|
|
|
with pytest.raises(MountJobError):
|
|
await coresys.mounts.create_mount(mount)
|
|
|
|
with pytest.raises(MountJobError):
|
|
await coresys.mounts.reload_mount("media_test")
|
|
|
|
with pytest.raises(MountJobError):
|
|
await coresys.mounts.remove_mount("media_test")
|
|
|
|
|
|
async def test_create_share_mount(
|
|
coresys: CoreSys,
|
|
all_dbus_services: dict[str, DBusServiceMock],
|
|
tmp_supervisor_data,
|
|
path_extern,
|
|
mount_propagation,
|
|
mock_is_mount,
|
|
):
|
|
"""Test creating a share mount."""
|
|
systemd_service: SystemdService = all_dbus_services["systemd"]
|
|
systemd_service.StartTransientUnit.calls.clear()
|
|
|
|
await coresys.mounts.load()
|
|
|
|
mount = Mount.from_dict(coresys, SHARE_TEST_DATA)
|
|
|
|
assert mount.state is None
|
|
assert mount not in coresys.mounts
|
|
assert "share_test" not in coresys.mounts
|
|
assert not mount.local_where.exists()
|
|
assert not any(coresys.config.path_share.iterdir())
|
|
|
|
# Create the mount
|
|
systemd_service.response_get_unit = [
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
ERROR_NO_UNIT,
|
|
"/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
|
|
]
|
|
await coresys.mounts.create_mount(mount)
|
|
|
|
assert mount.state == UnitActiveState.ACTIVE
|
|
assert mount in coresys.mounts
|
|
assert "share_test" in coresys.mounts
|
|
assert mount.local_where.exists()
|
|
assert (coresys.config.path_share / "share_test").exists()
|
|
|
|
assert [call[0] for call in systemd_service.StartTransientUnit.calls] == [
|
|
"mnt-data-supervisor-mounts-share_test.mount",
|
|
"mnt-data-supervisor-share-share_test.mount",
|
|
]
|