Panel I/O: PanelDef.io ports, scoped publish and upload, I/O section in the Panels dialog

backend/tests/api/routes/test_panels.py

@@ -437,6 +437,84 @@ def test_a_locked_dashboard_entitles_a_panel_to_read_and_not_to_publish(
     )
 
 
+PANEL_IO = [
+    {
+        "port": "gpio.4",
+        "message": "demo.doorbell",
+        "direction": "in",
+        "config": {"pull": "up", "debounce_ms": 50},
+    },
+    {"port": "backlight", "message": "demo.backlight", "direction": "out"},
+]
+
+
+def test_a_panels_io_is_saved_and_read_back(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    _panels(
+        client,
+        superuser_token_headers,
+        {"panels": [{"id": "porch_io", "io": PANEL_IO}]},
+    )
+    read = client.get(f"{PREFIX}/porch_io", headers=superuser_token_headers).json()
+    assert read["io"] == [PANEL_IO[0], {**PANEL_IO[1], "config": {}}]
+
+
+def test_a_panels_own_ports_publish_through_a_lock(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    """An ``in`` port is the device's own sensor, which a dashboard lock is not about.
+
+    An ``out`` port is the device following a message, so it reads and does
+    not write — and only an ``in`` port's name opens the artifact upload.
+    """
+    _dashboard_with(client, superuser_token_headers, "panel_io_locked", PANEL_WIDGETS)
+    _publish_settings(
+        client, superuser_token_headers, "panel_io_locked", {"locked": {"value": True}}
+    )
+    _panels(
+        client,
+        superuser_token_headers,
+        {"panels": [{"id": "door", "dashboards": ["panel_io_locked"], "io": PANEL_IO}]},
+    )
+    panel_headers = _pair(client, superuser_token_headers, "door")
+    messages = f"{settings.API_V1_STR}/messages"
+
+    # 404: past the gate, refused by an engine no flow here declares it in.
+    assert (
+        client.post(
+            f"{messages}/demo.doorbell", headers=panel_headers, json={"value": True}
+        ).status_code
+        == 404
+    )
+    assert (
+        client.post(
+            f"{messages}/demo.backlight", headers=panel_headers, json={"value": 50}
+        ).status_code
+        == 403
+    )
+    assert (
+        client.get(
+            f"{messages}/demo.backlight/history", headers=panel_headers
+        ).status_code
+        == 200
+    )
+
+    artifacts = f"{settings.API_V1_STR}/artifacts"
+    for name, status in (
+        ("demo.doorbell", 200),
+        ("demo.backlight", 403),
+        ("demo.unrelated", 403),
+    ):
+        uploaded = client.put(
+            artifacts,
+            headers=panel_headers,
+            params={"name": name, "media_type": "audio/wav", "volatile": True},
+            content=b"RIFF",
+        )
+        assert uploaded.status_code == status, name
+
+
 def test_a_panel_may_read_its_dashboards_own_settings(
     client: TestClient, superuser_token_headers: dict[str, str]
 ) -> None: