test: infamy: Unify {NET,REST}CONF operational data formatting

Replace the existing remove-module-names-pass in the RESTCONF
transport -- which only worked for top-level containers -- with
libyang's standard `.print_dict()` used by the NETCONF transport
implementation.

As a bouns, the implementation of `.get_iface()` is now transport
agnostic.
This commit is contained in:
Tobias Waldekranz
2024-12-02 12:42:28 +01:00
parent fbe18a4ae0
commit 6d18ba4c39
3 changed files with 11 additions and 39 deletions
-12
View File
@@ -15,7 +15,6 @@ import libyang
import lxml
import netconf_client.connect
import netconf_client.ncclient
import infamy.iface as iface
from infamy.transport import Transport,infer_put_dict
from netconf_client.error import RpcError
from . import env
@@ -382,17 +381,6 @@ class Device(Transport):
with open(outdir+"/"+schema["filename"], "w") as f:
f.write(data.schema)
def get_iface(self, name):
"""Fetch target dict for iface and extract param from JSON"""
content = self.get_data(iface.get_xpath(name))
interface = content.get("interfaces", {}).get("interface", None)
if interface is None:
return None
# Restconf (rousette) address by id, otherwise (netopper2) by name
return interface[name]
def delete_xpath(self, xpath):
# Split out the model and the container from xpath'
pattern = r"^/(?P<module>[^:]+):(?P<path>[^/]+)"
+3 -23
View File
@@ -321,19 +321,11 @@ class Device(Transport):
"""Get operational data"""
uri = xpath_to_uri(xpath) if xpath is not None else None
data = self.get_operational(uri, parse)
if parse is False:
return data
if data is None:
return None
if parse and data:
return data.print_dict()
data = json.loads(data.print_mem("json", with_siblings=True,
pretty=False))
for k, v in data.items():
model, container = k.split(":")
break
return {container: v}
return data
def copy(self, source, target):
factory = self.get_datastore(source)
@@ -372,18 +364,6 @@ class Device(Transport):
data = json.loads(data)
return data["ietf-system:system-state"]["clock"]["current-datetime"]
def get_iface(self, iface):
"""Fetch target dict for iface and extract param from JSON"""
xpath = f"/ietf-interfaces:interfaces/interface={iface}"
content = self.get_data(xpath)
interface = content.get("interfaces", {}).get("interface", None)
if interface is None:
return None
# RESTCONF (at least rousette) addresses by id, but NETCONF (at
# least netopper2) addresses by name
return interface[0]
def delete_xpath(self, xpath):
"""Delete XPath from running config"""
path = f"/ds/ietf-datastores:running/{xpath_to_uri(xpath)}"
+8 -4
View File
@@ -1,5 +1,6 @@
from abc import ABC, abstractmethod
from infamy.neigh import ll6ping
import infamy.iface
def infer_put_dict(name, models):
if not models.get("ietf-system"):
@@ -50,10 +51,13 @@ class Transport(ABC):
def call_action(self, xpath):
pass
@abstractmethod
def get_iface(self, iface):
"""Should be common, but is not due to bug in rousette"""
pass
def get_iface(self, name):
"""Fetch target dict for iface and extract param from JSON"""
content = self.get_data(infamy.iface.get_xpath(name))
interfaces = content.get("interfaces", {}).get("interface", {})
# KeyedList does not support `.get()`
return interfaces[name] if name in interfaces else None
def get_mgmt_ip(self):
"""Return managment IP address used for RESTCONF/NETCONF"""