fix judgment for tos

Author: ZQlQZ

tests/test_tos.py

@@ -14,14 +14,29 @@
 
 import pytest
 from unittest import mock
-import veadk.integrations.ve_tos.ve_tos as tos_mod
+
+# Check if tos module is available
+import importlib
+
+TOS_AVAILABLE = False
+try:
+    importlib.import_module("veadk.integrations.ve_tos.ve_tos")
+    TOS_AVAILABLE = True
+except ImportError:
+    pass
+
+# Skip tests that require tos module if it's not available
+require_tos = pytest.mark.skipif(not TOS_AVAILABLE, reason="tos module not available")
 
 # 使用 pytest-asyncio
 pytest_plugins = ("pytest_asyncio",)
 
 
 @pytest.fixture
+@require_tos
 def mock_client(monkeypatch):
+    import veadk.integrations.ve_tos.ve_tos as tos_mod
+
     fake_client = mock.Mock()
 
     monkeypatch.setenv("DATABASE_TOS_REGION", "test-region")
@@ -33,9 +48,17 @@ def mock_client(monkeypatch):
 
     class FakeExceptions:
         class TosServerError(Exception):
-            def __init__(self, msg):
+            def __init__(
+                self,
+                msg: str,
+                code: int = 0,
+                host_id: str = "",
+                resource: str = "",
+                request_id: str = "",
+                header=None,
+            ):
                 super().__init__(msg)
-                self.status_code = None
+                self.status_code = code
 
     monkeypatch.setattr(tos_mod.tos, "exceptions", FakeExceptions)
     monkeypatch.setattr(
@@ -51,27 +74,34 @@ def __init__(self, msg):
 
 
 @pytest.fixture
+@require_tos
 def tos_client(mock_client):
+    import veadk.integrations.ve_tos.ve_tos as tos_mod
+
     return tos_mod.VeTOS()
 
 
+@require_tos
 def test_create_bucket_exists(tos_client, mock_client):
     mock_client.head_bucket.return_value = None  # head_bucket 正常返回表示存在
     result = tos_client.create_bucket()
     assert result is True
     mock_client.create_bucket.assert_not_called()
 
 
+@require_tos
 def test_create_bucket_not_exists(tos_client, mock_client):
-    exc = tos_mod.tos.exceptions.TosServerError("not found")
-    exc.status_code = 404
+    import veadk.integrations.ve_tos.ve_tos as tos_mod
+
+    exc = tos_mod.tos.exceptions.TosServerError(msg="not found", code=404)
     mock_client.head_bucket.side_effect = exc
 
     result = tos_client.create_bucket()
     assert result is True
     mock_client.create_bucket.assert_called_once()
 
 
+@require_tos
 @pytest.mark.asyncio
 async def test_upload_bytes_success(tos_client, mock_client):
     mock_client.head_bucket.return_value = True
@@ -83,6 +113,7 @@ async def test_upload_bytes_success(tos_client, mock_client):
     mock_client.close.assert_called_once()
 
 
+@require_tos
 @pytest.mark.asyncio
 async def test_upload_file_success(tmp_path, tos_client, mock_client):
     mock_client.head_bucket.return_value = True
@@ -95,6 +126,7 @@ async def test_upload_file_success(tmp_path, tos_client, mock_client):
     mock_client.close.assert_called_once()
 
 
+@require_tos
 def test_download_success(tmp_path, tos_client, mock_client):
     save_path = tmp_path / "out.txt"
     mock_client.get_object.return_value = [b"abc", b"def"]
@@ -104,7 +136,32 @@ def test_download_success(tmp_path, tos_client, mock_client):
     assert save_path.read_bytes() == b"abcdef"
 
 
+@require_tos
 def test_download_fail(tos_client, mock_client):
     mock_client.get_object.side_effect = Exception("boom")
     result = tos_client.download("obj-key", "somewhere.txt")
     assert result is False
+
+
+@require_tos
+@pytest.mark.skipif(TOS_AVAILABLE, reason="tos module is available")
+def test_tos_import_error():
+    """Test VeTOS behavior when tos module is not installed"""
+    # Remove tos from sys.modules to simulate it's not installed
+    import sys
+
+    original_tos = sys.modules.get("tos")
+    if "tos" in sys.modules:
+        del sys.modules["tos"]
+
+    try:
+        # Try to import ve_tos module, which should raise ImportError
+        with pytest.raises(ImportError) as exc_info:
+            pass
+
+        # Check that the error message contains installation instructions
+        assert "pip install tos" in str(exc_info.value)
+    finally:
+        # Restore original state
+        if original_tos is not None:
+            sys.modules["tos"] = original_tos

veadk/runner.py

@@ -100,6 +100,8 @@ def _convert_messages(
             tos_url = "<tos_url>"
             if upload_inline_data_to_tos:
                 try:
+                    from veadk.integrations.ve_tos.ve_tos import VeTOS
+
                     ve_tos = VeTOS()
                     object_key, tos_url = ve_tos.build_tos_url(
                         self.user_id, self.app_name, session_id, messages.media