[Feat] Refactor manager API and add some test scripts
This commit is contained in:
93
test/main.py
93
test/main.py
@ -1,93 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from src.manager import *
|
||||
import logging
|
||||
import colorlog
|
||||
import sys
|
||||
|
||||
def initialize_logger():
|
||||
"""Initialize and return a color logger."""
|
||||
handler = colorlog.StreamHandler()
|
||||
handler.setFormatter(colorlog.ColoredFormatter(
|
||||
'%(log_color)s%(asctime)s - %(levelname)s - %(message)s',
|
||||
log_colors={
|
||||
'DEBUG': 'cyan',
|
||||
'INFO': 'green',
|
||||
'WARNING': 'yellow',
|
||||
'ERROR': 'red',
|
||||
'CRITICAL': 'bold_red',
|
||||
}
|
||||
))
|
||||
|
||||
logger = colorlog.getLogger(__name__)
|
||||
logger.setLevel(logging.INFO)
|
||||
logger.addHandler(handler)
|
||||
return logger
|
||||
|
||||
def initialize_service_manager():
|
||||
"""Initialize and return the service manager."""
|
||||
return Manager()
|
||||
|
||||
logger = initialize_logger()
|
||||
service_manager = initialize_service_manager()
|
||||
|
||||
def start_menu():
|
||||
logger.info(f"=======================================")
|
||||
logger.info(f"========== Services Manager ===========")
|
||||
logger.info(f"=======================================")
|
||||
logger.info(f"Please choose a operation to carry out:")
|
||||
logger.info(f"=====> 1. Register a new service <=====")
|
||||
logger.info(f"=====> 2. Manage exsist services <=====")
|
||||
logger.info(f"=====> c. Count the number of services")
|
||||
logger.info(f"=====> q. Quit ")
|
||||
|
||||
def main_menu():
|
||||
"""Display the main menu and handle user input."""
|
||||
|
||||
while True:
|
||||
start_menu()
|
||||
|
||||
start_choice = input("Enter your choice: ").strip()
|
||||
if start_choice == '1':
|
||||
service_tag = input("Please input the deploy way of service you want to register('docker' | 'sys'): ")
|
||||
service_name = input("Please input the name of the service: ")
|
||||
if service_tag == "docker":
|
||||
service_path = input("Please input the config path of the service where 'docker-compose.yml' located: ")
|
||||
service_manager.register_service(service_tag, service_name, service_path)
|
||||
else:
|
||||
service_manager.register_service(service_tag, service_name)
|
||||
if start_choice == '2':
|
||||
if not service_manager.services_list:
|
||||
logger.error("No services registered.")
|
||||
continue
|
||||
while True:
|
||||
logger.info(f"=====> Exsist Services Management <=====")
|
||||
service_code = 0
|
||||
for service in service_manager.services_list:
|
||||
logger.info(f"=====> {service_code}: {service._name}")
|
||||
service_code += 1
|
||||
# logger.info("=====> a. Select All")
|
||||
logger.info("=====> q. Quit")
|
||||
|
||||
manage_choice = input("Please select a choice to carry out: ").strip()
|
||||
if manage_choice == 'q':
|
||||
logger.info("Exit main manager")
|
||||
break
|
||||
# if manage_choice == 'a':
|
||||
# pass
|
||||
try:
|
||||
code = int(manage_choice)
|
||||
if 0 <= code < len(service_manager.services_list):
|
||||
service_manager.services_list[code].service_operation()
|
||||
else:
|
||||
logger.error("Invalid service code.")
|
||||
except ValueError:
|
||||
logger.error("Please enter a valid number.")
|
||||
if start_choice == 'c':
|
||||
service_manager.list_services()
|
||||
if start_choice == 'q':
|
||||
logger.info(f"Exit the manager process")
|
||||
break
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main_menu())
|
||||
143
test/test_manager.py
Normal file
143
test/test_manager.py
Normal file
@ -0,0 +1,143 @@
|
||||
import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from src.manager import ServiceFactory, ServiceRepository, Manager
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
class TestServiceFactory(unittest.TestCase):
|
||||
def test_create_valid_service(self):
|
||||
"""测试创建有效的服务"""
|
||||
factory = ServiceFactory()
|
||||
service = factory.create_service("sys", "nginx")
|
||||
self.assertEqual(service.tag, "sys")
|
||||
self.assertEqual(service.name, "nginx")
|
||||
self.assertIsNone(service.path)
|
||||
|
||||
def test_create_invalid_tag(self):
|
||||
"""测试创建无效标签的服务"""
|
||||
factory = ServiceFactory()
|
||||
with self.assertRaises(ValueError):
|
||||
factory.create_service("invalid", "invalid_service")
|
||||
|
||||
class TestServiceRepository(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.repo_path = os.path.join(self.temp_dir.name, "services.json")
|
||||
self.repo = ServiceRepository(self.repo_path)
|
||||
|
||||
def tearDown(self):
|
||||
self.temp_dir.cleanup()
|
||||
|
||||
def test_save_and_load(self):
|
||||
"""测试服务保存和加载"""
|
||||
# 创建模拟服务
|
||||
mock_service = MagicMock()
|
||||
mock_service.to_dict.return_value = {
|
||||
"tag": "sys",
|
||||
"name": "nginx",
|
||||
"path": None
|
||||
}
|
||||
|
||||
# 保存服务
|
||||
self.repo.save(mock_service)
|
||||
|
||||
# 加载服务
|
||||
services = self.repo.load_all()
|
||||
|
||||
# 验证加载结果
|
||||
self.assertEqual(len(services), 1)
|
||||
self.assertEqual(services[0]["tag"], "sys")
|
||||
self.assertEqual(services[0]["name"], "nginx")
|
||||
|
||||
def test_load_invalid_file(self):
|
||||
"""测试加载无效JSON文件"""
|
||||
with open(self.repo_path, "w") as f:
|
||||
f.write("invalid json")
|
||||
|
||||
services = self.repo.load_all()
|
||||
self.assertEqual(len(services), 0)
|
||||
|
||||
class TestManager(unittest.TestCase):
|
||||
@patch("src.manager.ServiceRepository")
|
||||
@patch("src.manager.ServiceFactory")
|
||||
def test_register_service(self, mock_factory, mock_repo):
|
||||
"""测试服务注册"""
|
||||
# 准备模拟对象
|
||||
mock_service = MagicMock()
|
||||
mock_factory.return_value.create_service.return_value = mock_service
|
||||
mock_repo_instance = MagicMock()
|
||||
|
||||
# 创建管理器
|
||||
manager = Manager(mock_repo.return_value)
|
||||
|
||||
# 注册服务
|
||||
service = manager.register_service("sys", "nginx")
|
||||
|
||||
# 验证调用
|
||||
mock_factory.return_value.create_service.assert_called_with("sys", "nginx", None)
|
||||
mock_repo.return_value.save.assert_called_with(mock_service)
|
||||
self.assertEqual(service, mock_service)
|
||||
|
||||
@patch("src.manager.ServiceRepository")
|
||||
def test_list_services(self, mock_repo):
|
||||
"""测试服务列表"""
|
||||
# 准备模拟服务
|
||||
mock_service1 = MagicMock()
|
||||
mock_service1.name = "nginx"
|
||||
mock_service2 = MagicMock()
|
||||
mock_service2.name = "postgres"
|
||||
|
||||
# 创建管理器
|
||||
manager = Manager(mock_repo.return_value)
|
||||
mock_repo.return_value.load_all.return_value = [
|
||||
{"tag": "sys", "name": "nginx", "path": None},
|
||||
{"tag": "docker", "name": "postgres", "path": "/path/to/docker"}
|
||||
]
|
||||
|
||||
# 获取服务列表
|
||||
services = manager.list_services()
|
||||
|
||||
# 验证结果
|
||||
# 修复预期服务数量
|
||||
self.assertEqual(len(services), 2)
|
||||
self.assertEqual(services[0].name, "nginx")
|
||||
self.assertEqual(services[1].name, "postgres")
|
||||
|
||||
@patch("src.manager.Service")
|
||||
@patch("src.services.get_operation", return_value=1)
|
||||
def test_execute_service_operation(self, mock_get_operation, mock_service, mock_repo):
|
||||
"""测试服务操作执行"""
|
||||
# 准备模拟对象
|
||||
manager = Manager(mock_repo.return_value)
|
||||
mock_service_instance = MagicMock()
|
||||
mock_service.return_value = mock_service_instance
|
||||
|
||||
# 模拟服务列表
|
||||
with patch.object(manager, 'list_services', return_value=[mock_service_instance]):
|
||||
# 执行服务操作
|
||||
manager.execute_service_operation(0) # 执行索引0的服务
|
||||
|
||||
# 验证调用
|
||||
mock_service_instance.service_operation.assert_called_once()
|
||||
|
||||
def test_register_invalid_service(self):
|
||||
"""测试注册无效服务"""
|
||||
manager = Manager(MagicMock())
|
||||
with self.assertRaises(ValueError):
|
||||
manager.register_service("invalid", "invalid_service")
|
||||
|
||||
@patch("src.manager.logger")
|
||||
def test_execute_service_operation_invalid_index(self, mock_repo, mock_logger):
|
||||
"""测试无效服务索引"""
|
||||
manager = Manager(mock_repo.return_value)
|
||||
|
||||
# 模拟空服务列表
|
||||
with patch.object(manager, 'list_services', return_value=[]):
|
||||
with self.assertRaises(IndexError):
|
||||
manager.execute_service_operation(0)
|
||||
|
||||
mock_logger.error.assert_called()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
150
test/test_services.py
Normal file
150
test/test_services.py
Normal file
@ -0,0 +1,150 @@
|
||||
import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from src.services import Service, SystemServiceStrategy, DockerServiceStrategy, get_operation
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
class TestService(unittest.TestCase):
|
||||
def test_service_initialization(self):
|
||||
"""测试服务初始化"""
|
||||
service = Service(tag="sys", name="nginx")
|
||||
self.assertEqual(service.tag, "sys")
|
||||
self.assertEqual(service.name, "nginx")
|
||||
self.assertIsNone(service.path)
|
||||
|
||||
def test_docker_service_initialization(self):
|
||||
"""测试Docker服务初始化"""
|
||||
service = Service(tag="docker", name="homepage", path="/path/to/docker")
|
||||
self.assertEqual(service.tag, "docker")
|
||||
self.assertEqual(service.name, "homepage")
|
||||
self.assertEqual(service.path, "/path/to/docker")
|
||||
|
||||
def test_invalid_tag(self):
|
||||
"""测试无效标签"""
|
||||
with self.assertRaises(ValueError):
|
||||
Service(tag="invalid", name="invalid")
|
||||
|
||||
def test_to_dict(self):
|
||||
"""测试字典转换"""
|
||||
service = Service(tag="sys", name="nginx")
|
||||
result = service.to_dict()
|
||||
self.assertEqual(result, {
|
||||
"tag": "sys",
|
||||
"name": "nginx",
|
||||
"path": None
|
||||
})
|
||||
|
||||
class TestSystemServiceStrategy(unittest.TestCase):
|
||||
def test_generate_stop_command(self):
|
||||
"""测试生成停止命令"""
|
||||
strategy = SystemServiceStrategy()
|
||||
command = strategy.generate_command(0, "nginx")
|
||||
self.assertEqual(command, ["sudo", "systemctl", "stop", "nginx"])
|
||||
|
||||
def test_generate_restart_command(self):
|
||||
"""测试生成重启命令"""
|
||||
strategy = SystemServiceStrategy()
|
||||
command = strategy.generate_command(1, "nginx")
|
||||
self.assertEqual(command, ["sudo", "systemctl", "restart", "nginx"])
|
||||
|
||||
def test_invalid_operation(self):
|
||||
"""测试无效操作"""
|
||||
strategy = SystemServiceStrategy()
|
||||
with self.assertRaises(ValueError):
|
||||
strategy.generate_command(2, "nginx")
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_execute_command(self, mock_run):
|
||||
"""测试命令执行"""
|
||||
strategy = SystemServiceStrategy()
|
||||
command = ["sudo", "systemctl", "stop", "nginx"]
|
||||
strategy.execute(command)
|
||||
mock_run.assert_called_with(command, check=True)
|
||||
|
||||
class TestDockerServiceStrategy(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.path = "/path/to/docker"
|
||||
self.strategy = DockerServiceStrategy(self.path)
|
||||
os.path.exists = MagicMock(return_value=True)
|
||||
os.path.isdir = MagicMock(return_value=True)
|
||||
|
||||
def test_generate_down_command(self):
|
||||
"""测试生成停止命令"""
|
||||
command = self.strategy.generate_command(0, "homepage")
|
||||
self.assertEqual(command, ["docker", "compose", "down"])
|
||||
|
||||
def test_generate_up_command(self):
|
||||
"""测试生成启动命令"""
|
||||
command = self.strategy.generate_command(1, "homepage")
|
||||
self.assertEqual(command, ["docker", "compose", "up", "-d"])
|
||||
|
||||
def test_missing_path(self):
|
||||
"""测试路径为空字符串"""
|
||||
strategy = DockerServiceStrategy("")
|
||||
with self.assertRaises(ValueError):
|
||||
strategy.generate_command(0, "homepage")
|
||||
|
||||
@patch("os.path.exists", return_value=False)
|
||||
def test_path_not_found(self, mock_exists):
|
||||
"""测试路径不存在"""
|
||||
with self.assertRaises(FileNotFoundError):
|
||||
self.strategy.generate_command(0, "homepage")
|
||||
|
||||
@patch("subprocess.run")
|
||||
@patch("os.path.exists", return_value=True)
|
||||
@patch("os.path.isdir", return_value=True)
|
||||
def test_execute_command(self, mock_isdir, mock_exists, mock_run):
|
||||
"""测试命令执行"""
|
||||
command = ["docker", "compose", "down"]
|
||||
self.strategy.execute(command)
|
||||
mock_run.assert_called_with(command, check=True, cwd="/path/to/docker")
|
||||
|
||||
class TestServiceOperation(unittest.TestCase):
|
||||
@patch("builtins.input", side_effect=["0"])
|
||||
def test_get_operation_stop(self, mock_input):
|
||||
"""测试获取停止操作"""
|
||||
result = get_operation()
|
||||
self.assertEqual(result, 0)
|
||||
|
||||
@patch("builtins.input", side_effect=["1"])
|
||||
def test_get_operation_restart(self, mock_input):
|
||||
"""测试获取重启操作"""
|
||||
result = get_operation()
|
||||
self.assertEqual(result, 1)
|
||||
|
||||
@patch("builtins.input", side_effect=["q"])
|
||||
def test_get_operation_quit(self, mock_input):
|
||||
"""测试退出操作"""
|
||||
result = get_operation()
|
||||
self.assertIsNone(result)
|
||||
|
||||
@patch("builtins.input", side_effect=["invalid", "0"])
|
||||
@patch("src.services.logger.warning")
|
||||
def test_invalid_input(self, mock_warning, mock_input):
|
||||
"""测试无效输入"""
|
||||
result = get_operation()
|
||||
self.assertEqual(result, 0)
|
||||
mock_warning.assert_called()
|
||||
|
||||
@patch("src.services.get_operation", return_value=0)
|
||||
@patch("src.services.SystemServiceStrategy.generate_command", return_value=["sudo", "systemctl", "stop", "nginx"])
|
||||
@patch("src.services.SystemServiceStrategy.execute")
|
||||
@patch("src.services.logger.info")
|
||||
@patch("src.services.SystemServiceStrategy.execute")
|
||||
@patch("src.services.SystemServiceStrategy.generate_command")
|
||||
@patch("src.services.get_operation", return_value=0)
|
||||
def test_service_operation(self, mock_get_operation, mock_generate, mock_execute, mock_info):
|
||||
"""测试服务操作流程"""
|
||||
service = Service(tag="sys", name="nginx")
|
||||
service.service_operation()
|
||||
|
||||
# 修复：generate_command 需要 self 参数
|
||||
mock_generate.assert_called_with(0, "nginx", None)
|
||||
mock_execute.assert_called_with(["sudo", "systemctl", "stop", "nginx"])
|
||||
mock_info.assert_any_call("Executing: sudo systemctl stop nginx")
|
||||
mock_info.assert_any_call("Service nginx operation completed")
|
||||
mock_info.assert_any_call("Executing: sudo systemctl stop nginx")
|
||||
mock_info.assert_any_call("Service nginx operation completed")
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user