重构(test): test_work_order_real 改用 constants 统一定义
This commit is contained in:
175
test_work_order_real.py
Normal file
175
test_work_order_real.py
Normal file
@@ -0,0 +1,175 @@
|
||||
"""
|
||||
工单创建测试 — 使用真实告警数据 + 真实截图
|
||||
|
||||
1. 从 MySQL 查询最近一条带截图的告警
|
||||
2. 用该告警数据创建工单到 192.168.0.104:48080
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import time
|
||||
import uuid
|
||||
|
||||
import httpx
|
||||
import pymysql
|
||||
|
||||
from app.constants import ALARM_TYPE_NAMES, ALARM_LEVEL_NAMES
|
||||
|
||||
# ===== 数据库配置 =====
|
||||
DB_CONFIG = {
|
||||
"host": "124.221.55.225",
|
||||
"port": 3306,
|
||||
"user": "vsp_user",
|
||||
"password": "VspPass2024!",
|
||||
"database": "aiot_alarm",
|
||||
"charset": "utf8mb4",
|
||||
}
|
||||
|
||||
# ===== 工单 API 配置 =====
|
||||
BASE_URL = "http://192.168.0.104:48080"
|
||||
APP_ID = "alarm-system"
|
||||
APP_SECRET = "tQ3v5q1z2ZLu7hrU1yseaHwg1wJUcmF1"
|
||||
TENANT_ID = "1"
|
||||
|
||||
|
||||
def query_recent_alarm_with_snapshot():
|
||||
"""从数据库查询最近一条带截图的告警"""
|
||||
conn = pymysql.connect(**DB_CONFIG)
|
||||
try:
|
||||
with conn.cursor(pymysql.cursors.DictCursor) as cur:
|
||||
sql = """
|
||||
SELECT alarm_id, alarm_type, device_id, scene_id,
|
||||
event_time, alarm_level, confidence_score,
|
||||
snapshot_url, edge_node_id, area_id
|
||||
FROM alarm_event
|
||||
WHERE snapshot_url IS NOT NULL
|
||||
AND snapshot_url != ''
|
||||
ORDER BY event_time DESC
|
||||
LIMIT 5
|
||||
"""
|
||||
cur.execute(sql)
|
||||
rows = cur.fetchall()
|
||||
return rows
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def build_sign(query_str: str, body_json: str, nonce: str, timestamp: str) -> str:
|
||||
"""构建 SHA256 签名"""
|
||||
header_str = f"appId={APP_ID}&nonce={nonce}×tamp={timestamp}"
|
||||
sign_string = f"{query_str}{body_json}{header_str}{APP_SECRET}"
|
||||
return hashlib.sha256(sign_string.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def build_headers(body_json: str) -> dict:
|
||||
"""构造完整请求头"""
|
||||
nonce = uuid.uuid4().hex[:16]
|
||||
timestamp = str(int(time.time() * 1000))
|
||||
sign = build_sign("", body_json, nonce, timestamp)
|
||||
return {
|
||||
"Content-Type": "application/json",
|
||||
"tenant-id": TENANT_ID,
|
||||
"appId": APP_ID,
|
||||
"timestamp": timestamp,
|
||||
"nonce": nonce,
|
||||
"sign": sign,
|
||||
}
|
||||
|
||||
|
||||
async def create_order(alarm: dict) -> str:
|
||||
"""用真实告警数据创建工单"""
|
||||
alarm_type = alarm["alarm_type"]
|
||||
alarm_level = alarm.get("alarm_level", 2)
|
||||
type_name = ALARM_TYPE_NAMES.get(alarm_type, alarm_type)
|
||||
level_name = ALARM_LEVEL_NAMES.get(alarm_level, "一般")
|
||||
device_id = alarm["device_id"]
|
||||
event_time = str(alarm["event_time"])[:19] if alarm["event_time"] else ""
|
||||
|
||||
body = {
|
||||
"title": f"【{level_name}】{type_name}告警 - {device_id}",
|
||||
"areaId": alarm.get("area_id") or 1317,
|
||||
"alarmId": alarm["alarm_id"],
|
||||
"alarmType": alarm_type,
|
||||
"description": f"设备 {device_id} 于 {event_time} 检测到{type_name}，置信度 {alarm.get('confidence_score', 0):.2f}",
|
||||
"priority": alarm_level,
|
||||
"triggerSource": "自动上报",
|
||||
"cameraId": device_id,
|
||||
"imageUrl": alarm["snapshot_url"],
|
||||
}
|
||||
|
||||
body_json = json.dumps(body, ensure_ascii=False, separators=(",", ":"))
|
||||
headers = build_headers(body_json)
|
||||
url = f"{BASE_URL}/open-api/ops/security/order/create"
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"创建工单 — 使用真实告警数据")
|
||||
print(f"{'='*60}")
|
||||
print(f" 告警ID: {alarm['alarm_id']}")
|
||||
print(f" 告警类型: {type_name}")
|
||||
print(f" 设备ID: {device_id}")
|
||||
print(f" 事件时间: {event_time}")
|
||||
print(f" 置信度: {alarm.get('confidence_score', 'N/A')}")
|
||||
print(f" 截图URL: {alarm['snapshot_url'][:80]}...")
|
||||
print(f" 区域ID: {body['areaId']}")
|
||||
print(f" 工单标题: {body['title']}")
|
||||
print()
|
||||
print(f" 请求URL: {url}")
|
||||
print(f" 请求Body: {body_json[:200]}...")
|
||||
print()
|
||||
|
||||
async with httpx.AsyncClient(timeout=15) as client:
|
||||
resp = await client.post(url, content=body_json, headers=headers)
|
||||
|
||||
print(f" HTTP Status: {resp.status_code}")
|
||||
print(f" Response Body: {resp.text[:500]}")
|
||||
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
if data.get("code") == 0:
|
||||
order_id = str(data.get("data", ""))
|
||||
print(f"\n ✅ 工单创建成功! orderId = {order_id}")
|
||||
return order_id
|
||||
else:
|
||||
print(f"\n ❌ API 返回错误: code={data.get('code')}, msg={data.get('msg')}")
|
||||
else:
|
||||
print(f"\n ❌ HTTP 请求失败: {resp.status_code}")
|
||||
return ""
|
||||
|
||||
|
||||
async def main():
|
||||
print(f"工单系统地址: {BASE_URL}")
|
||||
print(f"数据库: {DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}")
|
||||
print()
|
||||
|
||||
# 1. 查询最近带截图的告警
|
||||
print("正在查询最近带截图的告警...")
|
||||
alarms = query_recent_alarm_with_snapshot()
|
||||
|
||||
if not alarms:
|
||||
print("❌ 未找到带截图的告警，无法测试")
|
||||
return
|
||||
|
||||
print(f"\n找到 {len(alarms)} 条带截图的告警：")
|
||||
print(f"{'─'*80}")
|
||||
for i, a in enumerate(alarms):
|
||||
type_name = ALARM_TYPE_NAMES.get(a["alarm_type"], a["alarm_type"])
|
||||
print(f" [{i+1}] {a['alarm_id'][:40]}...")
|
||||
print(f" 类型={type_name} 设备={a['device_id']} 时间={str(a['event_time'])[:19]}")
|
||||
print(f" 截图={a['snapshot_url'][:70]}...")
|
||||
print()
|
||||
|
||||
# 2. 用第一条创建工单
|
||||
alarm = alarms[0]
|
||||
print(f"将使用第 1 条告警创建工单...")
|
||||
order_id = await create_order(alarm)
|
||||
|
||||
if order_id:
|
||||
print(f"\n{'='*60}")
|
||||
print(f"测试完成 — 工单 {order_id} 已创建")
|
||||
print(f"请在 AIoT 平台确认工单内容和截图是否正确")
|
||||
print(f"{'='*60}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user