338 lines
13 KiB
Python
338 lines
13 KiB
Python
import importlib.util
|
|
import json
|
|
import pathlib
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
SCRIPT_PATH = REPO_ROOT / "tools" / "render_config.py"
|
|
|
|
|
|
def load_module():
|
|
spec = importlib.util.spec_from_file_location("render_config", SCRIPT_PATH)
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class RenderConfigTest(unittest.TestCase):
|
|
def test_render_supports_profile_publish_placeholders(self):
|
|
template_path = REPO_ROOT / "configs" / "templates" / "workshop_face_shoe_alarm.json"
|
|
profile_path = REPO_ROOT / "configs" / "profiles" / "local_3588_test.json"
|
|
|
|
template = json.loads(template_path.read_text(encoding="utf-8"))
|
|
profile = json.loads(profile_path.read_text(encoding="utf-8"))
|
|
|
|
publish_node = next(node for node in template["template"]["nodes"] if node["id"] == "publish")
|
|
alarm_node = next(node for node in template["template"]["nodes"] if node["id"] == "alarm")
|
|
params = profile["instances"][0]["params"]
|
|
|
|
self.assertEqual(publish_node["outputs"][0]["path"], "${publish_hls_path}")
|
|
self.assertEqual(publish_node["outputs"][1]["port"], "${publish_rtsp_port}")
|
|
self.assertEqual(publish_node["outputs"][1]["path"], "${publish_rtsp_path}")
|
|
self.assertEqual(alarm_node["actions"]["external_api"]["channelNo"], "${channel_no}")
|
|
|
|
self.assertIn("publish_hls_path", params)
|
|
self.assertIn("publish_rtsp_port", params)
|
|
self.assertIn("publish_rtsp_path", params)
|
|
self.assertIn("channel_no", params)
|
|
self.assertNotIn("minio_endpoint", params)
|
|
self.assertNotIn("external_get_token_url", params)
|
|
self.assertNotIn("tenant_code", params)
|
|
|
|
shared_params = template["params"]
|
|
self.assertEqual(shared_params["minio_endpoint"], "http://10.0.0.49:9000")
|
|
self.assertEqual(shared_params["external_get_token_url"], "http://10.0.0.49:8080/api/getToken")
|
|
self.assertEqual(shared_params["tenant_code"], "32")
|
|
|
|
def test_render_profile_no_longer_requires_face_gallery_path(self):
|
|
profile_path = REPO_ROOT / "configs" / "profiles" / "local_3588_test.json"
|
|
profile = json.loads(profile_path.read_text(encoding="utf-8"))
|
|
params = profile["instances"][0]["params"]
|
|
|
|
self.assertNotIn("face_gallery_path", params)
|
|
self.assertNotIn("rga_gate", params)
|
|
|
|
def test_render_template_internalizes_clear_rga_gate_name(self):
|
|
template_path = REPO_ROOT / "configs" / "templates" / "workshop_face_shoe_alarm.json"
|
|
template = json.loads(template_path.read_text(encoding="utf-8"))
|
|
nodes = {node["id"]: node for node in template["template"]["nodes"]}
|
|
|
|
self.assertEqual(nodes["pre_rgb"]["rga_gate"], "main_pipeline_rga")
|
|
self.assertEqual(nodes["person_det"]["rga_gate"], "main_pipeline_rga")
|
|
self.assertEqual(nodes["shoe_det"]["rga_gate"], "main_pipeline_rga")
|
|
self.assertEqual(nodes["pre_osd"]["rga_gate"], "main_pipeline_rga")
|
|
|
|
def test_renders_profile_and_overlay(self):
|
|
module = load_module()
|
|
template = {
|
|
"name": "pipeline",
|
|
"params": {"tenant_code": "32"},
|
|
"template": {
|
|
"executor": {"batch_size": 2},
|
|
"nodes": [
|
|
{"id": "in", "type": "input_rtsp", "url": "${rtsp_url}"},
|
|
{"id": "face_recog", "type": "ai_face_recog", "debug": {"enabled": False}},
|
|
],
|
|
"edges": [["in", "face_recog"]],
|
|
},
|
|
}
|
|
profile = {
|
|
"queue": {"size": 8},
|
|
"instances": [
|
|
{
|
|
"name": "cam1",
|
|
"params": {"rtsp_url": "rtsp://example/cam1"},
|
|
}
|
|
],
|
|
}
|
|
overlay = {
|
|
"instance_overrides": {
|
|
"*": {
|
|
"override": {
|
|
"nodes": {
|
|
"face_recog": {"debug": {"enabled": True, "log_matches": True}}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
root = {
|
|
"templates": {module.template_name(template, pathlib.Path("x.json")): module.template_body(template)},
|
|
"instances": module.merge_template_params(module.profile_instances(profile, "pipeline"), module.template_params(template)),
|
|
"queue": profile["queue"],
|
|
}
|
|
rendered = module.apply_overlay(root, overlay)
|
|
|
|
self.assertEqual(rendered["queue"]["size"], 8)
|
|
self.assertEqual(rendered["instances"][0]["template"], "pipeline")
|
|
self.assertEqual(rendered["instances"][0]["params"]["rtsp_url"], "rtsp://example/cam1")
|
|
self.assertEqual(rendered["instances"][0]["params"]["tenant_code"], "32")
|
|
node_override = rendered["instances"][0]["override"]["nodes"]["face_recog"]
|
|
self.assertTrue(node_override["debug"]["enabled"])
|
|
self.assertTrue(node_override["debug"]["log_matches"])
|
|
|
|
def test_template_params_fill_placeholders_without_staying_in_profile(self):
|
|
module = load_module()
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
tmp = pathlib.Path(tmp_dir)
|
|
template_path = tmp / "template.json"
|
|
profile_path = tmp / "profile.json"
|
|
|
|
template_path.write_text(
|
|
"""
|
|
{
|
|
"name": "pipeline",
|
|
"params": {
|
|
"minio_endpoint": "http://10.0.0.49:9000",
|
|
"tenant_code": "32"
|
|
},
|
|
"template": {
|
|
"nodes": [{"id": "alarm", "type": "alarm", "endpoint": "${minio_endpoint}", "tenant": "${tenant_code}"}],
|
|
"edges": []
|
|
}
|
|
}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
profile_path.write_text(
|
|
"""
|
|
{
|
|
"name": "local_3588_test",
|
|
"instances": [
|
|
{
|
|
"name": "cam1",
|
|
"params": {"rtsp_url": "rtsp://example/cam1"}
|
|
}
|
|
]
|
|
}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
rendered = module.render(template_path, profile_path, [])
|
|
params = rendered["instances"][0]["params"]
|
|
self.assertEqual(params["minio_endpoint"], "http://10.0.0.49:9000")
|
|
self.assertEqual(params["tenant_code"], "32")
|
|
self.assertEqual(params["rtsp_url"], "rtsp://example/cam1")
|
|
|
|
def test_render_applies_multiple_overlays_in_order(self):
|
|
module = load_module()
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
tmp = pathlib.Path(tmp_dir)
|
|
template_path = tmp / "template.json"
|
|
profile_path = tmp / "profile.json"
|
|
overlay1_path = tmp / "overlay1.json"
|
|
overlay2_path = tmp / "overlay2.json"
|
|
|
|
template_path.write_text(
|
|
"""
|
|
{
|
|
"name": "pipeline",
|
|
"template": {
|
|
"nodes": [{"id": "alarm", "type": "alarm"}],
|
|
"edges": []
|
|
}
|
|
}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
profile_path.write_text(
|
|
"""
|
|
{
|
|
"instances": [{"name": "cam1", "params": {}}]
|
|
}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
overlay1_path.write_text(
|
|
"""
|
|
{
|
|
"instance_overrides": {
|
|
"*": {
|
|
"override": {
|
|
"nodes": {
|
|
"alarm": {
|
|
"face_debug": {"log_unknown_candidates": true},
|
|
"rules": [{"name": "shoe", "cooldown_ms": 1000}]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
overlay2_path.write_text(
|
|
"""
|
|
{
|
|
"instance_overrides": {
|
|
"*": {
|
|
"override": {
|
|
"nodes": {
|
|
"alarm": {
|
|
"face_debug": {"unknown_candidate_interval_ms": 0},
|
|
"rules": [{"name": "shoe", "cooldown_ms": 15000}]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
rendered = module.render(template_path, profile_path, [overlay1_path, overlay2_path])
|
|
alarm = rendered["instances"][0]["override"]["nodes"]["alarm"]
|
|
self.assertTrue(alarm["face_debug"]["log_unknown_candidates"])
|
|
self.assertEqual(alarm["face_debug"]["unknown_candidate_interval_ms"], 0)
|
|
self.assertEqual(alarm["rules"][0]["cooldown_ms"], 15000)
|
|
|
|
def test_render_adds_trace_metadata_when_requested(self):
|
|
module = load_module()
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
tmp = pathlib.Path(tmp_dir)
|
|
template_path = tmp / "template.json"
|
|
profile_path = tmp / "profile.json"
|
|
|
|
template_path.write_text(
|
|
"""
|
|
{
|
|
"name": "pipeline",
|
|
"template": {
|
|
"nodes": [{"id": "in", "type": "input_rtsp"}],
|
|
"edges": []
|
|
}
|
|
}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
profile_path.write_text(
|
|
"""
|
|
{
|
|
"name": "cam1",
|
|
"params": {"rtsp_url": "rtsp://example/cam1"}
|
|
}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
rendered = module.render(
|
|
template_path,
|
|
profile_path,
|
|
[],
|
|
metadata={
|
|
"config_id": "local_3588_face_debug",
|
|
"config_version": "20260419.001",
|
|
"rendered_at": "2026-04-19T00:00:00Z",
|
|
"rendered_by": "test",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(rendered["metadata"]["config_id"], "local_3588_face_debug")
|
|
self.assertEqual(rendered["metadata"]["config_version"], "20260419.001")
|
|
self.assertEqual(rendered["metadata"]["template"], "pipeline")
|
|
self.assertEqual(rendered["metadata"]["profile"], "cam1")
|
|
self.assertEqual(rendered["metadata"]["overlays"], [])
|
|
self.assertEqual(rendered["metadata"]["rendered_by"], "test")
|
|
|
|
def test_render_metadata_includes_instance_identity_summary(self):
|
|
module = load_module()
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
tmp = pathlib.Path(tmp_dir)
|
|
template_path = tmp / "template.json"
|
|
profile_path = tmp / "profile.json"
|
|
|
|
template_path.write_text(
|
|
"""
|
|
{
|
|
"name": "pipeline",
|
|
"template": {
|
|
"nodes": [{"id": "in", "type": "input_rtsp"}],
|
|
"edges": []
|
|
}
|
|
}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
profile_path.write_text(
|
|
"""
|
|
{
|
|
"name": "local_3588_test",
|
|
"business_name": "A厂区视觉识别",
|
|
"instances": [
|
|
{
|
|
"name": "cam1",
|
|
"params": {
|
|
"display_name": "东门入口",
|
|
"rtsp_url": "rtsp://example/cam1"
|
|
}
|
|
}
|
|
]
|
|
}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
rendered = module.render(
|
|
template_path,
|
|
profile_path,
|
|
[],
|
|
metadata={"config_id": "cfg1", "rendered_by": "test"},
|
|
)
|
|
|
|
self.assertEqual(rendered["metadata"]["profile"], "local_3588_test")
|
|
self.assertEqual(rendered["metadata"]["business_name"], "A厂区视觉识别")
|
|
self.assertEqual(rendered["metadata"]["instance_names"], ["cam1"])
|
|
self.assertEqual(rendered["metadata"]["instance_display_names"], ["东门入口"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|