Panels dialog saves a ready I/O port draft on close and keeps an unfinished one; one qualified message-name rule on both sides
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx
This commit is contained in:
@@ -12,6 +12,7 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
import re
|
||||
from enum import Enum
|
||||
from typing import Any
|
||||
|
||||
@@ -320,6 +321,24 @@ class MessageSpec(BaseModel):
|
||||
return hash((self.name, self.port))
|
||||
|
||||
|
||||
#: A message named with its flow: the flow's own name, then the name inside it.
|
||||
#: The one rule for anywhere a name has to arrive qualified; the frontend holds
|
||||
#: a copy (``QUALIFIED_MESSAGE`` in ``deriveEdges.ts``).
|
||||
QUALIFIED_PATTERN = re.compile(r"[a-z][a-z0-9_]*(\.[A-Za-z0-9_]+)+")
|
||||
|
||||
QUALIFIED_RULE = (
|
||||
"Name the message as flow.name: a lowercase flow name, a dot, "
|
||||
"then letters, digits and underscores"
|
||||
)
|
||||
|
||||
|
||||
def validate_qualified(value: str) -> str:
|
||||
"""A qualified message name, or a ``ValueError`` saying what one looks like."""
|
||||
if not QUALIFIED_PATTERN.fullmatch(value):
|
||||
raise ValueError(QUALIFIED_RULE)
|
||||
return value
|
||||
|
||||
|
||||
def qualify(flow: str, name: str) -> str:
|
||||
"""Resolve a message name against its flow namespace."""
|
||||
if not name:
|
||||
|
||||
@@ -21,6 +21,7 @@ from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
from fluksio.core.config import settings
|
||||
from fluksio.flow.dashboards import DashboardDef, DashboardNotFound, DashboardStore
|
||||
from fluksio.flow.messages import validate_qualified
|
||||
from fluksio.flow.schemas import _validate_name
|
||||
|
||||
#: Free text the device interprets: ``gpio.4``, ``backlight``, ``mic``.
|
||||
@@ -51,14 +52,7 @@ class PanelPort(BaseModel):
|
||||
raise ValueError("Use lowercase letters, digits, underscores and dots")
|
||||
return value
|
||||
|
||||
@field_validator("message")
|
||||
@classmethod
|
||||
def _check_message(cls, value: str) -> str:
|
||||
flow, _, name = value.partition(".")
|
||||
if not name:
|
||||
raise ValueError("Name the message with its flow, as flow.name")
|
||||
_validate_name(flow)
|
||||
return value
|
||||
_check_message = field_validator("message")(validate_qualified)
|
||||
|
||||
|
||||
class PanelDef(BaseModel):
|
||||
|
||||
@@ -2,7 +2,13 @@ import json
|
||||
|
||||
import pytest
|
||||
|
||||
from fluksio.flow.messages import DType, MessageSpec, flow_of, qualify
|
||||
from fluksio.flow.messages import (
|
||||
DType,
|
||||
MessageSpec,
|
||||
flow_of,
|
||||
qualify,
|
||||
validate_qualified,
|
||||
)
|
||||
|
||||
|
||||
def test_port_defaults_to_last_name_segment():
|
||||
@@ -201,3 +207,14 @@ def test_a_value_that_refers_to_itself_does_not_hang_the_check():
|
||||
loop: dict = {}
|
||||
loop["self"] = loop
|
||||
spec.check(loop)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["home.doorbell", "house.kitchen.temperature"])
|
||||
def test_a_qualified_name_names_its_flow(name):
|
||||
assert validate_qualified(name) == name
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["doorbell", "Home.x", "home.", "home.a-b", "home..x"])
|
||||
def test_anything_else_is_refused_with_the_rule(name):
|
||||
with pytest.raises(ValueError, match="flow.name"):
|
||||
validate_qualified(name)
|
||||
|
||||
Reference in New Issue
Block a user