[Fix] Fix remove index error

.gitignore

@@ -1,3 +1,4 @@
 .venv/
 dist/
 WebServicesManager.egg-info/
+__pycache__/

README.md

@@ -0,0 +1,54 @@
+# WebServicesManager
+
+**A script set for web services management.**
+
+## Dependencies
+
+- [astral uv: Python package manager](https://docs.astral.sh/uv/)
+
+## Usage
+
+- First clone the repository
+  ```bash
+  # the remote repository where located on my cloud server
+  git clone https://gitea.virtualguard101.com/virtualguard101/webscripts.git
+  cd webscripts
+  ```
+
+- Initialize Python virtual environment and install dependencies
+  ```bash
+  uv venv --python=3.12
+  uv pip install -r requirement.txt
+  ```
+
+- Launch the manager with `main.py`
+
+  >[!IMPORTANT]
+  >If you have some reasons have to give up using uv, you should create virtual environment and install the dependencies manually, otherwise, it is recommended to use uv to manage those because of its convenience.
+
+  ```bash
+  uv run main.py
+  ```
+
+  Add `-h` as parameter to find usage:
+  ```bash
+  uv run main.py -h
+  usage: main.py [-h] {register,list,operate} ...
+
+  =====> Web Services Manager <=====
+
+  positional arguments:
+    {register,list,operate}
+      register            Register a new service
+      list                List all services
+      operate             Service operations to carry out
+
+  options:
+    -h, --help            show this help message and exit
+  ```
+
+## TODO
+
+- [x] Add the function that can remove services
+
+- [ ] Package src as a module

data/services.json

@@ -0,0 +1 @@
+[]

main.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+
+import argparse
+import json
+import os
+from src.manager import ServiceFactory, ServiceRepository, Manager
+from src.services import DockerServiceStrategy, SystemServiceStrategy
+
+def main():
+    parser = argparse.ArgumentParser(description="=====> Web Services Manager <=====")
+    subparsers = parser.add_subparsers(dest="command", required=True)
+    
+    # 注册服务命令
+    register_parser = subparsers.add_parser("register", help="Register a new service")
+    register_parser.add_argument("tag", choices=["sys", "docker"], help="Tag of the service")
+    register_parser.add_argument("name", help="Name of the service")
+    register_parser.add_argument("--path", help="Config/Data path of the service where the 'docker-compose.yml' located (docker-based services only)")
+    
+    # 列出服务命令
+    subparsers.add_parser("list", help="List all services")
+    
+    # 执行操作命令
+    operate_parser = subparsers.add_parser("operate", help="Service operations to carry out")
+    operate_parser.add_argument("index", type=int, help="Index of the service, which is a integer")
+    
+    # 移除服务命令
+    remove_parser = subparsers.add_parser("remove", help="Remove a service")
+    remove_parser.add_argument("index", type=int, help="Index of the service to remove")
+    
+    args = parser.parse_args()
+    
+    # 初始化仓库和管理器
+    # 确保 data 目录存在
+    os.makedirs("data", exist_ok=True)
+    repo = ServiceRepository("data/services.json")
+    manager = Manager(repo)
+    
+    if args.command == "register":
+        # 创建服务
+        service = manager.register_service(args.tag, args.name, args.path)
+        print(f"Register a new service successfully: {service.name}")
+        
+    elif args.command == "list":
+        # 列出所有服务
+        services = manager.list_services()
+        for i, service in enumerate(services):
+            print(f"{i}: [{service.tag}] {service.name} {f'(path: {service.path})' if service.path else ''}")
+            
+    elif args.command == "operate":
+        # 执行服务操作
+        try:
+            manager.execute_service_operation(args.index)
+            print("Operation success!")
+        except IndexError:
+            print("Error: invalid index")
+        except Exception as e:
+            print(f"Operation failed: {str(e)}")
+            
+    elif args.command == "remove":
+        try:
+            manager.remove_service(args.index)
+            print(f"Service at index {args.index} removed successfully")
+        except IndexError:
+            print("Error: invalid index")
+        except Exception as e:
+            print(f"Remove failed: {str(e)}")
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 authors = [
   { name="virtualguard101", email="virtualguard101@gmail.com"}
 ]
-description = "A template for web services management."
+description = "A script set for web services management."
 readme = "README.md"
 requires-python = ">=3.12"
 classifiers = [
@@ -18,6 +18,6 @@ dependencies = [
     "colorlog==6.9.0",
 ]
 
-[project.urls]
-Homepage = "https://github.com/virtualguard101/WebServicesManager"
-Issues = "https://github.com/virtualguard101/WebServicesManager/issues"
+# [project.urls]
+# Homepage = "https://github.com/virtualguard101/WebServicesManager"
+# Issues = "https://github.com/virtualguard101/WebServicesManager/issues"

src/manager.py

@@ -1,8 +1,11 @@
 #!/usr/bin/env python3
 
-from src.services import *
+from src.services import Service
 import logging
+import json
+import os
 import colorlog
+from typing import Optional, List, Dict
 
 # Initialize color logging
 handler = colorlog.StreamHandler()
@@ -22,62 +25,246 @@ logger.setLevel(logging.INFO)
 logger.addHandler(handler)
 
 
-# nginx = Service(tag="sys", name="nginx")
-# minecraft = Service(tag="sys", name="minecraft", path="~/web/minecraftService")
-# homepage = Service(tag="docker", name="docker", path="~/web/homepageService")
-# status = Service(tag="docker", name="status", path="~/web/statusService")
-# gitea = Service(tag="docker", name="gitea", path="~/web/giteaService")
-
-class Manager:
-    """Interface set of services management operations.
-
-    Attributes:
-        service_list (list): List use for storing service instances.
+class ServiceFactory:
+    """Factory for creating Service instances based on tag.
+    
     """
-
-    def __init__(self):
-       """Initialize the Manager with an empty services list."""
-       self.services_list = []
-
-    def append_service(self, service_instance: Service) -> None:
-        """Append a service instance to the services list.
+    
+    @staticmethod
+    def create_service(tag: str, name: str, path: Optional[str] = None) -> Service:
+        """Create a Service instance with validation.
         
         Args:
-            service_instance (Service): The service instance to append.
-        """
-        self.services_list.append(service_instance)
-
-    def register_service(self, service_tag: str, service_name: str, service_path: str) -> Service:
-        """Register a new service.
-
-        Args:
-            service_tag ('sys' | 'docker'): The tag that marks the service instance to use which way to deploy ('sys' or 'docker').
-            service_name (str): The name of the service instance.
-            service_path (str): The configuration and data path of the service instance.
-
+            tag: Service type ('sys' or 'docker')
+            name: Service name
+            path: Configuration path (required for docker services)
+            
         Returns:
-            service: A web service instance.
+            Service instance
+            
+        Raises:
+            ValueError: For invalid inputs
         """
-        if service_tag == "docker":
-            if not os.path.exists(service_path):
-                raise ValueError(f"Invalid service path: {service_path}")
-            service = Service(tag=service_tag, name=service_name, path=service_path)
-            # service.code = len(self.services_list) + 1
-            self.append_service(service_instance=service)
+        if tag == "docker":
+            if not path:
+                raise ValueError("Path is required for docker services")
+            if not os.path.exists(path):
+                raise ValueError(f"Invalid service path: {path}")
+            if not os.path.isdir(path):
+                raise ValueError(f"Service path must be a directory: {path}")
+            logger.info(f"Validated docker service path: {path}")
+        return Service(tag=tag, name=name, path=path)
+
+
+class ServiceRepository:
+    """Repository for service persistence.
+
+    """
+    
+    def __init__(self, file_path: str = 'data/services.json'):
+        self.file_path = file_path
+        # Make sure the path has existed
+        dir_path = os.path.dirname(file_path)
+        if dir_path:
+            os.makedirs(dir_path, exist_ok=True)
+        
+    def save(self, service: Service) -> None:
+        """Save a service to the repository.
+        
+        Args:
+            service: Service instance to save
+        """
+        # Load existing services
+        services = self.load_all()
+        
+        # Add new service
+        services.append(service.to_dict())
+        
+        # Save back to file
+        with open(self.file_path, 'w') as file:
+            json.dump(services, file, indent=4)
+        logger.info(f"Saved service to repository: {service.name}")
+            
+    def load_all(self) -> List[Dict]:
+        """Load all services from repository.
+        
+        Returns:
+            List of service dictionaries
+        """
+        if os.path.exists(self.file_path):
+            try:
+                with open(self.file_path, 'r') as file:
+                    return json.load(file)
+            except (json.JSONDecodeError, FileNotFoundError):
+                logger.warning("Service repository file corrupted or missing")
+                return []
+        logger.info("Service repository file not found, starting fresh")
+        return []
+        
+    def remove(self, service_name: str) -> None:
+        """Remove a service by name.
+        
+        Args:
+            service_name: Name of the service to remove
+            
+        Raises:
+            ValueError: If service not found
+        """
+        try:
+            # Load all servives
+            services = self.load_all()
+            
+            # Search for matching services (case insensitive)
+            original_count = len(services)
+            services = [s for s in services
+                       if s['name'].strip().lower() != service_name.strip().lower()]
+            
+            # Check and remove
+            if len(services) == original_count:
+                raise ValueError(f"Service '{service_name}' not found")
+            
+            # Update services,json
+            with open(self.file_path, 'w') as file:
+                json.dump(services, file, indent=4)
+                
+            logger.info(f"Removed service: {service_name}")
+            
+        except FileNotFoundError:
+            logger.warning("Services file not found, nothing to remove")
+            raise ValueError("Services file does not exist")
+        except json.JSONDecodeError:
+            logger.error("Invalid JSON format in services file")
+            raise ValueError("Invalid services data format")
+        except PermissionError:
+            logger.error("Permission denied when writing services file")
+            raise
+        except Exception as e:
+            logger.error(f"Error removing service: {str(e)}")
+            raise
+
+
+class Manager:
+    """Interface for services management operations.
+    
+    """
+    
+    def __init__(self,
+                 repository: Optional[ServiceRepository] = None,
+                 factory: Optional[ServiceFactory] = None):
+        """Initialize Manager with dependencies.
+        
+        Args:
+            repository: Service repository instance (optional)
+            factory: Service factory instance (optional)
+        """
+        self.repository = repository or ServiceRepository()
+        self.factory = factory or ServiceFactory()
+        logger.info("Manager initialized")
+
+    def register_service(self,
+                         service_tag: str,
+                         service_name: str,
+                         service_path: Optional[str] = None) -> Service:
+        """Register and persist a new service.
+        
+        Args:
+            service_tag: Service type ('sys' or 'docker')
+            service_name: Service name
+            service_path: Configuration path (required for docker)
+            
+        Returns:
+            Registered service instance
+            
+        Raises:
+            ValueError: If service registration fails
+        """
+        try:
+            # Create service via factory
+            service = self.factory.create_service(service_tag, service_name, service_path)
+            
+            # Persist service
+            self.repository.save(service)
+            
+            logger.info(f"Successfully registered service: {service_name} ({service_tag})")
             return service
-        service = Service(tag=service_tag, name=service_name)
-        # service.code = len(self.services_list) + 1
-        self.append_service(service_instance=service)
-        return service
-
-    def list_services(self) -> None:
-        """Method use for count the num of service instances and list the names of them.
+            
+        except ValueError as e:
+            logger.error(f"Service registration failed: {str(e)}")
+            raise
 
+    def remove_service(self, index: int) -> None:
+        """Remove a service by index.
+        
+        Args:
+            index: Index of the service to remove
+            
+        Raises:
+            IndexError: If index is out of bounds
+            ValueError: If service not found
         """
+        services = self.list_services()
+        if index < 0 or index >= len(services):
+            raise IndexError(f"Invalid service index: {index}")
+        service_name = services[index].name
+        self.repository.remove(service_name)
 
-        sum_of_service = 0
-        service_name_list = []
-        for services in self.services_list:
-            service_name_list.append(services.name)
-            sum_of_service += 1
-        logger.info(f"The manager has registered {sum_of_service} services: {', '.join(service_name_list)}")
+    def list_services(self) -> List[Service]:
+        """List all registered services.
+        
+        Returns:
+            List of Service objects
+        """
+        services_data = self.repository.load_all()
+        if not services_data:
+            logger.info("No services registered")
+            return []
+            
+        services = []
+        for s in services_data:
+            try:
+                service = Service(tag=s["tag"], name=s["name"], path=s["path"])
+                services.append(service)
+            except Exception as e:
+                logger.error(f"Invalid service data: {s}, error: {str(e)}")
+                
+        service_names = [s.name for s in services]
+        logger.info(f"Registered services ({len(services)}): {', '.join(service_names)}")
+        return services
+        
+    def execute_service_operation(self, index: int) -> None:
+        """Execute service operation for the service at the given index.
+        
+        Args:
+            index: Index of the service in the list returned by list_services()
+            
+        Raises:
+            IndexError: If index is out of bounds
+        """
+        services = self.list_services()
+        if index < 0 or index >= len(services):
+            logger.error(f"Invalid service index: {index}")
+            raise IndexError(f"Invalid service index: {index}")
+            
+        service = services[index]
+        service.service_operation()
+
+if __name__ == "__main__":
+    # 测试服务移除功能
+    manager = Manager()
+    try:
+        # 注册测试服务
+        test_service = manager.register_service("sys", "test-service")
+        print("Registered test service")
+        
+        # 移除测试服务（索引0）
+        manager.remove_service(0)
+        print("Successfully removed test service")
+        
+        # 尝试移除不存在的服务（索引越界）
+        try:
+            manager.remove_service(100)
+        except IndexError as e:
+            print(f"Expected error: {str(e)}")
+            
+    except Exception as e:
+        print(f"Test failed: {str(e)}")

src/services.py

@@ -4,6 +4,8 @@ import subprocess
 import os
 import logging
 import colorlog
+from abc import ABC, abstractmethod
+from typing import Optional, Tuple, List, Callable
 
 # Initialize color logging
 handler = colorlog.StreamHandler()
@@ -23,11 +25,16 @@ logger.setLevel(logging.INFO)
 logger.addHandler(handler)
 
 
-def get_operation():
+def get_operation() -> Optional[int]:
+    """Get service operation from user input.
+    
+    Returns:
+        0: stop, 1: restart, None: quit
+    """
     while True:
         choice = input("Please select an operation (0 to stop, 1 to restart, q to quit): ").strip().lower()
         if choice == 'q':
-            return 'q'
+            return None
         try:
             op = int(choice)
             if op in (0, 1):
@@ -37,80 +44,152 @@ def get_operation():
         logger.warning("Invalid input, please enter 0, 1, or q.")
 
 
-class Service:
-    """A template management for web services
+class ServiceStrategy(ABC):
+    """Abstract base class for service operation strategies."""
+    
+    @abstractmethod
+    def generate_command(self, operation: int, service_name: str, path: Optional[str] = None) -> List[str]:
+        """Generate command for service operation.
+        
+        Args:
+            operation: 0=stop, 1=restart
+            service_name: Name of the service
+            path: Service path (for docker)
+            
+        Returns:
+            Command list for subprocess
+        """
+        pass
+    
+    @abstractmethod
+    def execute(self, command: List[str]) -> None:
+        """Execute the service command.
+        
+        Args:
+            command: Command to execute
+            
+        Raises:
+            subprocess.CalledProcessError: If command fails
+        """
+        pass
 
+
+class SystemServiceStrategy(ServiceStrategy):
+    """Strategy for systemd services."""
+    
+    def __init__(self, path: Optional[str] = None):
+        pass
+    
+    def generate_command(self, operation: int, service_name: str, path: Optional[str] = None) -> List[str]:
+        if operation == 0:
+            return ["sudo", "systemctl", "stop", service_name]
+        elif operation == 1:
+            return ["sudo", "systemctl", "restart", service_name]
+        raise ValueError(f"Invalid operation for system service: {operation}")
+    
+    def execute(self, command: List[str]) -> None:
+        subprocess.run(command, check=True)
+
+
+class DockerServiceStrategy(ServiceStrategy):
+    """Strategy for docker-compose services."""
+    
+    def __init__(self, path: str):
+        self.path = path
+    
+    def generate_command(self, operation: int, service_name: str, path: Optional[str] = None) -> List[str]:
+        if not self.path:
+            raise ValueError("Docker service requires a path")
+        
+        expanded_path = os.path.expanduser(self.path)
+        if not os.path.exists(expanded_path):
+            raise FileNotFoundError(f"Docker path not found: {expanded_path}")
+        if not os.path.isdir(expanded_path):
+            raise NotADirectoryError(f"Docker path must be a directory: {expanded_path}")
+        
+        if operation == 0:
+            return ["docker", "compose", "down"]
+        elif operation == 1:
+            return ["docker", "compose", "up", "-d"]
+        raise ValueError(f"Invalid operation for docker service: {operation}")
+    
+    def execute(self, command: List[str]) -> None:
+        expanded_path = os.path.expanduser(self.path)
+        if not os.path.exists(expanded_path):
+            raise FileNotFoundError(f"Path not found: {expanded_path}")
+        if not os.path.isdir(expanded_path):
+            raise NotADirectoryError(f"Docker path must be directory: {expanded_path}")
+        subprocess.run(command, check=True, cwd=expanded_path)
+
+
+class Service:
+    """Management interface for web services.
+    
     Attributes:
-        tag ('sys' | 'docker'): The tag that marks the service instance to use which way to deploy.
-        name (str): The name of the service instance.
-        path (str): The configuration and data path of the service instance.
+        tag: Service type ('sys' or 'docker')
+        name: Service name
+        path: Configuration path (for docker services)
+        strategy: Service operation strategy
     """
 
-    def __init__(self, tag: str, name: str, path=None):
-        self._tag = tag
-        self._name = name
-        self._path = path
-
-    def command_gen(self):
-        """Generate service management commands based on the service's tag.
+    STRATEGIES = {
+        'sys': SystemServiceStrategy,
+        'docker': DockerServiceStrategy
+    }
 
+    def __init__(self, tag: str, name: str, path: Optional[str] = None):
+        """Initialize a service instance.
+        
+        Args:
+            tag: Service type ('sys' or 'docker')
+            name: Service name
+            path: Configuration path (required for docker)
+            
+        Raises:
+            ValueError: For invalid tag
         """
+        if tag not in self.STRATEGIES:
+            raise ValueError(f"Invalid service tag: {tag}")
+        
+        self.tag = tag
+        self.name = name
+        self.path = path
+        self.strategy = self.STRATEGIES[tag](path=self.path)
+        logger.info(f"Initialized {tag} service: {name}")
+        
+    def to_dict(self) -> dict:
+        """Return dictionary representation of the service."""
+        return {
+            "tag": self.tag,
+            "name": self.name,
+            "path": self.path
+        }
 
-        system_services_command = "sudo systemctl"
-        docker_services_command = f"cd {os.path.expanduser(f'{self._path}')} && docker compose"
-        if self._tag == "sys":
-            return system_services_command
-        elif self._tag == "docker":
-            return docker_services_command
-        else:
-            raise ValueError(f"The service tag {self._tag} was not included.")
-
-    def service_operation(self):
-        """Manage the service based on user input.
-
+    def service_operation(self) -> None:
+        """Perform service operation based on user input.
+        
         """
-
         operation = get_operation()
-        if operation == 'q':
-            logger.info("User chose to quit the service management.")
+        if operation is None:
+            logger.info("Operation cancelled by user")
             return
         
-        command = self.command_gen()
-        full_command = None  # Initialize full_command
-        
-        if self._tag == "sys":
-            if operation == 0:
-                # Stop the service
-                full_command = f"{command} stop {self._name}"
-                logger.info(f"Stopping service: {self._name}")
-            elif operation == 1:
-                # Restart the service
-                full_command = f"{command} restart {self._name}"
-                logger.info(f"Restarting service: {self._name}")
+        try:
+            # Generate and execute command
+            if self.path is not None:
+                command = self.strategy.generate_command(operation, self.name, self.path)
             else:
-                logger.warning("Invalid operation; no service management executed.")
-                return  # Exit if the operation is invalid
-        if self._tag == "docker":
-            if operation == 0:
-                # Stop the service
-                full_command = f"{command} down"
-                logger.info(f"Stopping service: {self._name}")
-            elif operation == 1:
-                # Restart the service
-                full_command = f"{command} up -d"
-                logger.info(f"Restarting service: {self._name}")
-            else:
-                logger.warning("Invalid operation; no service management executed.")
-                return  # Exit if the operation is invalid
-        
-        if full_command:  # Ensure full_command is defined
-            try:
-                subprocess.run(full_command, check=True, shell=True)
-                logger.info(f"Service {self._name} operation completed successfully.")
-            except subprocess.CalledProcessError as e:
-                logger.error(f"Failed to manage service {self._name}: {e}")
+                command = self.strategy.generate_command(operation, self.name)
+            logger.info(f"Executing: {' '.join(command)}")
+            self.strategy.execute(command)
+            logger.info(f"Service {self.name} operation completed")
+        except (ValueError, FileNotFoundError, NotADirectoryError, subprocess.CalledProcessError) as e:
+            logger.error(f"Service operation failed: {str(e)}")
 
-# Example call
+# Example usage
 if __name__ == "__main__":
-    service = Service(tag="docker", name="homepage", path="~/web/homepageService")
-    service.service_operation()
+    try:
+        service = Service(tag="docker", name="homepage", path="~/projects/webservices/homepage")
+        service.service_operation()
+    except Exception as e:
+        logger.error(f"Unhandled exception: {str(e)}")

test.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+set -ue
+
+source .venv/bin/activate && python -m unittest discover -s test

test/test_manager.py

@@ -0,0 +1,303 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from src.manager import ServiceFactory, ServiceRepository, Manager
+import os
+import json
+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()
+        # 在临时目录下创建 data 子目录
+        data_dir = os.path.join(self.temp_dir.name, "data")
+        os.makedirs(data_dir, exist_ok=True)
+        self.repo_path = os.path.join(data_dir, "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)
+        
+    def test_remove_service_success(self):
+        """测试正常移除服务"""
+        # 添加测试服务
+        mock_service = MagicMock()
+        mock_service.to_dict.return_value = {
+            "tag": "sys",
+            "name": "nginx",
+            "path": None
+        }
+        self.repo.save(mock_service)
+        
+        # 移除服务
+        self.repo.remove("nginx")
+        
+        # 验证服务已被移除
+        services = self.repo.load_all()
+        self.assertEqual(len(services), 0)
+        
+    def test_remove_service_case_insensitive(self):
+        """测试大小写不敏感移除"""
+        # 添加测试服务
+        mock_service = MagicMock()
+        mock_service.to_dict.return_value = {
+            "tag": "sys",
+            "name": "Nginx",
+            "path": None
+        }
+        self.repo.save(mock_service)
+        
+        # 使用小写名称移除
+        self.repo.remove("nginx")
+        
+        # 验证服务已被移除
+        services = self.repo.load_all()
+        self.assertEqual(len(services), 0)
+        
+    def test_remove_service_whitespace(self):
+        """测试移除带空格的服务名"""
+        # 添加测试服务
+        mock_service = MagicMock()
+        mock_service.to_dict.return_value = {
+            "tag": "sys",
+            "name": " nginx ",
+            "path": None
+        }
+        self.repo.save(mock_service)
+        
+        # 使用带空格名称移除
+        self.repo.remove("  nginx  ")
+        
+        # 验证服务已被移除
+        services = self.repo.load_all()
+        self.assertEqual(len(services), 0)
+        
+    def test_remove_nonexistent_service(self):
+        """测试移除不存在的服务"""
+        with self.assertRaises(ValueError):
+            self.repo.remove("nonexistent")
+
+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.ServiceRepository")
+    @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.ServiceRepository")
+    def test_remove_service_success(self, mock_repo):
+        """测试正常移除服务"""
+        manager = Manager(mock_repo.return_value)
+        # 模拟服务列表
+        mock_service = MagicMock()
+        mock_service.name = "nginx"
+        with patch.object(manager, 'list_services', return_value=[mock_service]):
+            manager.remove_service(0)
+        
+        # 验证ServiceRepository的remove方法被正确调用
+        mock_repo.return_value.remove.assert_called_once_with("nginx")
+        
+    @patch("src.manager.ServiceRepository")
+    def test_remove_service_case_insensitive(self, mock_repo):
+        """测试大小写不敏感移除"""
+        manager = Manager(mock_repo.return_value)
+        # 模拟服务列表
+        mock_service = MagicMock()
+        mock_service.name = "Nginx"  # 服务名称是大写开头
+        with patch.object(manager, 'list_services', return_value=[mock_service]):
+            # 使用索引0移除
+            manager.remove_service(0)
+        
+        # 验证调用时使用原始大小写
+        mock_repo.return_value.remove.assert_called_once_with("Nginx")
+        
+    @patch("src.manager.ServiceRepository")
+    def test_remove_service_whitespace(self, mock_repo):
+        """测试移除带空格的服务名"""
+        manager = Manager(mock_repo.return_value)
+        # 模拟服务列表
+        mock_service = MagicMock()
+        mock_service.name = " nginx "  # 服务名带空格
+        with patch.object(manager, 'list_services', return_value=[mock_service]):
+            manager.remove_service(0)
+        
+        # 验证调用时使用原始名称（带空格）
+        mock_repo.return_value.remove.assert_called_once_with(" nginx ")
+        
+    @patch("src.manager.ServiceRepository")
+    def test_remove_service_nonexistent(self, mock_repo):
+        """测试索引越界"""
+        manager = Manager(mock_repo.return_value)
+        # 模拟空服务列表
+        with patch.object(manager, 'list_services', return_value=[]):
+            with self.assertRaises(IndexError):
+                manager.remove_service(0)
+            
+    @patch("src.manager.ServiceRepository")
+    def test_remove_service_file_not_found(self, mock_repo):
+        """测试文件不存在时的错误处理"""
+        manager = Manager(mock_repo.return_value)
+        # 模拟服务列表
+        mock_service = MagicMock()
+        mock_service.name = "nginx"
+        # 设置remove方法抛出FileNotFoundError
+        mock_repo.return_value.remove.side_effect = FileNotFoundError
+        
+        with patch.object(manager, 'list_services', return_value=[mock_service]):
+            with self.assertRaises(FileNotFoundError):
+                manager.remove_service(0)
+            
+    @patch("src.manager.ServiceRepository")
+    def test_remove_service_invalid_json(self, mock_repo):
+        """测试无效JSON文件时的异常处理"""
+        manager = Manager(mock_repo.return_value)
+        # 模拟服务列表
+        mock_service = MagicMock()
+        mock_service.name = "nginx"
+        # 设置remove方法抛出JSONDecodeError
+        mock_repo.return_value.remove.side_effect = json.JSONDecodeError("Expecting value", "", 0)
+        
+        with patch.object(manager, 'list_services', return_value=[mock_service]):
+            with self.assertRaises(json.JSONDecodeError):
+                manager.remove_service(0)
+                
+    @patch("src.manager.ServiceRepository")
+    def test_remove_service_invalid_index(self, mock_repo):
+        """测试无效索引"""
+        manager = Manager(mock_repo.return_value)
+        # 模拟服务列表
+        mock_service = MagicMock()
+        mock_service.name = "nginx"
+        
+        # 测试负索引
+        with patch.object(manager, 'list_services', return_value=[mock_service]):
+            with self.assertRaises(IndexError):
+                manager.remove_service(-1)
+                
+        # 测试超出范围的索引
+        with patch.object(manager, 'list_services', return_value=[mock_service]):
+            with self.assertRaises(IndexError):
+                manager.remove_service(1)
+    @patch("src.manager.ServiceRepository")
+    @patch("src.manager.logger")
+    def test_execute_service_operation_invalid_index(self, mock_logger, mock_repo):
+        """测试无效服务索引"""
+        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_with("Invalid service index: 0")
+            
+
+if __name__ == "__main__":
+    unittest.main()

test/test_services.py

@@ -0,0 +1,162 @@
+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.SystemServiceStrategy.execute")
+    @patch("src.services.SystemServiceStrategy.generate_command", return_value=["sudo", "systemctl", "stop", "nginx"])
+    @patch("src.services.get_operation", return_value=0)
+    @patch("src.services.logger.info")
+    def test_service_operation(self, mock_info, mock_get_operation, mock_generate, mock_execute):
+        """测试服务操作流程"""
+        service = Service(tag="sys", name="nginx")
+        service.service_operation()
+        
+        # 验证调用
+        mock_get_operation.assert_called_once()
+        mock_generate.assert_called_with(0, "nginx")
+        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")
+
+    @patch("src.services.DockerServiceStrategy.execute")
+    @patch("src.services.DockerServiceStrategy.generate_command", return_value=["docker", "compose", "down"])
+    @patch("src.services.get_operation", return_value=0)
+    @patch("src.services.logger.info")
+    def test_docker_service_operation(self, mock_info, mock_get_operation, mock_generate, mock_execute):
+        """测试Docker服务操作流程"""
+        service = Service(tag="docker", name="homepage", path="/path/to/docker")
+        service.service_operation()
+        
+        # 验证调用
+        mock_get_operation.assert_called_once()
+        mock_generate.assert_called_with(0, "homepage", "/path/to/docker")
+        mock_execute.assert_called_with(["docker", "compose", "down"])
+        mock_info.assert_any_call("Executing: docker compose down")
+        mock_info.assert_any_call("Service homepage operation completed")
+
+if __name__ == "__main__":
+    unittest.main()