diff --git a/tests/unit/test_agent_message.py b/tests/unit/test_agent_message.py index b205a23f5..6a39ce21a 100644 --- a/tests/unit/test_agent_message.py +++ b/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" diff --git a/tests/unit/test_network_utils.py b/tests/unit/test_network_utils.py index 6a41097f8..3c73deae8 100644 --- a/tests/unit/test_network_utils.py +++ b/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") diff --git a/tests/unit/test_provider_entities.py b/tests/unit/test_provider_entities.py index 469385e39..c4e369a2d 100644 --- a/tests/unit/test_provider_entities.py +++ b/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"]) diff --git a/tests/unit/test_storage_cleaner.py b/tests/unit/test_storage_cleaner.py index 76edb72f6..506277895 100644 --- a/tests/unit/test_storage_cleaner.py +++ b/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 diff --git a/tests/unit/test_temp_dir_cleaner.py b/tests/unit/test_temp_dir_cleaner.py index 5c16d1eea..74c87c1dc 100644 --- a/tests/unit/test_temp_dir_cleaner.py +++ b/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"