diff --git a/test/infamy/env.py b/test/infamy/env.py
index 368d9160..be6abd9a 100644
--- a/test/infamy/env.py
+++ b/test/infamy/env.py
@@ -35,7 +35,7 @@ class Env(object):
if not self.ltop.map_to(self.ptop):
raise tap.TestSkip()
- def attach(self, node, port):
+ def attach(self, node, port, factory_default=True):
if self.ltop:
mapping = self.ltop.mapping[node]
node, port = self.ltop.xlate(node, port)
@@ -55,7 +55,8 @@ class Env(object):
raise Exception(f"Failed, cannot find mgmt IP for {node}")
return netconf.Device(
- location=netconf.Location(mgmtip, password),
+ location=netconf.Location(cport, mgmtip, password),
mapping=mapping,
- yangdir=self.args.yangdir
+ yangdir=self.args.yangdir,
+ factory_default=factory_default
)
diff --git a/test/infamy/netconf.py b/test/infamy/netconf.py
index db36e30f..30766510 100644
--- a/test/infamy/netconf.py
+++ b/test/infamy/netconf.py
@@ -13,9 +13,22 @@ import libyang
import lxml
import netconf_client.connect
import netconf_client.ncclient
+from infamy.neigh import ll6ping
from netconf_client.error import RpcError
+def netconf_syn(addr):
+ try:
+ ai = socket.getaddrinfo(addr, 830, 0, 0, socket.SOL_TCP)
+ sock = socket.socket(ai[0][0], ai[0][1], 0)
+ sock.connect(ai[0][4])
+ sock.close()
+ print(f"{addr} answers to TCP connections on port 830 (NETCONF)")
+ return True
+ except:
+ return False
+
+
modinfo_fields = ("identifier", "version", "format", "namespace")
ModInfoTuple = namedtuple("ModInfoTuple", modinfo_fields)
class ModInfo(ModInfoTuple):
@@ -63,6 +76,7 @@ class NccGetSchemaReply:
@dataclass
class Location:
+ interface: str
host: str
password: str
username: str = "admin"
@@ -72,9 +86,11 @@ class Device(object):
def __init__(self,
location: Location,
mapping: dict,
- yangdir: None | str = None):
+ yangdir: None | str = None,
+ factory_default = True):
self.mapping = mapping
+ self.location = location
self.ly = libyang.Context(yangdir)
self._ncc_init(location)
self.modules = {}
@@ -83,7 +99,21 @@ class Device(object):
del self.ly
self.ly = libyang.Context(yangdir)
self._ly_init(yangdir)
- self.ncc.dispatch('')
+ if factory_default:
+ self.ncc.dispatch('')
+
+ def get_mgmt_ip(self):
+ return location.host
+
+ def get_mgmt_iface(self):
+ return self.location.interface
+
+ def reachable(self):
+ neigh=ll6ping(self.location.interface, flags=["-w1", "-c1", "-L", "-n"])
+ if neigh:
+ return True
+
+ return False
def _ncc_init(self, location):
ai = socket.getaddrinfo(location.host, location.port,
@@ -171,6 +201,24 @@ class Device(object):
pieces.append("")
return self._ncc_make_rpc("".join(pieces), msg_id=msg_id)
+ def copy(self, source, target):
+ cmd=f'''
+
+ <{target}/>
+
+
+ <{source}/>
+
+ '''
+ self.ncc._send_rpc(self._ncc_make_rpc(cmd))
+
+ def reboot(self):
+ cmd=''
+ return self.call_dict("ietf-system",
+ {
+ "system-restart" : {}
+ })
+
def _get(self, xpath, getter, as_xml=False):
# Figure out which modules we are referencing
mods = self._modules_in_xpath(xpath)
@@ -303,4 +351,3 @@ class Device(object):
with open(outdir+"/"+schema["filename"], "w") as f:
f.write(data.schema)
-