test: utils test expansion - network, command_parser, storage, tempdir, file_extract

tests/unit/test_agent_message.py

@@ -74,8 +74,9 @@ class TestMessageConstruction:
 
     def test_invalid_role_raises(self):
         """An invalid role is rejected."""
+        invalid_role: str = "invalid_role"
         with pytest.raises(ValidationError):
-            Message(role="invalid_role", content="hi")
+            Message(role=invalid_role, content="hi")
 
 
 class TestCheckpointMessage:
@@ -190,6 +191,7 @@ class TestMessageSegments:
         cp = CheckpointData(id="cp_1")
         msg = CheckpointMessageSegment(content=cp)
         assert msg.role == "_checkpoint"
+        assert msg.content is not None
         assert msg.content.id == "cp_1"
 
 

tests/unit/test_network_utils.py

@@ -111,9 +111,10 @@ class TestIsConnectionError:
         for exc in (
             httpx.HTTPStatusError("err", request=request, response=response),
             httpx.InvalidURL("invalid"),
-            httpx.CookieConflict("cookies"),
+            httpx.DecodingError,
         ):
-            assert is_connection_error(exc) is False
+            instance = exc("msg") if isinstance(exc, type) else exc
+            assert is_connection_error(instance) is False
 
     def test_cause_chain_unwraps_to_network_error(self):
         inner = httpx.ConnectError("inner")

tests/unit/test_provider_entities.py

@@ -228,7 +228,7 @@ class TestProviderRequest:
         tcr = MagicMock(spec=ToolCallsResult)
         req.append_tool_calls_result(tcr)
         assert isinstance(req.tool_calls_result, list)
-        assert len(req.tool_calls_result) == 1
+        assert len(req.tool_calls_result or []) == 1
         assert req.tool_calls_result[0] is tcr
 
     def test_append_tool_calls_result_single_to_list(self):
@@ -237,7 +237,7 @@ class TestProviderRequest:
         tcr2 = MagicMock(spec=ToolCallsResult)
         req.append_tool_calls_result(tcr2)
         assert isinstance(req.tool_calls_result, list)
-        assert len(req.tool_calls_result) == 2
+        assert len(req.tool_calls_result or []) == 2
         assert req.tool_calls_result[0] is tcr1
         assert req.tool_calls_result[1] is tcr2
 
@@ -247,7 +247,7 @@ class TestProviderRequest:
         req = ProviderRequest(tool_calls_result=[tcr1])
         tcr3 = MagicMock(spec=ToolCallsResult)
         req.append_tool_calls_result(tcr3)
-        assert len(req.tool_calls_result) == 2
+        assert len(req.tool_calls_result or []) == 2
 
     def test_print_friendly_context_no_contexts(self):
         req = ProviderRequest(prompt="hello", image_urls=["a.png"], audio_urls=["b.wav"])

tests/unit/test_storage_cleaner.py

@@ -5,7 +5,7 @@ Uses tmp_path for filesystem-level assertions.
 
 import os
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -215,33 +215,29 @@ class TestStorageCleanerEdgeCases:
         result = cleaner.cleanup("all")
         assert result["failed_files"] == 0
 
-    def test_cleanup_handles_stat_os_error(self, tmp_path: Path):
-        data = tmp_path / "data"
-        logs = data / "logs"
-        _make_file(logs / "astrbot.log", 100)
+    def test_summarize_files_catches_stat_os_error(self):
+        """_summarize_files should catch OSError from stat() and skip the file."""
+        bad = MagicMock()
+        bad.exists.return_value = True
+        bad.is_file.return_value = True
+        bad.stat.side_effect = OSError(13, "Permission denied")
 
-        cleaner = StorageCleaner(
-            {"log_file_enable": True, "log_file_path": "logs/astrbot.log"},
-            data_dir=data,
-            temp_dir=data / "temp",
+        good = MagicMock()
+        good.exists.return_value = True
+        good.is_file.return_value = True
+        good.stat.return_value = os.stat_result(
+            [0, 0, 0, 0, 0, 0, 200, 0, 0, 0]
         )
 
-        original_stat = (logs / "astrbot.log").stat
-
-        def _broken_stat():
-            raise OSError(13, "Permission denied")
-
-        (logs / "astrbot.log").stat = _broken_stat  # type: ignore[method-assign]
-
-        result = cleaner.cleanup("logs")
-        assert result["failed_files"] == 1
-
-        (logs / "astrbot.log").stat = original_stat
+        size, count = StorageCleaner._summarize_files([bad, good])
+        assert size == 200
+        assert count == 1
 
     def test_cleanup_handles_unlink_os_error(self, tmp_path: Path):
+        """_cleanup_target catches OSError from unlink() via mocked file."""
         data = tmp_path / "data"
         logs = data / "logs"
-        _make_file(logs / "astrbot.log", 100)
+        _make_file(logs / "astrbot.log", 50)
 
         cleaner = StorageCleaner(
             {"log_file_enable": False},
@@ -249,14 +245,17 @@ class TestStorageCleanerEdgeCases:
             temp_dir=data / "temp",
         )
 
-        original_unlink = (logs / "astrbot.log").unlink
+        bad_file = MagicMock()
+        bad_file.exists.return_value = True
+        bad_file.is_file.return_value = True
+        bad_file.stat.return_value = os.stat_result(
+            [0, 0, 0, 0, 0, 0, 100, 0, 0, 0]
+        )
+        bad_file.unlink.side_effect = OSError(13, "Permission denied")
+        bad_file.__eq__ = lambda self, other: False  # never in active_log_files
 
-        def _broken_unlink():
-            raise OSError(13, "Permission denied")
+        with patch.object(cleaner, "_collect_log_files", return_value={bad_file}):
+            result = cleaner.cleanup("logs")
 
-        (logs / "astrbot.log").unlink = _broken_unlink  # type: ignore[method-assign]
-
-        result = cleaner.cleanup("logs")
-        assert result["failed_files"] == 1
-
-        (logs / "astrbot.log").unlink = original_unlink
+        assert result["results"]["logs"]["failed_files"] == 1
+        assert result["results"]["logs"]["deleted_files"] == 0

tests/unit/test_temp_dir_cleaner.py

@@ -124,18 +124,17 @@ class TestCleanupOnce:
             temp_dir=temp_dir,
         )
 
-        def _broken_unlink():
-            raise OSError(13, "Permission denied")
-
-        (temp_dir / "bad.bin").unlink = _broken_unlink  # type: ignore[method-assign]
-
-        # Should log warning but not crash
-        with patch("astrbot.core.utils.temp_dir_cleaner.logger.warning") as mock_warn:
-            cleaner.cleanup_once()
-            mock_warn.assert_called()
-            assert any(
-                "Permission denied" in str(c) for c in mock_warn.call_args_list
-            )
+        # Patch unlink at the class level (instance-level setattr is not
+        # supported on Python 3.12+ pathlib which uses __slots__).
+        with patch.object(Path, "unlink", side_effect=OSError(13, "Permission denied")):
+            with patch(
+                "astrbot.core.utils.temp_dir_cleaner.logger.warning"
+            ) as mock_warn:
+                cleaner.cleanup_once()
+                mock_warn.assert_called()
+                assert any(
+                    "Permission denied" in str(c) for c in mock_warn.call_args_list
+                )
 
     def test_invalid_config_falls_back_to_default(self, tmp_path: Path):
         temp_dir = tmp_path / "temp"