1#!/usr/bin/env python3 2# coding=utf-8 3 4# 5# Copyright (c) Huawei Device Co., Ltd. 2025. All right reserved. 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17# 18 19import uuid 20from datetime import datetime 21from typing import Dict, List, Union 22 23from pydantic import BaseModel 24from sqlmodel import Field, SQLModel 25 26from _core.constants import State 27from _core.environment.device_state import DeviceAllocationState 28 29 30class Device(SQLModel): 31 id: Union[int, None] = Field(default=None, primary_key=True) 32 ip: str = Field(index=True) 33 sn: str = Field(index=True) 34 os: str 35 type: str 36 model: str 37 version: str 38 worker_url: str = Field(description="worker地址") 39 state: str = Field(index=True, description="设备状态") 40 41 42class TestDevice(Device, table=True): 43 usage_state: str = Field(default=DeviceAllocationState.available, description="占用状态") 44 create_time: datetime = Field(default_factory=datetime.now, description="创建时间") 45 update_time: datetime = Field(default_factory=datetime.now, description="更新时间", 46 sa_column_kwargs={"onupdate": datetime.now}) 47 48 49class Project(BaseModel): 50 url: str = Field(description="工程地址") 51 mode: str = Field(description="工程模式(git/svn/http/local)") 52 branch: str = Field(default="", description="分支名称") 53 username: str = Field(default="", description="用户名") 54 password: str = Field(default="", description="密码") 55 relative_path: str = Field(default="", description="相对路径") 56 57 58class TaskStopParam(BaseModel): 59 task_id: str 60 block_id: str 61 62 63class TaskTemplate(BaseModel): 64 project: Project 65 cases: List[str] = [] 66 devices: List[Dict[str, Union[int, str]]] = [] 67 68 69class TaskInfo(TaskTemplate): 70 task_id: str 71 block_id: str = "" 72 73 74class Task(SQLModel): 75 id: str = Field(primary_key=True, description="任务编号") 76 state: str = Field(default=State.Waiting, index=True, description="运行状态") 77 begin_time: Union[datetime, None] = Field(default=None, description="开始时间") 78 end_time: Union[datetime, None] = Field(default=None, description="结束时间") 79 error_message: str = Field(default="", description="报错信息") 80 report_url: str = Field(default="", description="报告地址") 81 create_time: datetime = Field(default_factory=datetime.now, description="创建时间") 82 83 84class TestTask(Task, table=True): 85 info: str = Field(description="任务模板信息") 86 update_time: datetime = Field(default_factory=datetime.now, description="更新时间", 87 sa_column_kwargs={"onupdate": datetime.now}) 88 89 90class TestBlock(SQLModel, table=True): 91 id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True, description="任务块编号") 92 task_id: str = Field(index=True, description="任务编号") 93 info: str = Field(description="任务块信息") 94 environment: str = Field(default="", description="环境描述") 95 state: str = Field(default=State.Waiting, index=True, description="运行状态") 96 begin_time: Union[datetime, None] = Field(default=None, description="开始时间") 97 end_time: Union[datetime, None] = Field(default=None, description="结束时间") 98 worker_url: str = Field(default="", description="worker地址") 99 create_time: datetime = Field(default_factory=datetime.now, description="创建时间") 100 update_time: datetime = Field(default_factory=datetime.now, description="更新时间", 101 sa_column_kwargs={"onupdate": datetime.now}) 102 103 104class ResponseMsg(BaseModel): 105 status: str = "ok" 106 message: str = "" 107