"""
基本功能集成测试
"""
import pytest
import tempfile
import os
from fs_mcp.rustfs_client import RustFSClient
from fs_mcp.upload_tool import create_upload_tool
from fs_mcp.download_tool import create_download_tool
class TestBasicFunctionality:
"""基本功能集成测试"""
@pytest.mark.integration
def test_rustfs_client_creation(self, fs_client):
"""测试RustFS客户端创建"""
assert fs_client is not None
assert hasattr(fs_client, 'upload_file')
assert hasattr(fs_client, 'download_file')
assert hasattr(fs_client, 'parse_rustfs_url')
@pytest.mark.integration
def test_upload_tool_creation(self, upload_tool):
"""测试上传工具创建"""
assert upload_tool is not None
assert callable(upload_tool)
@pytest.mark.integration
def test_download_tool_creation(self, download_tool):
"""测试下载工具创建"""
assert download_tool is not None
assert callable(download_tool)
@pytest.mark.integration
def test_url_parsing(self, fs_client):
"""测试URL解析功能"""
# RustFS URL
rustfs_url = "http://jk.hazel.sale:49000/agentfiles/test.txt"
result = fs_client.parse_rustfs_url(rustfs_url)
assert result is not None
assert result == ("agentfiles", "test.txt")
# 非RustFS URL
http_url = "https://example.com/file.txt"
result = fs_client.parse_rustfs_url(http_url)
assert result is None
@pytest.mark.slow
@pytest.mark.integration
def test_basic_upload_download_cycle(self, fs_client, sample_file):
"""测试基本上传下载循环"""
# 使用临时文件测试
try:
# 上传文件
result = fs_client.upload_file(sample_file)
assert result["success"] is True
assert result["bucket"] == "agentfiles"
assert "filename" in result
assert "access_url" in result
# 获取文件信息
file_info = fs_client.get_file_info(result["filename"])
assert file_info is not None
assert file_info["content_length"] > 0
except Exception as e:
pytest.skip(f"Upload test skipped due to: {e}")
