feat(synctool): 增加同步功能
This commit is contained in:
305
tests/interface/test_synctool.py
Normal file
305
tests/interface/test_synctool.py
Normal file
@@ -0,0 +1,305 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
SyncScreen 和 SyncService 的测试.
|
||||
"""
|
||||
import pathlib
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch, Mock
|
||||
|
||||
from heurams.context import ConfigContext
|
||||
from heurams.services.config import ConfigFile
|
||||
from heurams.services.sync_service import SyncService, SyncConfig, SyncMode, ConflictStrategy
|
||||
|
||||
|
||||
class TestSyncServiceUnit(unittest.TestCase):
|
||||
"""SyncService 的单元测试."""
|
||||
|
||||
def setUp(self):
|
||||
"""在每个测试之前运行, 设置临时目录和模拟客户端."""
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.temp_path = pathlib.Path(self.temp_dir.name)
|
||||
|
||||
# 创建测试文件
|
||||
self.test_file = self.temp_path / "test.txt"
|
||||
self.test_file.write_text("测试内容")
|
||||
|
||||
# 模拟 WebDAV 客户端
|
||||
self.mock_client = MagicMock()
|
||||
|
||||
# 创建同步配置
|
||||
self.config = SyncConfig(
|
||||
enabled=True,
|
||||
url="https://example.com/dav/",
|
||||
username="test",
|
||||
password="test",
|
||||
remote_path="/heurams/",
|
||||
sync_mode=SyncMode.BIDIRECTIONAL,
|
||||
conflict_strategy=ConflictStrategy.NEWER,
|
||||
verify_ssl=True,
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
"""在每个测试之后清理."""
|
||||
self.temp_dir.cleanup()
|
||||
|
||||
@patch('heurams.services.sync_service.Client')
|
||||
def test_sync_service_initialization(self, mock_client_class):
|
||||
"""测试同步服务初始化."""
|
||||
mock_client_class.return_value = self.mock_client
|
||||
|
||||
service = SyncService(self.config)
|
||||
|
||||
# 验证客户端已创建
|
||||
mock_client_class.assert_called_once()
|
||||
self.assertIsNotNone(service.client)
|
||||
self.assertEqual(service.config, self.config)
|
||||
|
||||
@patch('heurams.services.sync_service.Client')
|
||||
def test_sync_service_disabled(self, mock_client_class):
|
||||
"""测试同步服务未启用."""
|
||||
config = SyncConfig(enabled=False)
|
||||
service = SyncService(config)
|
||||
|
||||
# 客户端不应初始化
|
||||
mock_client_class.assert_not_called()
|
||||
self.assertIsNone(service.client)
|
||||
|
||||
@patch('heurams.services.sync_service.Client')
|
||||
def test_test_connection_success(self, mock_client_class):
|
||||
"""测试连接测试成功."""
|
||||
mock_client_class.return_value = self.mock_client
|
||||
self.mock_client.list.return_value = []
|
||||
|
||||
service = SyncService(self.config)
|
||||
result = service.test_connection()
|
||||
|
||||
self.assertTrue(result)
|
||||
self.mock_client.list.assert_called_once()
|
||||
|
||||
@patch('heurams.services.sync_service.Client')
|
||||
def test_test_connection_failure(self, mock_client_class):
|
||||
"""测试连接测试失败."""
|
||||
mock_client_class.return_value = self.mock_client
|
||||
self.mock_client.list.side_effect = Exception("连接失败")
|
||||
|
||||
service = SyncService(self.config)
|
||||
result = service.test_connection()
|
||||
|
||||
self.assertFalse(result)
|
||||
self.mock_client.list.assert_called_once()
|
||||
|
||||
@patch('heurams.services.sync_service.Client')
|
||||
def test_upload_file(self, mock_client_class):
|
||||
"""测试上传单个文件."""
|
||||
mock_client_class.return_value = self.mock_client
|
||||
|
||||
service = SyncService(self.config)
|
||||
result = service.upload_file(self.test_file)
|
||||
|
||||
self.assertTrue(result)
|
||||
self.mock_client.upload_file.assert_called_once()
|
||||
|
||||
@patch('heurams.services.sync_service.Client')
|
||||
def test_download_file(self, mock_client_class):
|
||||
"""测试下载单个文件."""
|
||||
mock_client_class.return_value = self.mock_client
|
||||
|
||||
service = SyncService(self.config)
|
||||
remote_path = "/heurams/test.txt"
|
||||
local_path = self.temp_path / "downloaded.txt"
|
||||
|
||||
result = service.download_file(remote_path, local_path)
|
||||
|
||||
self.assertTrue(result)
|
||||
self.mock_client.download_file.assert_called_once()
|
||||
self.assertTrue(local_path.parent.exists())
|
||||
|
||||
@patch('heurams.services.sync_service.Client')
|
||||
def test_sync_directory_no_files(self, mock_client_class):
|
||||
"""测试同步空目录."""
|
||||
mock_client_class.return_value = self.mock_client
|
||||
self.mock_client.list.return_value = []
|
||||
self.mock_client.mkdir.return_value = None
|
||||
|
||||
service = SyncService(self.config)
|
||||
result = service.sync_directory(self.temp_path)
|
||||
|
||||
self.assertTrue(result['success'])
|
||||
self.assertEqual(result['uploaded'], 0)
|
||||
self.assertEqual(result['downloaded'], 0)
|
||||
self.mock_client.mkdir.assert_called_once()
|
||||
|
||||
@patch('heurams.services.sync_service.Client')
|
||||
def test_sync_directory_upload_only(self, mock_client_class):
|
||||
"""测试仅上传模式."""
|
||||
mock_client_class.return_value = self.mock_client
|
||||
self.mock_client.list.return_value = []
|
||||
self.mock_client.mkdir.return_value = None
|
||||
|
||||
config = SyncConfig(
|
||||
enabled=True,
|
||||
url="https://example.com/dav/",
|
||||
username="test",
|
||||
password="test",
|
||||
remote_path="/heurams/",
|
||||
sync_mode=SyncMode.UPLOAD_ONLY,
|
||||
conflict_strategy=ConflictStrategy.NEWER,
|
||||
)
|
||||
|
||||
service = SyncService(config)
|
||||
result = service.sync_directory(self.temp_path)
|
||||
|
||||
self.assertTrue(result['success'])
|
||||
self.mock_client.mkdir.assert_called_once()
|
||||
|
||||
@patch('heurams.services.sync_service.Client')
|
||||
def test_conflict_strategy_newer(self, mock_client_class):
|
||||
"""测试 NEWER 冲突策略."""
|
||||
mock_client_class.return_value = self.mock_client
|
||||
|
||||
# 模拟远程文件存在
|
||||
self.mock_client.list.return_value = ["test.txt"]
|
||||
self.mock_client.info.return_value = {'size': 100, 'modified': '2023-01-01T00:00:00Z'}
|
||||
self.mock_client.mkdir.return_value = None
|
||||
|
||||
service = SyncService(self.config)
|
||||
result = service.sync_directory(self.temp_path)
|
||||
|
||||
self.assertTrue(result['success'])
|
||||
# 应该有一个冲突
|
||||
self.assertGreaterEqual(result.get('conflicts', 0), 0)
|
||||
|
||||
@patch('heurams.services.sync_service.Client')
|
||||
def test_create_sync_service_from_config(self, mock_client_class):
|
||||
"""测试从配置文件创建同步服务."""
|
||||
mock_client_class.return_value = self.mock_client
|
||||
|
||||
# 创建临时配置文件
|
||||
config_data = {
|
||||
'sync': {
|
||||
'webdav': {
|
||||
'enabled': True,
|
||||
'url': 'https://example.com/dav/',
|
||||
'username': 'test',
|
||||
'password': 'test',
|
||||
'remote_path': '/heurams/',
|
||||
'sync_mode': 'bidirectional',
|
||||
'conflict_strategy': 'newer',
|
||||
'verify_ssl': True,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
# 模拟 config_var
|
||||
with patch('heurams.services.sync_service.config_var') as mock_config_var:
|
||||
mock_config = MagicMock()
|
||||
mock_config.data = config_data
|
||||
mock_config_var.get.return_value = mock_config
|
||||
|
||||
from heurams.services.sync_service import create_sync_service_from_config
|
||||
service = create_sync_service_from_config()
|
||||
|
||||
self.assertIsNotNone(service)
|
||||
self.assertIsNotNone(service.client)
|
||||
|
||||
|
||||
class TestSyncScreenUnit(unittest.TestCase):
|
||||
"""SyncScreen 的单元测试."""
|
||||
|
||||
def setUp(self):
|
||||
"""在每个测试之前运行."""
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.temp_path = pathlib.Path(self.temp_dir.name)
|
||||
|
||||
# 创建默认配置
|
||||
default_config_path = (
|
||||
pathlib.Path(__file__).parent.parent.parent
|
||||
/ "src/heurams/default/config/config.toml"
|
||||
)
|
||||
self.config = ConfigFile(default_config_path)
|
||||
|
||||
# 更新配置中的路径
|
||||
config_data = self.config.data
|
||||
config_data["paths"]["nucleon_dir"] = str(self.temp_path / "nucleon")
|
||||
config_data["paths"]["electron_dir"] = str(self.temp_path / "electron")
|
||||
config_data["paths"]["orbital_dir"] = str(self.temp_path / "orbital")
|
||||
config_data["paths"]["cache_dir"] = str(self.temp_path / "cache")
|
||||
|
||||
# 添加同步配置
|
||||
if 'sync' not in config_data:
|
||||
config_data['sync'] = {}
|
||||
config_data['sync']['webdav'] = {
|
||||
'enabled': False,
|
||||
'url': '',
|
||||
'username': '',
|
||||
'password': '',
|
||||
'remote_path': '/heurams/',
|
||||
'sync_mode': 'bidirectional',
|
||||
'conflict_strategy': 'newer',
|
||||
'verify_ssl': True,
|
||||
}
|
||||
|
||||
# 创建目录
|
||||
for dir_key in ["nucleon_dir", "electron_dir", "orbital_dir", "cache_dir"]:
|
||||
pathlib.Path(config_data["paths"][dir_key]).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 使用 ConfigContext 设置配置
|
||||
self.config_ctx = ConfigContext(self.config)
|
||||
self.config_ctx.__enter__()
|
||||
|
||||
def tearDown(self):
|
||||
"""在每个测试之后清理."""
|
||||
self.config_ctx.__exit__(None, None, None)
|
||||
self.temp_dir.cleanup()
|
||||
|
||||
@patch('heurams.interface.screens.synctool.create_sync_service_from_config')
|
||||
def test_sync_screen_compose(self, mock_create_service):
|
||||
"""测试 SyncScreen 的 compose 方法."""
|
||||
from heurams.interface.screens.synctool import SyncScreen
|
||||
|
||||
# 模拟同步服务
|
||||
mock_service = MagicMock()
|
||||
mock_service.client = MagicMock()
|
||||
mock_create_service.return_value = mock_service
|
||||
|
||||
screen = SyncScreen()
|
||||
|
||||
# 测试 compose 方法
|
||||
from textual.app import ComposeResult
|
||||
result = screen.compose()
|
||||
widgets = list(result)
|
||||
|
||||
# 检查基本部件
|
||||
from textual.widgets import Footer, Header, Button, Static, ProgressBar
|
||||
from textual.containers import ScrollableContainer
|
||||
|
||||
header_present = any(isinstance(w, Header) for w in widgets)
|
||||
footer_present = any(isinstance(w, Footer) for w in widgets)
|
||||
self.assertTrue(header_present)
|
||||
self.assertTrue(footer_present)
|
||||
|
||||
# 检查容器
|
||||
container_present = any(isinstance(w, ScrollableContainer) for w in widgets)
|
||||
self.assertTrue(container_present)
|
||||
|
||||
@patch('heurams.interface.screens.synctool.create_sync_service_from_config')
|
||||
def test_sync_screen_load_config(self, mock_create_service):
|
||||
"""测试 SyncScreen 加载配置."""
|
||||
from heurams.interface.screens.synctool import SyncScreen
|
||||
|
||||
mock_service = MagicMock()
|
||||
mock_service.client = MagicMock()
|
||||
mock_create_service.return_value = mock_service
|
||||
|
||||
screen = SyncScreen()
|
||||
screen.load_config()
|
||||
|
||||
# 验证配置已加载
|
||||
self.assertIsNotNone(screen.sync_config)
|
||||
mock_create_service.assert_called_once()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user