71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
import importlib.util
|
|
import pathlib
|
|
import sys
|
|
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_renders_profile_and_overlay(self):
|
|
module = load_module()
|
|
template = {
|
|
"name": "pipeline",
|
|
"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.profile_instances(profile, "pipeline"),
|
|
"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")
|
|
node_override = rendered["instances"][0]["override"]["nodes"]["face_recog"]
|
|
self.assertTrue(node_override["debug"]["enabled"])
|
|
self.assertTrue(node_override["debug"]["log_matches"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|