vent_agent_2.0/db/test_lab_store.py

"""测试资料和执行快照；任务状态、结果仍由业务任务表管理。"""
import json

from db import db_conn


def assets(user):
    return db_conn.query('SELECT * FROM test_lab_assets WHERE user_name=? ORDER BY created_at DESC,id DESC', (user,))


def asset(identifier, user):
    row = db_conn.query_one('SELECT * FROM test_lab_assets WHERE id=? AND user_name=?', (identifier, user))
    if not row:
        raise ValueError('测试资料不存在')
    return row


def save_asset(user, filename, digest, size, identifier):
    db_conn.execute('INSERT INTO test_lab_assets (id,user_name,filename,sha256,size,created_at) VALUES (?,?,?,?,?,?)',
        (identifier,user,filename,digest,size,db_conn.fmt_dt()))
    return asset(identifier, user)


def save_run(job, label, scenario, mode, request):
    db_conn.execute('INSERT INTO test_lab_runs (job_id,user_name,agent_name,label,scenario_id,execution_mode,request_json,created_at) VALUES (?,?,?,?,?,?,?,?)',
        (job['id'],job['user_name'],job['agent_name'],label,scenario,mode,json.dumps(request,ensure_ascii=False,allow_nan=False),db_conn.fmt_dt()))


def run(identifier, user):
    row = db_conn.query_one('SELECT * FROM test_lab_runs WHERE job_id=? AND user_name=?', (identifier,user))
    if not row:
        raise ValueError('测试记录不存在')
    return row


def runs(user, limit=50):
    return db_conn.query('SELECT r.*,j.status,j.finished_at FROM test_lab_runs r JOIN business_jobs j ON j.id=r.job_id WHERE r.user_name=? ORDER BY r.created_at DESC,r.job_id DESC LIMIT ?', (user,limit))