supervisor/tests/resolution/fixup/test_mount_execute_remove.py
Stefan Agner 11ec6dd9ac
Wait until mount unit is deactivated on unmount (#4733)
* Wait until mount unit is deactivated on unmount

The current code does not wait until the (bind) mount unit has been
actually deactivated (state "inactive"). This is especially problematic
when restoring a backup, where we deactivate all bind mounts before
restoring the target folder. Before the tarball is actually restored,
we delete all contents of the target folder. This lead to the situation
where the "rm -rf" command got executed before the bind mount actually
got unmounted.

The current code polls the state using an exponentially increasing
delay. Wait up to 30s for the bind mount to actually deactivate.

* Fix function name

* Fix missing await

* Address pytest errors

Change state of systemd unit according to use cases. Note that this
is currently rather fragile, and ideally we should have a smarter
mock service instead.

* Fix pylint

* Fix remaining

* Check transition fo failed as well

* Used alternative mocking mechanism

* Remove state lists in test_manager

---------

Co-authored-by: Mike Degatano <michael.degatano@gmail.com>
2023-12-01 00:35:15 +01:00

58 lines
1.8 KiB
Python

"""Test fixup mount remove."""
from supervisor.coresys import CoreSys
from supervisor.mounts.mount import Mount
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
from supervisor.resolution.fixups.mount_execute_remove import FixupMountExecuteRemove
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
async def test_fixup(
coresys: CoreSys,
all_dbus_services: dict[str, DBusServiceMock],
path_extern,
mount_propagation,
):
"""Test fixup."""
systemd_service: SystemdService = all_dbus_services["systemd"]
systemd_unit_service: SystemdUnitService = all_dbus_services["systemd_unit"]
systemd_service.StopUnit.calls.clear()
mount_execute_remove = FixupMountExecuteRemove(coresys)
assert mount_execute_remove.auto is False
await coresys.mounts.create_mount(
Mount.from_dict(
coresys,
{
"name": "test",
"usage": "backup",
"type": "cifs",
"server": "test.local",
"share": "test",
},
)
)
coresys.resolution.create_issue(
IssueType.MOUNT_FAILED,
ContextType.MOUNT,
reference="test",
suggestions=[SuggestionType.EXECUTE_RELOAD, SuggestionType.EXECUTE_REMOVE],
)
systemd_unit_service.active_state = ["active", "inactive"]
await mount_execute_remove()
assert coresys.resolution.issues == []
assert coresys.resolution.suggestions == []
assert coresys.mounts.mounts == []
assert systemd_service.StopUnit.calls == [
("mnt-data-supervisor-mounts-test.mount", "fail")
]
coresys.mounts.save_data.assert_called_once()