"""智能终端 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)