pytest for missing and obsolete user management topics#3366
Conversation
There was a problem hiding this comment.
Pull request overview
Adds a pytest that cross-checks the topics in UpdateConfig.valid_topic against the ACLs in default-dynamic-security.json / role-templates.json, to detect missing topics in (or obsolete topics in) the mosquitto user management. Along the way, several obsolete topics are removed from update_config.py, setdata.py, general.py, and the dynsec defaults, and a few currently-used topics that the new test would otherwise flag are added.
Changes:
- New
missing_role_topics_test.pycomparing valid topics with role ACLs, with allow-lists for internal/non-persistent topics. - Remove obsolete topics (
external_buttons_hw,mqtt_bridge,notifications/*,internal_chargepoint/global_config,chargemode_config/pv_charging/phases_to_use) from config/validation/data classes and the dynsec default. - Add missing IO/MQTT/EP topics to
UpdateConfig.valid_topicand fix theonly_local_charge_pointsregex to be anchored with$; correct twoopenWB/general/internal_chargepoint/...ACL paths toopenWB/internal_chargepoint/....
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/helpermodules/mosquitto_dynsec/missing_role_topics_test.py | New pytest comparing UpdateConfig.valid_topic to mosquitto role ACLs. |
| packages/helpermodules/update_config.py | Removes obsolete topics, adds missing IO/MQTT/EP topics, fixes a missing $ anchor. |
| packages/helpermodules/setdata.py | Removes validation branches for the now-obsolete general/notification/external_buttons_hw/mqtt_bridge topics. |
| packages/control/general.py | Drops the obsolete external_buttons_hw field from GeneralData. |
| data/config/mosquitto/public/default-dynamic-security.json | Corrects two internal_chargepoint ACL topics and removes the obsolete pv_charging/phases_to_use ACL. |
Comments suppressed due to low confidence (1)
packages/helpermodules/mosquitto_dynsec/missing_role_topics_test.py:141
valid_topics = UpdateConfig.valid_topicis a reference to the class-level list, not a copy. The subsequent in-place reassignmentvalid_topics[i] = valid_topicmutates the sharedUpdateConfig.valid_topic(stripping the^...$anchors and rewriting character classes), which can affect any other test or code that importsUpdateConfigin the same pytest session. Take a copy (e.g.list(UpdateConfig.valid_topic)) before mutating.
valid_topics = UpdateConfig.valid_topic
for i, valid_topic in enumerate(valid_topics):
if valid_topic.startswith("^"):
valid_topic = valid_topic[1:]
if valid_topic.endswith("$"):
valid_topic = valid_topic[:-1]
valid_topic = re.sub(r"\[0-9]\+", "+", valid_topic)
valid_topic = re.sub(r"\[0-1]", "+", valid_topic)
valid_topic = re.sub(r"\[0-9]", "+", valid_topic)
valid_topics[i] = valid_topic
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
|
||
| from helpermodules.update_config import UpdateConfig | ||
|
|
||
|
|
||
| DEFAULT_DYNAMIC_SECURITY_PATH = "/var/www/html/openWB/data/config/mosquitto/public/default-dynamic-security.json" | ||
| ROLE_TEMPLATES_PATH = "/var/www/html/openWB/data/config/mosquitto/public/role-templates.json" |
| def parse_acl_topic_to_regex(acl_topic): | ||
| if "openWB/set" not in acl["topic"] and "#" != acl["topic"] and "openWB/#" != acl["topic"]: | ||
| topic = acl["topic"] | ||
| # convert mqtt topic pattern to regex pattern | ||
| topic_regex = topic.replace("+", "[^/]+").replace("#", ".*").replace("<id>", "[^/]+") | ||
| return topic_regex | ||
| return None |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.
Comments suppressed due to low confidence (1)
packages/helpermodules/mosquitto_dynsec/missing_role_topics_test.py:168
parse_acl_topic_to_regex(acl)is called with the entire ACL dict instead of the topic string. Inside the function,acl_topic.replace(...)will raise anAttributeErrorbecause dicts have noreplacemethod. The call should passacl["topic"]instead ofacl. This bug occurs in both loops (around lines 158 and 166).
for acl in role.get("acls", []):
if "topic" in acl:
topic_regex = parse_acl_topic_to_regex(acl)
if topic_regex:
role_topics.append(topic_regex)
with open(ROLE_TEMPLATES_PATH, "r") as f:
roles = json.load(f)
for role in roles:
for acl in role.get("acls", []):
if "topic" in acl:
topic_regex = parse_acl_topic_to_regex(acl)
if topic_regex:
role_topics.append(topic_regex)
| valid_topics = UpdateConfig.valid_topic | ||
| for i, valid_topic in enumerate(valid_topics): | ||
| if valid_topic.startswith("^"): | ||
| valid_topic = valid_topic[1:] | ||
| if valid_topic.endswith("$"): | ||
| valid_topic = valid_topic[:-1] | ||
| valid_topic = re.sub(r"\[0-9]\+", "+", valid_topic) | ||
| valid_topic = re.sub(r"\[0-1]", "+", valid_topic) | ||
| valid_topic = re.sub(r"\[0-9]", "+", valid_topic) | ||
| valid_topics[i] = valid_topic |
| 'openWB/general/grid_protection_active', | ||
| 'openWB/general/grid_protection_timestamp', | ||
| 'openWB/general/grid_protection_random_stop', | ||
| 'openWB/general/price_kwh', # obsolet topic |
There was a problem hiding this comment.
Bei Änderungen an den default ACLs bitte immer auch die Rolle mit der Versionsnummer hochsetzen.
pytest, der prüft, ob Topics in der Benutzerverwaltung fehlen oder obsolete Topics in der Benutzerverwaltung enthalten sind