test: add __str__ to Device class for reading device name

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
This commit is contained in:
Joachim Wiberg
2024-09-05 07:04:58 +02:00
parent a69f1c9272
commit f071dacf8b
5 changed files with 38 additions and 8 deletions
+1 -1
View File
@@ -103,7 +103,7 @@ with infamy.Test() as test:
test.fail()
with test.step(f"Verify user {USER} can log in with SSH ..."):
conn = ssh.Device(ssh.Location(address, factory, USER))
conn = ssh.Device("target", ssh.Location(address, factory, USER))
try:
pwd = conn.runsh(f"cat /etc/passwd | grep {USER}").stdout
print(f"Found {pwd.rstrip()}")
+9 -6
View File
@@ -77,6 +77,7 @@ class Env(object):
def attach(self, node, port, protocol=None, test_reset=True):
"""Attach to node on port using protocol."""
name = node
if self.ltop:
mapping = self.ltop.mapping[node]
node, port = self.ltop.xlate(node, port)
@@ -104,19 +105,21 @@ class Env(object):
password = self.get_password(node)
if protocol == "netconf":
dev = netconf.Device(
location=netconf.Location(cport, mgmtip, password),
mapping=mapping,
yangdir=self.args.yangdir)
dev = netconf.Device(name,
location=netconf.Location(cport, mgmtip,
password),
mapping=mapping,
yangdir=self.args.yangdir)
if test_reset:
dev.test_reset()
return dev
if protocol == "ssh":
return ssh.Device(ssh.Location(mgmtip, password))
return ssh.Device(name, ssh.Location(mgmtip, password))
if protocol == "restconf":
dev = restconf.Device(location=restconf.Location(cport,
dev = restconf.Device(name,
location=restconf.Location(cport,
mgmtip,
password),
mapping=mapping,
+9
View File
@@ -18,6 +18,7 @@ import netconf_client.ncclient
import infamy.iface as iface
from infamy.transport import Transport
from netconf_client.error import RpcError
from . import env
def netconf_syn(addr):
@@ -98,11 +99,13 @@ class Location:
class Device(Transport):
def __init__(self,
name: str,
location: Location,
mapping: dict,
yangdir: None | str = None):
print("Testing using NETCONF")
self.name = name
self.location = location
self.mapping = mapping
self.location = location
@@ -118,6 +121,12 @@ class Device(Transport):
self.ly = libyang.Context(yangdir)
self._ly_init(yangdir)
def __str__(self):
nm = f"{self.name}"
if env.ENV.ltop:
nm += f"({env.ENV.ltop.xlate(self.name)})"
return nm + " [NETCONF]"
def _ncc_init(self, location):
ai = socket.getaddrinfo(location.host, location.port,
0, 0, socket.SOL_TCP)
+10
View File
@@ -11,6 +11,7 @@ from requests.auth import HTTPBasicAuth
from urllib3.exceptions import InsecureRequestWarning
from dataclasses import dataclass
from infamy.transport import Transport
from . import env
# We know we have a self-signed certificate, silence warning about it
warnings.simplefilter('ignore', InsecureRequestWarning)
@@ -109,10 +110,13 @@ def restconf_reachable(neigh, password):
class Device(Transport):
def __init__(self,
name: str,
location: Location,
mapping: dict,
yangdir: None | str = None):
print("Testing using RESTCONF")
self.name = name
self.location = location
self.url_base = f"https://[{location.host}]:{location.port}"
self.restconf_url = f"{self.url_base}/restconf"
@@ -129,6 +133,12 @@ class Device(Transport):
self._ly_bootstrap(yangdir)
self._ly_init(yangdir)
def __str__(self):
nm = f"{self.name}"
if env.ENV.ltop:
nm += f"({env.ENV.ltop.xlate(self.name)})"
return nm + " [RESTCONF]"
def get_schemas_list(self):
modules = self.get_operational("/ietf-yang-library:modules-state")
data = modules.print_dict()
+9 -1
View File
@@ -1,6 +1,7 @@
import subprocess
from dataclasses import dataclass
from . import env
@dataclass
@@ -12,9 +13,16 @@ class Location:
class Device(object):
def __init__(self, location: Location):
def __init__(self, name: str, location: Location):
self.name = name
self.location = location
def __str__(self):
nm = f"{self.name}"
if env.ENV.ltop:
nm += f"({env.ENV.ltop.xlate(self.name)})"
return nm + " [SSH]"
def _mangle_subprocess_args(self, args, kwargs):
if not args:
return None