/
/
/
1"""Tests for the shared mount helpers."""
2
3from unittest.mock import AsyncMock, MagicMock, patch
4
5import pytest
6from music_assistant_models.errors import SetupFailedError
7
8from music_assistant.helpers.mount import error_summary, unmount
9
10MOUNT_PATH = "/tmp/filesystem_smb--test" # noqa: S108
11ISMOUNT = "music_assistant.helpers.mount.os.path.ismount"
12CHECK_OUTPUT = "music_assistant.helpers.mount.check_output"
13PLATFORM_SYSTEM = "music_assistant.helpers.mount.platform.system"
14
15
16def test_error_summary_drops_troubleshooting_pointer() -> None:
17 """The generic pointer line mount.cifs appends is left out of the summary."""
18 output = (
19 "mount error(16): Device or resource busy\n"
20 "Refer to the mount.cifs(8) manual page (e.g. man mount.cifs) "
21 "and kernel log messages (dmesg)"
22 )
23 assert error_summary(output) == "mount error(16): Device or resource busy"
24
25
26def test_error_summary_keeps_single_line_output() -> None:
27 """A single-line output is passed through unchanged."""
28 assert error_summary("mount: only root can do that") == "mount: only root can do that"
29
30
31async def test_unmount_skipped_when_not_mounted() -> None:
32 """A path that is not a mountpoint does not trigger any umount call."""
33 with (
34 patch(ISMOUNT, return_value=False),
35 patch(CHECK_OUTPUT, AsyncMock()) as check_output,
36 ):
37 await unmount(MOUNT_PATH, MagicMock())
38 check_output.assert_not_called()
39
40
41async def test_unmount_success() -> None:
42 """A successful umount is not escalated and not logged as a problem."""
43 logger = MagicMock()
44 with (
45 patch(ISMOUNT, return_value=True),
46 patch(CHECK_OUTPUT, AsyncMock(return_value=(0, b""))) as check_output,
47 ):
48 await unmount(MOUNT_PATH, logger)
49 check_output.assert_awaited_once_with("umount", MOUNT_PATH)
50 logger.warning.assert_not_called()
51
52
53async def test_unmount_busy_escalates_lazy_on_linux() -> None:
54 """A busy mountpoint is lazily detached on Linux and the failure is logged."""
55 logger = MagicMock()
56 check_output = AsyncMock(side_effect=[(1, b"umount: target is busy"), (0, b"")])
57 with (
58 patch(ISMOUNT, return_value=True),
59 patch(CHECK_OUTPUT, check_output),
60 patch(PLATFORM_SYSTEM, return_value="Linux"),
61 ):
62 await unmount(MOUNT_PATH, logger)
63 assert check_output.await_args_list[0].args == ("umount", MOUNT_PATH)
64 assert check_output.await_args_list[1].args == ("umount", "-l", MOUNT_PATH)
65 logger.warning.assert_called_once()
66
67
68async def test_unmount_busy_escalates_forced_on_macos() -> None:
69 """A busy mountpoint is force-detached on macOS, which has no lazy unmount."""
70 check_output = AsyncMock(side_effect=[(1, b"umount: target is busy"), (0, b"")])
71 with (
72 patch(ISMOUNT, return_value=True),
73 patch(CHECK_OUTPUT, check_output),
74 patch(PLATFORM_SYSTEM, return_value="Darwin"),
75 ):
76 await unmount(MOUNT_PATH, MagicMock())
77 assert check_output.await_args_list[1].args == ("umount", "-f", MOUNT_PATH)
78
79
80async def test_unmount_raises_when_still_mounted() -> None:
81 """A mountpoint that survives the escalation is reported as a setup failure."""
82 check_output = AsyncMock(side_effect=[(1, b"umount: target is busy"), (1, b"umount: failed")])
83 with (
84 patch(ISMOUNT, return_value=True),
85 patch(CHECK_OUTPUT, check_output),
86 pytest.raises(SetupFailedError) as exc_info,
87 ):
88 await unmount(MOUNT_PATH, MagicMock())
89 assert exc_info.value.translation_key == "unmount_failed"
90 assert exc_info.value.translation_args == ["umount: failed"]
91
92
93async def test_unmount_no_raise_when_detached() -> None:
94 """A non-zero escalation that did free the mountpoint is still a success."""
95 check_output = AsyncMock(side_effect=[(1, b"umount: target is busy"), (1, b"umount: failed")])
96 with (
97 patch(ISMOUNT, side_effect=[True, False]),
98 patch(CHECK_OUTPUT, check_output),
99 ):
100 await unmount(MOUNT_PATH, MagicMock())
101