fix(smb): clear stale state before reconnecting storage

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
         # 重置连接缓存

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()