diff --git a/app/modules/filemanager/storages/smb.py b/app/modules/filemanager/storages/smb.py index a1bfd32a0..71627e344 100644 --- a/app/modules/filemanager/storages/smb.py +++ b/app/modules/filemanager/storages/smb.py @@ -47,6 +47,7 @@ class SMB(StorageBase, metaclass=WeakSingleton): chunk_size = 10 * 1024 * 1024 def __init__(self): + """加载当前 SMB 配置并尝试建立连接。""" super().__init__() self._connected = False self._server_path = None @@ -58,8 +59,10 @@ class SMB(StorageBase, metaclass=WeakSingleton): def _init_connection(self): """ - 初始化SMB连接配置 + 按当前配置建立连接,清除上次状态以免配置无效时继续使用旧共享。 """ + self._connected = False + self._server_path = None from smbclient import ClientConfig, register_session try: conf = self.get_conf() @@ -228,7 +231,7 @@ class SMB(StorageBase, metaclass=WeakSingleton): def init_storage(self): """ - 初始化存储 + 保存或重置配置后清理旧会话并重新连接,即使配置未变或上次连接失败。 """ from smbclient import reset_connection_cache # 重置连接缓存 diff --git a/tests/test_smb_storage.py b/tests/test_smb_storage.py new file mode 100644 index 000000000..7aa50bb71 --- /dev/null +++ b/tests/test_smb_storage.py @@ -0,0 +1,85 @@ +"""SMB 保存配置后的重连与连接状态回归测试。""" + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +import smbclient + +from app.application.storage import StorageHelper +from app.modules.filemanager.module import FileManagerModule +from app.modules.filemanager.storages.smb import SMB, SMBConnectionError + + +@pytest.fixture +def smb_storage(monkeypatch): + """隔离配置持久化与 SMB 网络边界,保留真实保存和初始化流程。""" + state = {"conf": {}} + client = Mock() + client.listdir.return_value = [] + for name in ("ClientConfig", "register_session", "listdir", "reset_connection_cache"): + monkeypatch.setattr(smbclient, name, getattr(client, name)) + + def get_conf(_self): + """读取最近一次保存的配置。""" + return state["conf"].copy() + + def set_storage(_self, storage, conf): + """记录配置,确保重连发生在新配置保存之后。""" + assert storage == "smb" + state["conf"] = conf.copy() + + monkeypatch.setattr(SMB, "get_conf", get_conf) + monkeypatch.setattr(StorageHelper, "set_storage", set_storage) + storage = object.__new__(SMB) + SMB.__init__(storage) + module = object.__new__(FileManagerModule) + module._support_storages = ["smb"] + monkeypatch.setattr(module, "_FileManagerModule__get_storage_oper", Mock(return_value=storage)) + yield SimpleNamespace(storage=storage, module=module, client=client, state=state) + storage._connected = False + + +@pytest.mark.parametrize("failure_stage", ["register_session", "listdir"]) +def test_save_retries_unchanged_config_until_connection_recovers(smb_storage, failure_stage): + """连接或共享访问失败后,每次保存相同配置都重连,网络恢复即可使用。""" + conf = {"host": "smb.example", "share": "video", "username": "user", "password": "secret"} + getattr(smb_storage.client, failure_stage).side_effect = [OSError("unreachable"), OSError("unreachable"), []] + + for connected in (False, False, True): + smb_storage.client.reset_mock() + result = smb_storage.module.storage_manage(storage="smb", action="save_config", conf=conf) + + assert result["success"] is True + assert smb_storage.state["conf"] == conf + assert smb_storage.storage._connected is connected + calls = [call[0] for call in smb_storage.client.mock_calls] + assert calls[:3] == ["reset_connection_cache", "ClientConfig", "register_session"] + smb_storage.client.register_session.assert_called_once_with( + "smb.example", username="user", password="secret", port=445, encrypt=False, connection_timeout=60, + ) + if connected: + smb_storage.storage._check_connection() + smb_storage.client.listdir.assert_called_once_with(r"\\smb.example\video") + else: + with pytest.raises(SMBConnectionError, match="连接未建立"): + smb_storage.storage._check_connection() + + +@pytest.mark.parametrize("conf", [{}, {"host": "smb.example"}, {"share": "video"}]) +def test_save_incomplete_config_clears_previous_connection(smb_storage, conf): + """清空或保存不完整配置后,不得继续把旧共享标记为已连接。""" + smb_storage.module.storage_manage( + storage="smb", action="save_config", conf={"host": "smb.example", "share": "video"}, + ) + assert smb_storage.storage._connected is True + smb_storage.client.reset_mock() + + smb_storage.module.storage_manage(storage="smb", action="save_config", conf=conf) + + assert smb_storage.storage._connected is False + assert smb_storage.storage._server_path is None + smb_storage.client.reset_connection_cache.assert_called_once_with() + smb_storage.client.register_session.assert_not_called() + with pytest.raises(SMBConnectionError, match="连接未建立"): + smb_storage.storage._check_connection()