diff --git a/CHANGELOG.md b/CHANGELOG.md
index d99790c..d21307d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -70,6 +70,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
`nc:operation="delete"` elements (keyed list entries delete by their key
leaf, resolved against the running config), additions use the default merge
operation, and attribute-level changes raise `InvalidConfigError`.
+- gNMI-style JSON remediation rendering (#287):
+ `WorkflowRemediation.remediation_json()` (and
+ `hier_config.formats.hconfig_to_gnmi_json()`) render a remediation between
+ `HConfig.from_json()` trees as a gNMI-SetRequest-style structure — added
+ and changed values render into an `update` object (modified keyed list
+ entries keep their identity leaf), negations become xpath-ish `delete`
+ paths with `[key=value]` selectors resolved against the running config,
+ and attribute-level changes raise `InvalidConfigError`.
### Fixed
diff --git a/docs/dev/architecture.md b/docs/dev/architecture.md
index 31380d7..a0cadba 100644
--- a/docs/dev/architecture.md
+++ b/docs/dev/architecture.md
@@ -149,8 +149,9 @@ The formats module maps JSON (e.g. OpenConfig) and XML (e.g. NETCONF payloads) o
- `hconfig_from_json` / `hconfig_to_json` — invertible JSON mapping (keyed lists identified via `list_keys`).
- `hconfig_from_xml` / `hconfig_to_xml` — invertible XML mapping (attributes and text content become specially-encoded leaves).
- `hconfig_to_netconf_xml` — renders a remediation between `from_xml` trees as a NETCONF `edit-config` payload (deletions become `nc:operation="delete"` elements).
+- `hconfig_to_gnmi_json` — renders a remediation between `from_json` trees as a gNMI-SetRequest-style dict (additions render into an `update` object, deletions become xpath-ish paths with `[key=value]` selectors).
-These are exposed on `HConfig` as `from_json` / `from_xml` / `to_json` / `to_xml`, and on `WorkflowRemediation` as `remediation_netconf_xml()`. See [Loading Configurations](../user/loading-configs.md) for the mapping rules.
+These are exposed on `HConfig` as `from_json` / `from_xml` / `to_json` / `to_xml`, and on `WorkflowRemediation` as `remediation_netconf_xml()` / `remediation_json()`. See [Loading Configurations](../user/loading-configs.md) for the mapping rules.
---
diff --git a/docs/user/remediation-workflows.md b/docs/user/remediation-workflows.md
index c749adb..7dde690 100644
--- a/docs/user/remediation-workflows.md
+++ b/docs/user/remediation-workflows.md
@@ -202,6 +202,20 @@ payload = wfr.remediation_netconf_xml()
Deletions become elements with `nc:operation="delete"`; additions use the NETCONF default merge operation. Keyed list-entry deletions are expressed by their key leaf, resolved against the running config — pass `list_keys=` if your data does not use the default `name`/`id` keys. Attribute-level changes cannot be expressed as NETCONF operations and raise `InvalidConfigError`.
+## gNMI-style JSON remediation payloads
+
+When both configurations were built with [`HConfig.from_json()`](loading-configs.md#structured-formats-json-and-xml), the remediation can be rendered as a gNMI-SetRequest-style dict of update and delete sets:
+
+```python
+result = wfr.remediation_json()
+# {
+# "update": {"system": {"config": {"hostname": "new"}}},
+# "delete": ["interfaces/interface[name=eth1]"],
+# }
+```
+
+Added and changed values render into the `update` object using the same JSON mapping as `to_json()` (a modified keyed list entry keeps its identity leaf, so the update stays valid OpenConfig). Deletions become xpath-ish paths: keyed list entries get a `[key=value]` selector resolved against the running config — pass `list_keys=` if your data does not use the default `name`/`id` keys — while scalar leaves delete by their bare path (e.g. `system/config/hostname`). Backslashes and `]` inside selector values are escaped with a backslash. Element names themselves are not escaped, so keys containing `/` or `[` produce ambiguous paths.
+
## Next steps
- [Working with Tags](tags.md) — filter the remediation for phased deployment.
diff --git a/hier_config/formats.py b/hier_config/formats.py
index d5d7a42..10a2edf 100644
--- a/hier_config/formats.py
+++ b/hier_config/formats.py
@@ -38,6 +38,12 @@
``nc:operation="delete"`` elements; additions use the default merge
operation). Attribute-level changes cannot be expressed as NETCONF
operations and raise ``InvalidConfigError``.
+
+Remediation between ``hconfig_from_json`` trees can be rendered as a
+gNMI-SetRequest-style structure via ``hconfig_to_gnmi_json`` (deletions
+become xpath-ish paths with ``[key=value]`` selectors for keyed list
+entries; additions render into an ``update`` object using the JSON
+mapping above).
"""
from __future__ import annotations
@@ -45,7 +51,7 @@
import xml.etree.ElementTree as ET # ruff:ignore[suspicious-xml-etree-import]
from collections import Counter
from json import JSONDecodeError, dumps, loads
-from typing import TYPE_CHECKING, Any, TypeAlias, cast
+from typing import TYPE_CHECKING, Any, NamedTuple, TypeAlias, TypedDict, cast
from .exceptions import InvalidConfigError
from .registry import resolve_driver
@@ -66,6 +72,13 @@
)
+class GnmiRemediation(TypedDict):
+ """gNMI-SetRequest-style remediation: an update tree and delete paths."""
+
+ update: dict[str, JsonValue]
+ delete: list[str]
+
+
def hconfig_from_json(
platform_or_driver: Platform | str | HConfigDriverBase,
data: str | dict[str, Any],
@@ -387,15 +400,153 @@ def _netconf_delete_element(
if len(words) == 1:
return element
# A keyed list entry (branch in the running config) deletes by key leaf.
- if (
- running_parent is not None
- and (running_entry := running_parent.get_child(equals=positive_text))
- is not None
- and running_entry.children
- ):
- for key in list_keys:
- if running_entry.get_child(equals=f"{key} {words[1]}") is not None:
- ET.SubElement(element, key).text = _xml_text(words[1])
- return element
+ key = _running_entry_key(running_parent, positive_text, words[1], list_keys)
+ if key is not None:
+ ET.SubElement(element, key).text = _xml_text(words[1])
+ return element
element.text = _xml_text(words[1])
return element
+
+
+def _matching_list_key(
+ entry: HConfigBase,
+ raw_value: str,
+ list_keys: tuple[str, ...],
+) -> str | None:
+ for key in list_keys:
+ if entry.get_child(equals=f"{key} {raw_value}") is not None:
+ return key
+ return None
+
+
+def _running_entry_key(
+ running_parent: HConfigBase | None,
+ positive_text: str,
+ raw_value: str,
+ list_keys: tuple[str, ...],
+) -> str | None:
+ """Key leaf identifying `positive_text` as a keyed list entry, if any."""
+ if running_parent is None:
+ return None
+ running_entry = running_parent.get_child(equals=positive_text)
+ if running_entry is None or not running_entry.children:
+ return None
+ return _matching_list_key(running_entry, raw_value, list_keys)
+
+
+def hconfig_to_gnmi_json(
+ remediation: HConfig,
+ *,
+ running: HConfig | None = None,
+ list_keys: tuple[str, ...] | None = None,
+) -> GnmiRemediation:
+ """Render a remediation between `hconfig_from_json` trees as gNMI-style sets.
+
+ Negated nodes become xpath-ish delete paths; everything else renders
+ into the `update` object via the JSON mapping. When `running` is given,
+ deletions of keyed list entries get `[key=value]` selectors (keys found
+ via `list_keys`); without it, deletions fall back to bare leaf paths.
+ """
+ result: GnmiRemediation = {"update": {}, "delete": []}
+ context = _GnmiContext(
+ delete=result["delete"],
+ negation_prefix=remediation.driver.negation_prefix,
+ list_keys=list_keys or DEFAULT_LIST_KEYS,
+ )
+ _gnmi_into(remediation, result["update"], (), running, context)
+ return result
+
+
+class _GnmiContext(NamedTuple):
+ delete: list[str]
+ negation_prefix: str
+ list_keys: tuple[str, ...]
+
+
+def _gnmi_into(
+ node: HConfigBase,
+ update: dict[str, JsonValue],
+ path: tuple[str, ...],
+ running_node: HConfigBase | None,
+ context: _GnmiContext,
+) -> None:
+ for child in node.children:
+ if child.text.startswith(context.negation_prefix):
+ # A negated child is resolved against the parent's running node.
+ context.delete.append(
+ _gnmi_delete_path(
+ path,
+ child.text.removeprefix(context.negation_prefix),
+ running_node,
+ context.list_keys,
+ )
+ )
+ continue
+ words = child.text.split(maxsplit=1)
+ if not child.children:
+ value: JsonValue = _leaf_value(words[1]) if len(words) > 1 else {}
+ _store_json_member(update, words[0], value, force_list=False)
+ continue
+ running_child = (
+ running_node.get_child(equals=child.text) if running_node else None
+ )
+ segment = words[0]
+ key_name: str | None = None
+ if len(words) > 1:
+ key_name = _gnmi_identity_key(
+ child, running_child, words[1], context.list_keys
+ )
+ segment = (
+ f"{words[0]}[{key_name or context.list_keys[0]}"
+ f"={_gnmi_selector_value(words[1])}]"
+ )
+ child_update: dict[str, JsonValue] = {}
+ _gnmi_into(child, child_update, (*path, segment), running_child, context)
+ if not child_update:
+ # The branch contained only deletions.
+ continue
+ if key_name is not None and key_name not in child_update:
+ child_update = {key_name: _leaf_value(words[1]), **child_update}
+ _store_json_member(update, words[0], child_update, force_list=len(words) > 1)
+
+
+def _gnmi_identity_key(
+ entry: HConfigChild,
+ running_entry: HConfigBase | None,
+ raw_value: str,
+ list_keys: tuple[str, ...],
+) -> str | None:
+ for source in (entry, running_entry):
+ if source is None:
+ continue
+ key = _matching_list_key(source, raw_value, list_keys)
+ if key is not None:
+ return key
+ return None
+
+
+def _gnmi_selector_value(raw: str) -> str:
+ return _xml_text(raw).replace("\\", "\\\\").replace("]", "\\]")
+
+
+def _gnmi_delete_path(
+ parent_path: tuple[str, ...],
+ positive_text: str,
+ running_parent: HConfigBase | None,
+ list_keys: tuple[str, ...],
+) -> str:
+ words = positive_text.split(maxsplit=1)
+ if words[0].startswith("@"):
+ message = (
+ "Attribute changes cannot be expressed as gNMI delete paths:"
+ f" {positive_text!r}"
+ )
+ raise InvalidConfigError(message)
+ segment = words[0]
+ # A keyed list entry (branch in the running config) deletes by selector;
+ # a scalar leaf deletes by its bare path (the value is dropped).
+ if len(words) > 1:
+ key = _running_entry_key(running_parent, positive_text, words[1], list_keys)
+ if key is not None:
+ segment = f"{words[0]}[{key}={_gnmi_selector_value(words[1])}]"
+ return "/".join((*parent_path, segment))
diff --git a/hier_config/workflows.py b/hier_config/workflows.py
index 08a0da2..d156a59 100644
--- a/hier_config/workflows.py
+++ b/hier_config/workflows.py
@@ -1,10 +1,17 @@
-from collections.abc import Callable, Iterable
+from __future__ import annotations
+
from logging import getLogger
+from typing import TYPE_CHECKING
from .exceptions import IncompatibleDriverError
-from .models import TagRule
from .root import HConfig
+if TYPE_CHECKING:
+ from collections.abc import Callable, Iterable
+
+ from .formats import GnmiRemediation
+ from .models import TagRule
+
logger = getLogger(__name__)
@@ -132,9 +139,7 @@ def remediation_netconf_xml(
Keyed list-entry deletions are expressed by their key leaf, resolved
against the running config via `list_keys`.
"""
- from .formats import (
- hconfig_to_netconf_xml,
- )
+ from .formats import hconfig_to_netconf_xml
return hconfig_to_netconf_xml(
self.remediation_config,
@@ -142,6 +147,27 @@ def remediation_netconf_xml(
list_keys=list_keys,
)
+ def remediation_json(
+ self,
+ *,
+ list_keys: tuple[str, ...] | None = None,
+ ) -> GnmiRemediation:
+ """Render the remediation as a gNMI-SetRequest-style dict.
+
+ Requires running and generated configs built by `HConfig.from_json()`.
+ Returns `{"update": ..., "delete": [...]}` — added/changed values as a
+ JSON tree and deletions as xpath-ish paths. Keyed list-entry deletions
+ get `[key=value]` selectors, resolved against the running config via
+ `list_keys`.
+ """
+ from .formats import hconfig_to_gnmi_json
+
+ return hconfig_to_gnmi_json(
+ self.remediation_config,
+ running=self.running_config,
+ list_keys=list_keys,
+ )
+
def apply_remediation_tag_rules(self, tag_rules: tuple[TagRule, ...]) -> None:
"""Applies tag rules to selectively label parts of the remediation configuration.
diff --git a/tests/unit/test_formats.py b/tests/unit/test_formats.py
index 473efc1..06dc9e9 100644
--- a/tests/unit/test_formats.py
+++ b/tests/unit/test_formats.py
@@ -7,7 +7,7 @@
from hier_config import HConfig, Platform, WorkflowRemediation
from hier_config.exceptions import DuplicateChildError, InvalidConfigError
-from hier_config.formats import hconfig_to_netconf_xml
+from hier_config.formats import hconfig_to_gnmi_json, hconfig_to_netconf_xml
OPENCONFIG_STYLE = {
"system": {
@@ -323,3 +323,192 @@ def test_xml_diff_is_surgical_across_entry_counts() -> None:
line.strip() for line in remediation.to_lines() if "interface" in line
]
assert interfaces_lines == ["interfaces", 'no interface "eth1"']
+
+
+GNMI_RUNNING = {
+ "system": {"config": {"hostname": "old", "location": "hq"}},
+ "interfaces": {
+ "interface": [
+ {"name": "eth0", "config": {"mtu": 9000}},
+ {"name": "eth1", "config": {"mtu": 1500}},
+ ],
+ },
+}
+GNMI_GENERATED = {
+ "system": {"config": {"hostname": "new", "location": "hq"}},
+ "interfaces": {
+ "interface": [
+ {"name": "eth0", "config": {"mtu": 9000}},
+ ],
+ },
+}
+
+
+def test_gnmi_remediation_payload() -> None:
+ """Remediation between from_json trees renders as update/delete sets."""
+ running = HConfig.from_json(Platform.GENERIC, GNMI_RUNNING)
+ generated = HConfig.from_json(Platform.GENERIC, GNMI_GENERATED)
+ workflow = WorkflowRemediation(running, generated)
+ result = workflow.remediation_json()
+
+ assert result["delete"] == [
+ "system/config/hostname",
+ "interfaces/interface[name=eth1]",
+ ]
+ assert result["update"] == {"system": {"config": {"hostname": "new"}}}
+
+
+def test_gnmi_scalar_leaf_delete_prunes_branch() -> None:
+ """A branch containing only deletions must not appear in the update tree."""
+ running = HConfig.from_json(Platform.GENERIC, {"system": {"hostname": "old"}})
+ generated = HConfig.from_json(Platform.GENERIC, {"system": {}})
+ result = WorkflowRemediation(running, generated).remediation_json()
+
+ assert result == {"update": {}, "delete": ["system/hostname"]}
+
+
+def test_gnmi_keyed_entry_delete_custom_list_keys() -> None:
+ running = HConfig.from_json(
+ Platform.GENERIC,
+ {"vlans": {"vlan": [{"vid": 100}, {"vid": 200}]}},
+ list_keys=("vid",),
+ )
+ generated = HConfig.from_json(
+ Platform.GENERIC, {"vlans": {"vlan": [{"vid": 200}]}}, list_keys=("vid",)
+ )
+ result = hconfig_to_gnmi_json(
+ running.remediation(generated), running=running, list_keys=("vid",)
+ )
+
+ assert result == {"update": {}, "delete": ["vlans/vlan[vid=100]"]}
+
+
+def test_gnmi_nested_delete_under_keyed_ancestor() -> None:
+ """An ancestor keyed entry gets a selector resolved against the running config."""
+ running = HConfig.from_json(
+ Platform.GENERIC,
+ {
+ "interfaces": {
+ "interface": [
+ {"name": "eth0", "config": {"mtu": 9000, "description": "uplink"}},
+ ],
+ },
+ },
+ )
+ generated = HConfig.from_json(
+ Platform.GENERIC,
+ {
+ "interfaces": {
+ "interface": [
+ {"name": "eth0", "config": {"description": "uplink"}},
+ ],
+ },
+ },
+ )
+ result = WorkflowRemediation(running, generated).remediation_json()
+
+ assert result == {
+ "update": {},
+ "delete": ["interfaces/interface[name=eth0]/config/mtu"],
+ }
+
+
+def test_gnmi_update_reinjects_identity_leaf() -> None:
+ """A modified keyed entry's update carries its identity leaf and re-ingests."""
+ running = HConfig.from_json(
+ Platform.GENERIC,
+ {"interfaces": {"interface": [{"name": "eth0", "config": {"mtu": 9000}}]}},
+ )
+ generated = HConfig.from_json(
+ Platform.GENERIC,
+ {"interfaces": {"interface": [{"name": "eth0", "config": {"mtu": 1500}}]}},
+ )
+ result = WorkflowRemediation(running, generated).remediation_json()
+
+ assert result["update"] == {
+ "interfaces": {"interface": [{"name": "eth0", "config": {"mtu": 1500}}]},
+ }
+ assert HConfig.from_json(Platform.GENERIC, result["update"]) is not None
+
+
+def test_gnmi_pure_addition_has_empty_delete() -> None:
+ running = HConfig.from_json(Platform.GENERIC, {"system": {}})
+ generated = HConfig.from_json(
+ Platform.GENERIC, {"system": {}, "ntp": {"enabled": True, "port": 123}}
+ )
+ result = WorkflowRemediation(running, generated).remediation_json()
+
+ assert result == {"update": {"ntp": {"enabled": True, "port": 123}}, "delete": []}
+
+
+def test_gnmi_empty_remediation() -> None:
+ running = HConfig.from_json(Platform.GENERIC, GNMI_RUNNING)
+ generated = HConfig.from_json(Platform.GENERIC, GNMI_RUNNING)
+ result = WorkflowRemediation(running, generated).remediation_json()
+
+ assert result == {"update": {}, "delete": []}
+
+
+def test_gnmi_no_running_context_falls_back_to_scalar() -> None:
+ """Without a running config, keyed-entry deletes degrade to bare paths."""
+ running = HConfig.from_json(Platform.GENERIC, GNMI_RUNNING)
+ generated = HConfig.from_json(Platform.GENERIC, GNMI_GENERATED)
+ result = hconfig_to_gnmi_json(running.remediation(generated))
+
+ assert result["delete"] == ["system/config/hostname", "interfaces/interface"]
+ assert result["update"] == {"system": {"config": {"hostname": "new"}}}
+
+
+def test_gnmi_unresolved_identity_falls_back_to_default_key() -> None:
+ """An unresolvable entry key guesses the selector but skips injection.
+
+ A modified keyed entry's remediation subtree lacks its identity leaf, so
+ without a running config the key name cannot be resolved: the delete-path
+ selector falls back to the first `list_keys` name, and no identity leaf is
+ injected into the update (a guessed key would become applied config).
+ """
+ running = HConfig.from_json(
+ Platform.GENERIC,
+ {"interfaces": {"interface": [{"name": "eth0", "config": {"mtu": 9000}}]}},
+ )
+ generated = HConfig.from_json(
+ Platform.GENERIC,
+ {"interfaces": {"interface": [{"name": "eth0", "config": {"mtu": 1500}}]}},
+ )
+ result = hconfig_to_gnmi_json(running.remediation(generated))
+
+ assert result == {
+ "update": {"interfaces": {"interface": [{"config": {"mtu": 1500}}]}},
+ "delete": ["interfaces/interface[name=eth0]/config/mtu"],
+ }
+
+
+def test_gnmi_attribute_negation_raises() -> None:
+ """Attribute removals cannot be expressed as gNMI delete paths."""
+ running = HConfig.from_xml(Platform.GENERIC, '')
+ generated = HConfig.from_xml(Platform.GENERIC, "")
+ remediation = running.remediation(generated)
+ with pytest.raises(InvalidConfigError, match="Attribute"):
+ hconfig_to_gnmi_json(remediation, running=running)
+
+
+def test_gnmi_selector_value_escaping() -> None:
+ r"""Selector values escape `\` and `]` so paths stay parseable."""
+ running = HConfig.from_json(
+ Platform.GENERIC,
+ {
+ "policies": {
+ "policy": [
+ {"name": "a]b\\c", "action": "deny"},
+ {"name": "keep", "action": "permit"},
+ ],
+ },
+ },
+ )
+ generated = HConfig.from_json(
+ Platform.GENERIC,
+ {"policies": {"policy": [{"name": "keep", "action": "permit"}]}},
+ )
+ result = WorkflowRemediation(running, generated).remediation_json()
+
+ assert result["delete"] == ["policies/policy[name=a\\]b\\\\c]"]