120 lines
6.2 KiB
Python
120 lines
6.2 KiB
Python
import json
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
|
|
from app.models import AlgorithmModel, StreamModel
|
|
from .models import CameraCalibration, WorkshopSite
|
|
|
|
|
|
def make_stream(code, index):
|
|
return StreamModel.objects.create(
|
|
user_id=1, sort=index, code=code, app="live", name=code,
|
|
pull_stream_url="", pull_stream_type=21, pull_stream_transfer_mode=0,
|
|
pull_stream_ip="127.0.0.1", pull_stream_port=0,
|
|
pull_stream_username="", pull_stream_password="", nickname="Camera %d" % index,
|
|
remark="", forward_state=0, is_audio=0, snap_filepath="", camera_sum_num=1,
|
|
camera_name="Camera %d" % index, camera_manufacturer="test", camera_owner="",
|
|
camera_model="test", camera_device_id=code, camera_parent_id="", camera_civilcode="",
|
|
state=0,
|
|
)
|
|
|
|
|
|
class WorkshopApiTests(TestCase):
|
|
def setUp(self):
|
|
self.admin = get_user_model().objects.create_superuser("wm-admin", "wm@example.com", "pass")
|
|
self.client.force_login(self.admin)
|
|
self.detector = AlgorithmModel.objects.create(
|
|
name="detector", algorithm_type="yolo11", task_type="detect",
|
|
inference_engine="onnxruntime", device="cpu", model_file="missing.onnx",
|
|
labels='["person"]', state=1, is_default=1,
|
|
)
|
|
for i, code in enumerate((
|
|
"34020000001320000001", "34020000001320000002",
|
|
"34020000001320000003", "34020000001320000004",
|
|
), start=1):
|
|
make_stream(code, i)
|
|
|
|
def test_config_initializes_existing_four_streams(self):
|
|
response = self.client.get("/workshop/openConfig")
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(payload["code"], 1000)
|
|
self.assertEqual(len(payload["data"]["cameras"]), 4)
|
|
self.assertTrue(all(c["stream_id"] for c in payload["data"]["cameras"]))
|
|
self.assertTrue(all(c["preview_stream_id"] is None for c in payload["data"]["cameras"]))
|
|
self.assertEqual(payload["data"]["site"]["width_m"], 130.0)
|
|
|
|
def test_separate_preview_substream_is_persisted(self):
|
|
config = self.client.get("/workshop/openConfig").json()["data"]
|
|
preview = make_stream("34020000001320000102", 102)
|
|
site = config["site"]
|
|
camera = config["cameras"][0]
|
|
response = self.client.post("/workshop/openSaveConfig", data=json.dumps({
|
|
"name": site["name"], "width_m": site["width_m"], "height_m": site["height_m"],
|
|
"analysis_fps": site["analysis_fps"], "target_labels": site["target_labels"],
|
|
"fusion_radius_m": site["fusion_radius_m"],
|
|
"observation_window_sec": site["observation_window_sec"],
|
|
"max_speed_mps": site["max_speed_mps"], "lost_ttl_sec": site["lost_ttl_sec"],
|
|
"trail_sec": site["trail_sec"], "detector_id": site["detector_id"],
|
|
"reid_model_id": site["reid_model_id"],
|
|
"cameras": [{**camera, "preview_stream_id": preview.id}],
|
|
}), content_type="application/json")
|
|
payload = response.json()
|
|
self.assertEqual(payload["code"], 1000)
|
|
saved = payload["data"]["cameras"][0]
|
|
self.assertEqual(saved["preview_stream_id"], preview.id)
|
|
self.assertEqual(saved["preview_stream_name"], preview.name)
|
|
|
|
def test_preview_substream_cannot_equal_analysis_stream(self):
|
|
config = self.client.get("/workshop/openConfig").json()["data"]
|
|
site = config["site"]
|
|
camera = config["cameras"][0]
|
|
response = self.client.post("/workshop/openSaveConfig", data=json.dumps({
|
|
"name": site["name"], "width_m": site["width_m"], "height_m": site["height_m"],
|
|
"analysis_fps": site["analysis_fps"], "target_labels": site["target_labels"],
|
|
"fusion_radius_m": site["fusion_radius_m"],
|
|
"observation_window_sec": site["observation_window_sec"],
|
|
"max_speed_mps": site["max_speed_mps"], "lost_ttl_sec": site["lost_ttl_sec"],
|
|
"trail_sec": site["trail_sec"], "detector_id": site["detector_id"],
|
|
"reid_model_id": site["reid_model_id"],
|
|
"cameras": [{**camera, "preview_stream_id": camera["stream_id"]}],
|
|
}), content_type="application/json")
|
|
self.assertEqual(response.json()["code"], 0)
|
|
self.assertIn("不能与定位/标定流相同", response.json()["msg"])
|
|
|
|
def test_calibration_activation_requires_validation_under_one_meter(self):
|
|
config = self.client.get("/workshop/openConfig").json()["data"]
|
|
camera_id = config["cameras"][0]["id"]
|
|
rows = [
|
|
{"u": 0, "v": 0, "x": 0, "y": 0, "role": "fit"},
|
|
{"u": 1300, "v": 0, "x": 130, "y": 0, "role": "fit"},
|
|
{"u": 1300, "v": 500, "x": 130, "y": 50, "role": "fit"},
|
|
{"u": 0, "v": 500, "x": 0, "y": 50, "role": "fit"},
|
|
{"u": 200, "v": 100, "x": 20, "y": 10, "role": "verify"},
|
|
{"u": 650, "v": 250, "x": 65, "y": 25, "role": "verify"},
|
|
{"u": 1100, "v": 400, "x": 110, "y": 40, "role": "verify"},
|
|
]
|
|
response = self.client.post("/workshop/openCalibrate", data=json.dumps({
|
|
"camera_id": camera_id, "frame_width": 1300, "frame_height": 500,
|
|
"observations": rows, "activate": True,
|
|
}), content_type="application/json")
|
|
self.assertEqual(response.json()["code"], 1000)
|
|
calibration = CameraCalibration.objects.get()
|
|
self.assertTrue(calibration.is_active)
|
|
self.assertEqual(calibration.status, CameraCalibration.STATUS_VALID)
|
|
|
|
def test_viewer_cannot_change_configuration(self):
|
|
viewer = get_user_model().objects.create_user("wm-viewer", password="pass")
|
|
self.client.force_login(viewer)
|
|
response = self.client.post("/workshop/openSaveConfig", data="{}", content_type="application/json")
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_workshop_tables_do_not_create_control_or_alarm_rows(self):
|
|
self.client.get("/workshop/openConfig")
|
|
site = WorkshopSite.objects.get()
|
|
self.assertEqual(site.cameras.count(), 4)
|
|
from app.models import AlarmModel, ZoneModel
|
|
self.assertEqual(AlarmModel.objects.count(), 0)
|
|
self.assertEqual(ZoneModel.objects.count(), 0)
|