288 lines
10 KiB
Python
288 lines
10 KiB
Python
"""智能终端 API 接口测试"""
|
||
import uuid
|
||
|
||
from django.contrib.auth import get_user_model
|
||
from django.db import connection
|
||
from django.test import TestCase
|
||
from rest_framework.test import APIClient
|
||
|
||
from apps.core.models import AiSmartTerminal
|
||
|
||
User = get_user_model()
|
||
|
||
|
||
# ---------- 5 条测试设备数据 ----------
|
||
TEST_DEVICES = [
|
||
{
|
||
"stcd": "ZD001",
|
||
"stnm": "碧水电站-鱼类识别AI盒子",
|
||
"sttp": "01",
|
||
"lgtd": 104.065800,
|
||
"lttd": 30.572800,
|
||
"elev": 512.500000,
|
||
"rstcd": "BS001",
|
||
"base_id": "YJ001",
|
||
"rvcd": "MJ001",
|
||
"aipz": '{"cpu":"ARM Cortex-A78","memory":"8GB","storage":"64GB"}',
|
||
"ipaddr": "192.168.1.101",
|
||
"usfl": 1,
|
||
"dtin": 1,
|
||
"mway": 1,
|
||
"sync_check_interval": 30,
|
||
},
|
||
{
|
||
"stcd": "ZD002",
|
||
"stnm": "青湖-鸟类识别AI盒子",
|
||
"sttp": "02",
|
||
"lgtd": 104.125600,
|
||
"lttd": 30.681200,
|
||
"elev": 489.300000,
|
||
"rstcd": "QH002",
|
||
"base_id": "YJ001",
|
||
"rvcd": "MJ002",
|
||
"aipz": '{"cpu":"ARM Cortex-A78","memory":"4GB","storage":"32GB"}',
|
||
"ipaddr": "192.168.1.102",
|
||
"usfl": 1,
|
||
"dtin": 1,
|
||
"mway": 1,
|
||
"sync_check_interval": 60,
|
||
},
|
||
{
|
||
"stcd": "ZD003",
|
||
"stnm": "卧龙自然保护区-野生动物识别AI盒子",
|
||
"sttp": "03",
|
||
"lgtd": 103.298400,
|
||
"lttd": 30.985600,
|
||
"elev": 1820.000000,
|
||
"rstcd": "WL003",
|
||
"base_id": "YJ002",
|
||
"rvcd": "MJ003",
|
||
"aipz": '{"cpu":"NVIDIA Jetson Orin","memory":"16GB","storage":"256GB"}',
|
||
"ipaddr": "192.168.2.101",
|
||
"usfl": 1,
|
||
"dtin": 0,
|
||
"mway": 2,
|
||
"sync_check_interval": 120,
|
||
},
|
||
{
|
||
"stcd": "ZD004",
|
||
"stnm": "长江上游-鱼类分拣装置",
|
||
"sttp": "04",
|
||
"lgtd": 104.545200,
|
||
"lttd": 30.127800,
|
||
"elev": 345.000000,
|
||
"rstcd": "CJ004",
|
||
"base_id": "YJ003",
|
||
"rvcd": "CJ001",
|
||
"aipz": '{"cpu":"Intel i5-1135G7","memory":"8GB","storage":"128GB","grab_arm":true}',
|
||
"ipaddr": "192.168.3.201",
|
||
"usfl": 0,
|
||
"dtin": 0,
|
||
"mway": 0,
|
||
"sync_check_interval": None,
|
||
},
|
||
{
|
||
"stcd": "ZD005",
|
||
"stnm": "洞庭湖-鸟类识别AI盒子",
|
||
"sttp": "02",
|
||
"lgtd": 112.946800,
|
||
"lttd": 29.358700,
|
||
"elev": 35.200000,
|
||
"rstcd": "DT005",
|
||
"base_id": "YJ004",
|
||
"rvcd": "CJ002",
|
||
"aipz": '{"cpu":"ARM Cortex-A78","memory":"8GB","storage":"64GB"}',
|
||
"ipaddr": "192.168.4.101",
|
||
"usfl": 1,
|
||
"dtin": 1,
|
||
"mway": 1,
|
||
"sync_check_interval": 45,
|
||
},
|
||
]
|
||
|
||
|
||
class SmartTerminalAPITest(TestCase):
|
||
"""智能终端 API 集成测试 — 覆盖全部 7 个接口"""
|
||
|
||
@classmethod
|
||
def setUpTestData(cls):
|
||
"""类级别：测试库建表 + 创建 admin 用户 + 插入 5 条测试设备"""
|
||
cls.base_url = "/server/terminal"
|
||
# Django 测试每次新建 test_ 库，需在测试库中建表
|
||
with connection.schema_editor() as schema_editor:
|
||
schema_editor.create_model(AiSmartTerminal)
|
||
# 创建 admin 用户
|
||
cls.admin_user = User.objects.create_user(
|
||
username="admin", password="123456", is_active=True, is_staff=True,
|
||
)
|
||
# 通过 ORM 直接插入 5 条设备
|
||
cls.created_ids = []
|
||
for device in TEST_DEVICES:
|
||
obj = AiSmartTerminal.objects.create(
|
||
id=uuid.uuid4().hex,
|
||
record_user="admin",
|
||
**device,
|
||
)
|
||
cls.created_ids.append(obj.id)
|
||
assert AiSmartTerminal.objects.filter(is_deleted=0).count() == 5
|
||
|
||
def setUp(self):
|
||
"""每个测试用例前：用 admin 强制登录"""
|
||
self.client = APIClient()
|
||
self.client.force_login(self.admin_user)
|
||
|
||
# ==================== 1. 创建 ====================
|
||
def test_01_create_duplicate_stcd_should_reject(self):
|
||
"""重复的 stcd 应被拒绝"""
|
||
resp = self.client.post(f"{self.base_url}/create/", {
|
||
"stcd": "ZD001",
|
||
"stnm": "重复设备",
|
||
"sttp": "01",
|
||
}, format="json")
|
||
self.assertEqual(resp.status_code, 400)
|
||
|
||
# ==================== 2. 修改 ====================
|
||
def test_02_update_terminal_info(self):
|
||
"""修改设备基本信息"""
|
||
target = self.created_ids[0]
|
||
resp = self.client.put(f"{self.base_url}/update/", {
|
||
"id": target,
|
||
"stnm": "碧水电站-鱼类识别AI盒子（已升级）",
|
||
"sync_check_interval": 15,
|
||
}, format="json")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertEqual(resp.data["stnm"], "碧水电站-鱼类识别AI盒子（已升级）")
|
||
|
||
def test_03_update_protected_fields_should_be_ignored(self):
|
||
"""修改受保护字段应被忽略"""
|
||
target = self.created_ids[0]
|
||
resp = self.client.put(f"{self.base_url}/update/", {
|
||
"id": target,
|
||
"model_version": "hacked_v99",
|
||
"online_status": 0,
|
||
}, format="json")
|
||
self.assertEqual(resp.status_code, 200)
|
||
# 受保护字段不应被修改
|
||
self.assertIsNone(resp.data["model_version"])
|
||
self.assertIsNone(resp.data["online_status"])
|
||
|
||
# ==================== 3. 发布模型 ====================
|
||
def test_04_publish_model_to_single_terminal(self):
|
||
"""发布模型到单个终端"""
|
||
target = self.created_ids[0]
|
||
resp = self.client.put(f"{self.base_url}/publishModel/", {
|
||
"id": target,
|
||
"model_id": "model_fish_v2",
|
||
"model_version": "v2.1.0",
|
||
"model_path": "models/fish_detect/v2.1.0/model.onnx",
|
||
}, format="json")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertEqual(resp.data["published_count"], 1)
|
||
updated = AiSmartTerminal.objects.get(id=target)
|
||
self.assertEqual(updated.model_version, "v2.1.0")
|
||
self.assertIsNotNone(updated.model_publish_time)
|
||
|
||
def test_05_publish_model_to_multiple_terminals(self):
|
||
"""发布模型到多个终端"""
|
||
targets = self.created_ids[:3]
|
||
resp = self.client.put(f"{self.base_url}/publishModel/", {
|
||
"ids": targets,
|
||
"model_id": "model_bird_v1",
|
||
"model_version": "v1.5.0",
|
||
"model_path": "models/bird_detect/v1.5.0/model.onnx",
|
||
}, format="json")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertEqual(resp.data["published_count"], 3)
|
||
for tid in targets:
|
||
t = AiSmartTerminal.objects.get(id=tid)
|
||
self.assertEqual(t.model_version, "v1.5.0")
|
||
|
||
def test_06_publish_model_missing_params(self):
|
||
"""发布模型缺少参数应报错"""
|
||
resp = self.client.put(f"{self.base_url}/publishModel/", {
|
||
"id": self.created_ids[0],
|
||
"model_id": "m1",
|
||
}, format="json")
|
||
self.assertEqual(resp.status_code, 400)
|
||
|
||
# ==================== 4. 分页查询 ====================
|
||
def test_07_search_all_returns_paginated(self):
|
||
"""查询全部设备应返回分页数据"""
|
||
resp = self.client.get(f"{self.base_url}/list/")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertIn("count", resp.data)
|
||
self.assertIn("results", resp.data)
|
||
self.assertEqual(resp.data["count"], 5)
|
||
|
||
def test_08_search_by_sttp(self):
|
||
"""按终端分类过滤"""
|
||
resp = self.client.get(f"{self.base_url}/list/?sttp=02")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertEqual(resp.data["count"], 2) # ZD002 + ZD005
|
||
|
||
def test_09_search_by_usfl(self):
|
||
"""按启用状态过滤"""
|
||
resp = self.client.get(f"{self.base_url}/list/?usfl=0")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertEqual(resp.data["count"], 1) # ZD004
|
||
|
||
def test_10_search_by_stnm_fuzzy(self):
|
||
"""按终端名称模糊搜索"""
|
||
resp = self.client.get(f"{self.base_url}/list/?stnm=鱼类")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertGreaterEqual(resp.data["count"], 2) # ZD001 + ZD004
|
||
|
||
def test_11_pagination_page_size(self):
|
||
"""自定义分页大小"""
|
||
resp = self.client.get(f"{self.base_url}/list/?page_size=2")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertEqual(len(resp.data["results"]), 2)
|
||
self.assertEqual(resp.data["count"], 5)
|
||
|
||
# ==================== 5. 心跳上报 ====================
|
||
def test_12_heartbeat_updates_online_status(self):
|
||
"""心跳上报应更新在线状态和心跳时间"""
|
||
resp = self.client.put(f"{self.base_url}/heartbeat/", {
|
||
"stcd": "ZD001",
|
||
}, format="json")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertEqual(resp.data["online_status"], 1)
|
||
self.assertIsNotNone(resp.data["last_heartbeat_time"])
|
||
|
||
def test_13_heartbeat_missing_stcd(self):
|
||
"""心跳上报缺少 stcd 应报错"""
|
||
resp = self.client.put(f"{self.base_url}/heartbeat/", {}, format="json")
|
||
self.assertEqual(resp.status_code, 400)
|
||
|
||
def test_14_heartbeat_unknown_stcd(self):
|
||
"""心跳上报未知 stcd 应返回 404"""
|
||
resp = self.client.put(f"{self.base_url}/heartbeat/", {
|
||
"stcd": "UNKNOWN",
|
||
}, format="json")
|
||
self.assertEqual(resp.status_code, 404)
|
||
|
||
# ==================== 6. 下载模型 ====================
|
||
def test_15_download_model_no_model_published(self):
|
||
"""未发布模型的终端下载应返回 404"""
|
||
resp = self.client.get(f"{self.base_url}/downloadModel/?stcd=ZD005")
|
||
self.assertEqual(resp.status_code, 404)
|
||
|
||
def test_16_download_model_missing_stcd(self):
|
||
"""下载模型缺少 stcd 应报错"""
|
||
resp = self.client.get(f"{self.base_url}/downloadModel/")
|
||
self.assertEqual(resp.status_code, 400)
|
||
|
||
# ==================== 7. 逻辑删除 ====================
|
||
def test_17_delete_soft(self):
|
||
"""逻辑删除后 is_deleted 应为 1"""
|
||
target = self.created_ids[4]
|
||
resp = self.client.delete(f"{self.base_url}/delete/?id={target}")
|
||
self.assertEqual(resp.status_code, 200)
|
||
deleted = AiSmartTerminal.objects.get(id=target)
|
||
self.assertEqual(deleted.is_deleted, 1)
|
||
|
||
def test_18_delete_missing_id(self):
|
||
"""删除缺少 id 应报错"""
|
||
resp = self.client.delete(f"{self.base_url}/delete/")
|
||
self.assertEqual(resp.status_code, 400)
|