update: redis配置适配密码格式

create：加入自动化测试管理模块

backend/app/api/v1/services/autotest/api_case_service.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+
+from typing import List, Dict
+
+from app.api.v1.schemas.system.auth_schema import AuthSchema
+from app.api.v1.schemas.autotest.api_case_schema import APICaseOutSchema, APICaseCreateSchema, APICaseUpdateSchema
+from app.api.v1.params.autotest.api_case_param import APICaseQueryParams
+from app.api.v1.cruds.autotest.api_case_crud import APICaseCRUD
+
+
+class APICaseService:
+    """
+    接口用例服务层
+    """
+    
+    @classmethod
+    async def get_detail_services(cls, auth: AuthSchema, id: int) -> Dict:
+        obj = await APICaseCRUD(auth).get_obj_by_id(id=id)
+        return APICaseOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def get_list_services(cls, auth: AuthSchema, search: APICaseQueryParams = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
+        obj_list = await APICaseCRUD(auth).get_obj_list(search=search.__dict__, order_by=order_by)
+        return [APICaseOutSchema.model_validate(obj).model_dump() for obj in obj_list]
+    
+    @classmethod
+    async def create_services(cls, auth: AuthSchema, data: APICaseCreateSchema) -> Dict:
+        obj = await APICaseCRUD(auth).create_obj(data=data)
+        return APICaseOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def update_services(cls, auth: AuthSchema, data: APICaseUpdateSchema) -> Dict:
+        obj = await APICaseCRUD(auth).update_obj(id=data.id, data=data)
+        return APICaseOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def delete_services(cls, auth: AuthSchema, id: int) -> None:
+        await APICaseCRUD(auth).delete_obj(ids=[id])

backend/app/api/v1/services/autotest/environment_service.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+
+from typing import List, Dict
+
+from app.api.v1.schemas.system.auth_schema import AuthSchema
+from app.api.v1.schemas.autotest.environment_schema import EnvironmentOutSchema, EnvironmentCreateSchema, EnvironmentUpdateSchema
+from app.api.v1.params.autotest.environment_param import EnvironmentQueryParams
+from app.api.v1.cruds.autotest.environment_crud import EnvironmentCRUD
+
+
+class EnvironmentService:
+    """
+    环境服务层
+    """
+    
+    @classmethod
+    async def get_detail_services(cls, auth: AuthSchema, id: int) -> Dict:
+        obj = await EnvironmentCRUD(auth).get_obj_by_id(id=id)
+        return EnvironmentOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def get_list_services(cls, auth: AuthSchema, search: EnvironmentQueryParams = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
+        obj_list = await EnvironmentCRUD(auth).get_obj_list(search=search.__dict__, order_by=order_by)
+        return [EnvironmentOutSchema.model_validate(obj).model_dump() for obj in obj_list]
+    
+    @classmethod
+    async def create_services(cls, auth: AuthSchema, data: EnvironmentCreateSchema) -> Dict:
+        obj = await EnvironmentCRUD(auth).create_obj(data=data)
+        return EnvironmentOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def update_services(cls, auth: AuthSchema, data: EnvironmentUpdateSchema) -> Dict:
+        obj = await EnvironmentCRUD(auth).update_obj(id=data.id, data=data)
+        return EnvironmentOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def delete_services(cls, auth: AuthSchema, id: int) -> None:
+        await EnvironmentCRUD(auth).delete_obj(ids=[id])

backend/app/api/v1/services/autotest/global_data_service.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+
+from typing import List, Dict
+
+from app.api.v1.schemas.system.auth_schema import AuthSchema
+from app.api.v1.schemas.autotest.global_data_schema import GlobalDataOutSchema, GlobalDataCreateSchema, GlobalDataUpdateSchema
+from app.api.v1.params.autotest.global_data_param import GlobalDataQueryParams
+from app.api.v1.cruds.autotest.global_data_crud import GlobalDataCRUD
+
+
+class GlobalDataService:
+    """
+    全局参数服务层
+    """
+    
+    @classmethod
+    async def get_detail_services(cls, auth: AuthSchema, id: int) -> Dict:
+        obj = await GlobalDataCRUD(auth).get_obj_by_id(id=id)
+        return GlobalDataOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def get_list_services(cls, auth: AuthSchema, search: GlobalDataQueryParams = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
+        obj_list = await GlobalDataCRUD(auth).get_obj_list(search=search.__dict__, order_by=order_by)
+        return [GlobalDataOutSchema.model_validate(obj).model_dump() for obj in obj_list]
+    
+    @classmethod
+    async def create_services(cls, auth: AuthSchema, data: GlobalDataCreateSchema) -> Dict:
+        obj = await GlobalDataCRUD(auth).create_obj(data=data)
+        return GlobalDataOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def update_services(cls, auth: AuthSchema, data: GlobalDataUpdateSchema) -> Dict:
+        obj = await GlobalDataCRUD(auth).update_obj(id=data.id, data=data)
+        return GlobalDataOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def delete_services(cls, auth: AuthSchema, id: int) -> None:
+        await GlobalDataCRUD(auth).delete_obj(ids=[id])

backend/app/api/v1/services/autotest/module_service.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+
+from typing import List, Dict
+
+from app.api.v1.schemas.system.auth_schema import AuthSchema
+from app.api.v1.schemas.autotest.module_schema import ModuleOutSchema, ModuleCreateSchema, ModuleUpdateSchema
+from app.api.v1.params.autotest.module_param import ModuleQueryParams
+from app.api.v1.cruds.autotest.module_crud import ModuleCRUD
+
+
+class ModuleService:
+    """
+    模块服务层
+    """
+    
+    @classmethod
+    async def get_detail_services(cls, auth: AuthSchema, id: int) -> Dict:
+        obj = await ModuleCRUD(auth).get_obj_by_id(id=id)
+        return ModuleOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def get_list_services(cls, auth: AuthSchema, search: ModuleQueryParams = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
+        obj_list = await ModuleCRUD(auth).get_obj_list(search=search.__dict__, order_by=order_by)
+        return [ModuleOutSchema.model_validate(obj).model_dump() for obj in obj_list]
+    
+    @classmethod
+    async def create_services(cls, auth: AuthSchema, data: ModuleCreateSchema) -> Dict:
+        obj = await ModuleCRUD(auth).create_obj(data=data)
+        return ModuleOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def update_services(cls, auth: AuthSchema, data: ModuleUpdateSchema) -> Dict:
+        obj = await ModuleCRUD(auth).update_obj(id=data.id, data=data)
+        return ModuleOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def delete_services(cls, auth: AuthSchema, id: int) -> None:
+        await ModuleCRUD(auth).delete_obj(ids=[id])

backend/app/api/v1/services/autotest/notification_config_service.py

@@ -0,0 +1,39 @@
+# -*- coding: utf-8 -*-
+
+from typing import List, Dict
+
+from app.api.v1.schemas.system.auth_schema import AuthSchema
+from app.api.v1.schemas.autotest.notification_config_schema import NotificationConfigOutSchema, NotificationConfigCreateSchema, NotificationConfigUpdateSchema
+from app.api.v1.params.autotest.notification_config_param import NotificationConfigQueryParams
+from app.api.v1.cruds.autotest.notification_config_crud import NotificationConfigCRUD
+
+
+class NotificationConfigService:
+    """
+    通知配置服务层
+    """
+    
+    @classmethod
+    async def get_detail_services(cls, auth: AuthSchema, id: int) -> Dict:
+        obj = await NotificationConfigCRUD(auth).get_obj_by_id(id=id)
+        return NotificationConfigOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def get_list_services(cls, auth: AuthSchema, search: NotificationConfigQueryParams = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
+        obj_list = await NotificationConfigCRUD(auth).get_obj_list(search=search.__dict__, order_by=order_by)
+        return [NotificationConfigOutSchema.model_validate(obj).model_dump() for obj in obj_list]
+    
+    @classmethod
+    async def create_services(cls, auth: AuthSchema, data: NotificationConfigCreateSchema) -> Dict:
+        obj = await NotificationConfigCRUD(auth).create_obj(data=data)
+        return NotificationConfigOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def update_services(cls, auth: AuthSchema, data: NotificationConfigUpdateSchema) -> Dict:
+        obj = await NotificationConfigCRUD(auth).update_obj(id=data.id, data=data)
+        return NotificationConfigOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def delete_services(cls, auth: AuthSchema, id: int) -> None:
+        await NotificationConfigCRUD(auth).delete_obj(ids=[id])
+

backend/app/api/v1/services/autotest/project_service.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+
+from typing import List, Dict
+
+from app.api.v1.schemas.system.auth_schema import AuthSchema
+from app.api.v1.schemas.autotest.project_schema import ProjectOutSchema, ProjectCreateSchema, ProjectUpdateSchema
+from app.api.v1.params.autotest.project_param import ProjectQueryParams
+from app.api.v1.cruds.autotest.project_crud import ProjectCRUD
+
+
+class ProjectService:
+    """
+    项目服务层
+    """
+    
+    @classmethod
+    async def get_detail_services(cls, auth: AuthSchema, id: int) -> Dict:
+        obj = await ProjectCRUD(auth).get_obj_by_id(id=id)
+        return ProjectOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def get_list_services(cls, auth: AuthSchema, search: ProjectQueryParams = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
+        obj_list = await ProjectCRUD(auth).get_obj_list(search=search.__dict__, order_by=order_by)
+        return [ProjectOutSchema.model_validate(obj).model_dump() for obj in obj_list]
+    
+    @classmethod
+    async def create_services(cls, auth: AuthSchema, data: ProjectCreateSchema) -> Dict:
+        obj = await ProjectCRUD(auth).create_obj(data=data)
+        return ProjectOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def update_services(cls, auth: AuthSchema, data: ProjectUpdateSchema) -> Dict:
+        obj = await ProjectCRUD(auth).update_obj(id=data.id, data=data)
+        return ProjectOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def delete_services(cls, auth: AuthSchema, id: int) -> None:
+        await ProjectCRUD(auth).delete_obj(ids=[id])

backend/app/api/v1/services/autotest/repont_service.py

@@ -0,0 +1,39 @@
+# -*- coding: utf-8 -*-
+
+from typing import List, Dict
+
+from app.api.v1.schemas.system.auth_schema import AuthSchema
+from app.api.v1.schemas.autotest.repont_schema import ReportOutSchema, ReportCreateSchema, ReportUpdateSchema
+from app.api.v1.params.autotest.repont_param import ReportQueryParams
+from app.api.v1.cruds.autotest.repont_crud import ReportCRUD
+
+
+class ReportService:
+    """
+    报告服务层
+    """
+    
+    @classmethod
+    async def get_detail_services(cls, auth: AuthSchema, id: int) -> Dict:
+        obj = await ReportCRUD(auth).get_obj_by_id(id=id)
+        return ReportOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def get_list_services(cls, auth: AuthSchema, search: ReportQueryParams = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
+        obj_list = await ReportCRUD(auth).get_obj_list(search=search.__dict__, order_by=order_by)
+        return [ReportOutSchema.model_validate(obj).model_dump() for obj in obj_list]
+    
+    @classmethod
+    async def create_services(cls, auth: AuthSchema, data: ReportCreateSchema) -> Dict:
+        obj = await ReportCRUD(auth).create_obj(data=data)
+        return ReportOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def update_services(cls, auth: AuthSchema, data: ReportUpdateSchema) -> Dict:
+        obj = await ReportCRUD(auth).update_obj(id=data.id, data=data)
+        return ReportOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def delete_services(cls, auth: AuthSchema, id: int) -> None:
+        await ReportCRUD(auth).delete_obj(ids=[id])
+

backend/app/api/v1/services/autotest/suite_service.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+
+from typing import List, Dict
+
+from app.api.v1.schemas.system.auth_schema import AuthSchema
+from app.api.v1.schemas.autotest.suite_schema import SuiteOutSchema, SuiteCreateSchema, SuiteUpdateSchema
+from app.api.v1.params.autotest.suite_param import SuiteQueryParams
+from app.api.v1.cruds.autotest.suite_crud import SuiteCRUD
+
+
+class SuiteService:
+    """
+    套件服务层
+    """
+    
+    @classmethod
+    async def get_detail_services(cls, auth: AuthSchema, id: int) -> Dict:
+        obj = await SuiteCRUD(auth).get_obj_by_id(id=id)
+        return SuiteOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def get_list_services(cls, auth: AuthSchema, search: SuiteQueryParams = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
+        obj_list = await SuiteCRUD(auth).get_obj_list(search=search.__dict__, order_by=order_by)
+        return [SuiteOutSchema.model_validate(obj).model_dump() for obj in obj_list]
+    
+    @classmethod
+    async def create_services(cls, auth: AuthSchema, data: SuiteCreateSchema) -> Dict:
+        obj = await SuiteCRUD(auth).create_obj(data=data)
+        return SuiteOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def update_services(cls, auth: AuthSchema, data: SuiteUpdateSchema) -> Dict:
+        obj = await SuiteCRUD(auth).update_obj(id=data.id, data=data)
+        return SuiteOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def delete_services(cls, auth: AuthSchema, id: int) -> None:
+        await SuiteCRUD(auth).delete_obj(ids=[id])

backend/app/api/v1/services/autotest/task_service.py

@@ -0,0 +1,75 @@
+# -*- coding: utf-8 -*-
+
+from typing import List, Dict
+
+from app.api.v1.schemas.system.auth_schema import AuthSchema
+from app.api.v1.schemas.autotest.task_schema import TaskOutSchema, TaskCreateSchema, TaskUpdateSchema
+from app.api.v1.params.autotest.task_param import TaskQueryParams
+from app.api.v1.cruds.autotest.task_crud import TaskCRUD
+from app.api.v1.cruds.autotest.api_case_crud import APICaseCRUD
+from app.api.v1.cruds.autotest.project_crud import ProjectCRUD
+from app.api.v1.cruds.autotest.environment_crud import EnvironmentCRUD
+from app.api.v1.cruds.autotest.notification_config_crud import NotificationConfigCRUD
+from app.api.v1.cruds.autotest.global_data_crud import GlobalDataCRUD
+from app.api.v1.cruds.autotest.module_crud import ModuleCRUD
+from app.api.v1.cruds.autotest.repont_crud import ReportCRUD
+from app.utils.request_util import Requests
+
+
+
+class TaskService:
+    """
+    执行记录服务层
+    """
+    
+    @classmethod
+    async def get_detail_services(cls, auth: AuthSchema, id: int) -> Dict:
+        obj = await TaskCRUD(auth).get_obj_by_id(id=id)
+        return TaskOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def get_list_services(cls, auth: AuthSchema, search: TaskQueryParams = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
+        obj_list = await TaskCRUD(auth).get_obj_list(search=search.__dict__, order_by=order_by)
+        return [TaskOutSchema.model_validate(obj).model_dump() for obj in obj_list]
+    
+    @classmethod
+    async def create_services(cls, auth: AuthSchema, data: TaskCreateSchema, id: int) -> Dict:
+
+        api_case = await APICaseCRUD(auth).get_obj_by_id(id=id)
+        request_data = {
+            'url': api_case.url,
+            'method': api_case.method,
+            'data': api_case.body,
+            'headers': api_case.headers,
+        }
+
+        # 发送请求并获取响应
+        response = Requests().send_request(**request_data)
+        
+        # 验证结果
+        validation_results = [
+            api_case.expected_status_code == response['code'],
+            api_case.expected_response in response['body']['msg'],
+            api_case.assertions in str(response['body'])
+        ]
+        
+        status = 'pass' if all(validation_results) else 'fail'
+        
+        data.status = status
+        data.logs = response['time_total']
+        data.actual_response = response['body']
+        data.test_case_id = id
+        data.description = api_case.description
+
+
+        obj = await TaskCRUD(auth).create_obj(data=data)
+        return TaskOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def update_services(cls, auth: AuthSchema, data: TaskUpdateSchema) -> Dict:
+        obj = await TaskCRUD(auth).update_obj(id=data.id, data=data)
+        return TaskOutSchema.model_validate(obj).model_dump()
+    
+    @classmethod
+    async def delete_services(cls, auth: AuthSchema, id: int) -> None:
+        await TaskCRUD(auth).delete_obj(ids=[id])