Maas-Agent/apps/plugin/services/plugin_validator.py

# coding=utf-8
"""
插件验证服务
提供插件 Schema 验证、配置校验、测试执行等功能
"""
import time
from typing import Optional, Dict, Any
from django.db.models import QuerySet
from plugin.models.plugin import Plugin, PluginStatus
from plugin.models.plugin_test import PluginTest, PluginTestStatus
from plugin.models.plugin_version import PluginVersion
from plugin.services.error_codes import PluginErrorCode
from plugin.services.exception_handler import (
    PluginNotFoundException, PluginTestException, PluginTestTimeoutException,
    PluginSchemaException
)
from common.utils.logger import maxkb_logger


class PluginValidator:
    """插件验证器"""

    # 默认测试超时时间（毫秒）
    DEFAULT_TIMEOUT_MS = 30000

    @staticmethod
    def validate_schema(schema: dict) -> tuple:
        """
        验证插件 Schema 格式
        返回 (is_valid: bool, errors: list)
        """
        errors = []

        if not isinstance(schema, dict):
            errors.append('Schema 必须是字典类型')
            return False, errors

        # 检查必要的字段
        required_fields = ['name', 'description', 'parameters']
        for field in required_fields:
            if field not in schema:
                errors.append(f'Schema 缺少必要字段: {field}')

        # 检查 parameters 格式
        if 'parameters' in schema:
            params = schema['parameters']
            if not isinstance(params, dict):
                errors.append('parameters 必须是字典类型')
            elif 'properties' not in params:
                errors.append('parameters 缺少 properties 定义')

        return len(errors) == 0, errors

    @staticmethod
    def validate_config(config: dict, schema: dict) -> tuple:
        """
        验证插件配置是否符合 Schema
        返回 (is_valid: bool, errors: list)
        """
        errors = []

        if not isinstance(config, dict):
            errors.append('配置必须是字典类型')
            return False, errors

        if not isinstance(schema, dict) or 'parameters' not in schema:
            return True, errors

        params_schema = schema.get('parameters', {})
        properties = params_schema.get('properties', {})
        required = params_schema.get('required', [])

        # 检查必填字段
        for field in required:
            if field not in config:
                errors.append(f'缺少必填配置项: {field}')

        # 检查字段类型（简化检查）
        for field, field_schema in properties.items():
            if field in config:
                expected_type = field_schema.get('type')
                value = config[field]
                if expected_type == 'string' and not isinstance(value, str):
                    errors.append(f'配置项 {field} 应为字符串类型')
                elif expected_type == 'number' and not isinstance(value, (int, float)):
                    errors.append(f'配置项 {field} 应为数字类型')
                elif expected_type == 'boolean' and not isinstance(value, bool):
                    errors.append(f'配置项 {field} 应为布尔类型')

        return len(errors) == 0, errors

    @staticmethod
    def execute_test(plugin_id: str, test_input: dict = None,
                     timeout_ms: int = None) -> PluginTest:
        """
        执行插件测试
        """
        try:
            plugin = Plugin.objects.get(id=plugin_id)
        except Plugin.DoesNotExist:
            raise PluginNotFoundException(plugin_id)

        timeout_ms = timeout_ms or PluginValidator.DEFAULT_TIMEOUT_MS

        # 创建测试记录
        test_record = PluginTest.objects.create(
            plugin=plugin,
            version=plugin.version,
            test_type='smoke',
            status=PluginTestStatus.RUNNING,
            input_data=test_input or {}
        )

        start_time = time.time()

        try:
            # 执行测试逻辑（简化版：验证 Schema 和配置）
            is_schema_valid, schema_errors = PluginValidator.validate_schema(plugin.schema)
            if not is_schema_valid:
                raise PluginSchemaException(
                    message='Schema 验证失败',
                    detail={'errors': schema_errors}
                )

            is_config_valid, config_errors = PluginValidator.validate_config(
                plugin.config, plugin.schema
            )
            if not is_config_valid:
                raise PluginTestException(
                    message='配置验证失败',
                    detail={'errors': config_errors}
                )

            # 计算耗时
            duration_ms = int((time.time() - start_time) * 1000)

            # 更新测试记录
            test_record.status = PluginTestStatus.SUCCESS
            test_record.passed = True
            test_record.duration_ms = duration_ms
            test_record.output_data = {
                'schema_valid': True,
                'config_valid': True,
                'message': '插件验证通过'
            }
            test_record.save()

            return test_record

        except (PluginSchemaException, PluginTestException) as e:
            duration_ms = int((time.time() - start_time) * 1000)
            test_record.status = PluginTestStatus.FAILED
            test_record.passed = False
            test_record.duration_ms = duration_ms
            test_record.error_code = e.error_code
            test_record.error_message = e.message
            test_record.output_data = {'detail': e.detail}
            test_record.save()
            return test_record

        except Exception as e:
            duration_ms = int((time.time() - start_time) * 1000)
            maxkb_logger.error(f"Plugin test execution error: {e}", exc_info=True)
            test_record.status = PluginTestStatus.ERROR
            test_record.passed = False
            test_record.duration_ms = duration_ms
            test_record.error_code = 'PLUGIN_999'
            test_record.error_message = str(e)
            test_record.save()
            return test_record

    @staticmethod
    def get_test_history(plugin_id: str, limit: int = 20) -> QuerySet:
        """获取插件测试历史"""
        return PluginTest.objects.filter(
            plugin_id=plugin_id
        ).order_by('-create_time')[:limit]

    @staticmethod
    def get_test_detail(test_id: str) -> Optional[PluginTest]:
        """获取测试详情"""
        try:
            return PluginTest.objects.get(id=test_id)
        except PluginTest.DoesNotExist:
            return None