Maas-Agent/apps/plugin/views/plugin_test_view.py

# coding=utf-8
"""
插件测试和版本管理视图
提供插件测试执行、结果查询、版本历史等 API
"""
from django.utils.translation import gettext as _
from rest_framework.request import Request
from rest_framework.views import APIView

from common.auth import TokenAuth
from common.exception.app_exception import AppApiException
from common.result import result
from plugin.services.plugin_validator import PluginValidator
from plugin.services.exception_handler import PluginExceptionHandler


class PluginTestView(APIView):
    """插件测试管理"""
    authentication_classes = [TokenAuth]

    class Execute(APIView):
        """执行插件测试"""
        authentication_classes = [TokenAuth]

        def post(self, request: Request, plugin_id: str):
            test_input = request.data.get('input', {})
            timeout_ms = request.data.get('timeout_ms')

            try:
                test_record = PluginValidator.execute_test(
                    plugin_id=plugin_id,
                    test_input=test_input,
                    timeout_ms=timeout_ms
                )
            except Exception as e:
                error_resp = PluginExceptionHandler.to_api_response(e)
                raise AppApiException(error_resp['code'], error_resp['message'])

            return result.success({
                'id': str(test_record.id),
                'status': test_record.status,
                'passed': test_record.passed,
                'error_code': test_record.error_code,
                'error_message': test_record.error_message,
                'duration_ms': test_record.duration_ms,
                'input_data': test_record.input_data,
                'output_data': test_record.output_data,
                'create_time': test_record.create_time.isoformat() if test_record.create_time else None,
            })

    class History(APIView):
        """测试历史"""
        authentication_classes = [TokenAuth]

        def get(self, request: Request, plugin_id: str):
            limit = int(request.query_params.get('limit', 20))
            tests = PluginValidator.get_test_history(plugin_id, limit)

            data = [{
                'id': str(t.id),
                'version': t.version,
                'test_type': t.test_type,
                'status': t.status,
                'passed': t.passed,
                'error_code': t.error_code,
                'error_message': t.error_message,
                'duration_ms': t.duration_ms,
                'create_time': t.create_time.isoformat() if t.create_time else None,
            } for t in tests]

            return result.success(data)

    class Detail(APIView):
        """测试详情"""
        authentication_classes = [TokenAuth]

        def get(self, request: Request, plugin_id: str, test_id: str):
            test = PluginValidator.get_test_detail(test_id)
            if not test:
                raise AppApiException(404, _('测试记录不存在'))

            return result.success({
                'id': str(test.id),
                'plugin_id': str(test.plugin_id),
                'version': test.version,
                'test_type': test.test_type,
                'status': test.status,
                'passed': test.passed,
                'error_code': test.error_code,
                'error_message': test.error_message,
                'duration_ms': test.duration_ms,
                'input_data': test.input_data,
                'output_data': test.output_data,
                'create_time': test.create_time.isoformat() if test.create_time else None,
            })

    class ValidateSchema(APIView):
        """验证插件 Schema"""
        authentication_classes = [TokenAuth]

        def post(self, request: Request):
            schema = request.data.get('schema', {})
            is_valid, errors = PluginValidator.validate_schema(schema)

            return result.success({
                'valid': is_valid,
                'errors': errors
            })

    class ValidateConfig(APIView):
        """验证插件配置"""
        authentication_classes = [TokenAuth]

        def post(self, request: Request):
            config = request.data.get('config', {})
            schema = request.data.get('schema', {})

            is_valid, errors = PluginValidator.validate_config(config, schema)

            return result.success({
                'valid': is_valid,
                'errors': errors
            })