peter/infix
1
0
Fork 0
mirror of https://github.com/kernelkit/infix.git synced 2026-07-22 01:13:00 +02:00
Code Issues Packages Projects Releases Wiki Activity

test: minor, cleanup & pep-8 fixes

Browse Source 
Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
This commit is contained in:
Joachim Wiberg
2024-09-05 07:04:58 +02:00
parent 12b62f7803
commit 5e03b20bb2
12 changed files with 233 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/case/ietf_hardware/usb/test.py
+11 -10
View File
@@ -5,19 +5,19 @@ import copy
import infamy.usb as usb
import time
import infamy.netconf as netconf
from infamy.util import until,wait_boot
from infamy.util import until, wait_boot
with infamy.Test() as test:
with test.step("Initialize"):
env = infamy.Env()
target = env.attach("target", "mgmt")
available=usb.get_usb_ports(target)
available = usb.get_usb_ports(target)
if len(available) < 1:
test.skip()
with test.step("Unlock USB ports"):
components=[]
components = []
for port in available:
component = {
"name": port,
@@ -40,7 +40,7 @@ with infamy.Test() as test:
if len(available) > 1:
with test.step("Lock one port"):
components=[]
components = []
component = {
"name": available[1],
"class": "infix-hardware:usb",
@@ -60,7 +60,7 @@ with infamy.Test() as test:
until(lambda: usb.get_usb_state(target, available[0]) == "unlocked")
with test.step("Lock USB ports"):
components=[]
components = []
for port in available:
component = {
"class": "infix-hardware:usb",
@@ -80,6 +80,7 @@ with infamy.Test() as test:
with test.step("Verify USB ports locked"):
for port in available:
until(lambda: usb.get_usb_state(target, port) == "locked")
with test.step("Remove all hardware configuration"):
for port in available:
target.delete_xpath(f"/ietf-hardware:hardware/component[name='{port}']")
@@ -88,9 +89,8 @@ with infamy.Test() as test:
for port in available:
until(lambda: usb.get_usb_state(target, port) == "locked")
with test.step("Unlock USB ports"):
components=[]
components = []
for port in available:
component = {
"name": port,
@@ -107,7 +107,7 @@ with infamy.Test() as test:
}
})
with test.step("Verify USB ports unlocked"):
with test.step("Verify USB ports are unlocked"):
for port in available:
until(lambda: usb.get_usb_state(target, port) == "unlocked")
@@ -115,11 +115,12 @@ with infamy.Test() as test:
target.startup_override()
target.copy("running", "startup")
target.reboot()
if wait_boot(target) == False:
if not wait_boot(target, env):
test.fail()
target = env.attach("target", "mgmt", test_reset=False)
with test.step("Verify that all ports are unlocked"):
with test.step("Verify USB port remain unlocked after reboot"):
for port in available:
until(lambda: usb.get_usb_state(target, port) == "unlocked")
test.succeed()
test/case/infix_containers/container_basic/test.py
+4 -2
View File
@@ -8,14 +8,16 @@
#
import infamy
from infamy.util import until
from infamy.util import until
def _verify(server):
url = infamy.Furl(f"http://[{server}]:91/index.html")
return url.check("It works")
with infamy.Test() as test:
NAME = "web"
NAME = "web"
with test.step("Set up topology and attach to target DUT"):
env = infamy.Env()
test/case/meta/wait.py
+3 -15
View File
@@ -1,13 +1,11 @@
#!/usr/bin/env python3
import socket
import time
import requests
import infamy, infamy.neigh
import infamy
import infamy.neigh
import infamy.restconf as restconf
import infamy.netconf as netconf
import infamy.netconf as netconf
from requests.auth import HTTPBasicAuth
def ll6ping(node):
neigh = None
@@ -20,16 +18,6 @@ def ll6ping(node):
return None
def netconf_syn(neigh):
try:
ai = socket.getaddrinfo(neigh, 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"{neigh} answers to TCP connections on port 830 (NETCONF)")
return True
except:
return False
with infamy.Test() as test:
with test.step("Initialize"):
test/case/misc/start_from_startup/test.py
+4 -4
View File
@@ -2,7 +2,6 @@
import infamy
from infamy.util import wait_boot
import copy
with infamy.Test() as test:
with test.step("Initialize"):
@@ -17,16 +16,17 @@ with infamy.Test() as test:
})
target.delete_xpath("/ietf-hardware:hardware/component")
target.copy("running", "startup")
with test.step("Reboot and wait for the unit to come back"):
target.startup_override()
target.copy("running", "startup")
target.reboot()
if wait_boot(target) == False:
if not wait_boot(target, env):
test.fail()
target = env.attach("target", "mgmt", test_default = False)
target = env.attach("target", "mgmt", test_default=False)
with test.step("Verify hostname"):
data = target.get_dict("/ietf-system:system/hostname")
assert(data["system"]["hostname"] == "test")
assert data["system"]["hostname"] == "test"
test.succeed()
test/infamy/env.py
+15 -11
View File
@@ -1,5 +1,4 @@
import argparse
import networkx
import os
import pydot
import shlex
@@ -9,18 +8,21 @@ import inspect
from . import neigh, netconf, restconf, ssh, tap, topology
class NullEnv:
def attr(self, _, default=None):
return default
ENV = NullEnv()
class ArgumentParser(argparse.ArgumentParser):
def __init__(self, ltop):
def __init__(self, top):
super().__init__()
self.add_argument("-d", "--debug", default=False, action="store_true")
self.add_argument("-l", "--logical-topology", dest="ltop", default=ltop)
self.add_argument("-l", "--logical-topology", dest="ltop", default=top)
self.add_argument("-p", "--package", default=None)
self.add_argument("-y", "--yangdir", default=None)
self.add_argument("-t", "--transport", default=None)
@@ -80,13 +82,14 @@ class Env(object):
mapping = None
if protocol is None:
if not self.args.transport is None:
protocol=self.args.transport
if self.args.transport is not None:
protocol = self.args.transport
else:
random.seed(f"{sys.argv[0]}-{os.environ.get('PYTHONHASHSEED',0)}")
hseed = os.environ.get('PYTHONHASHSEED', 0)
random.seed(f"{sys.argv[0]}-{hseed}")
protocol = random.choice(["netconf", "restconf"])
password=self.get_password(node)
password = self.get_password(node)
ctrl = self.ptop.get_ctrl()
cport, _ = self.ptop.get_mgmt_link(ctrl, node)
@@ -107,12 +110,13 @@ class Env(object):
if protocol == "ssh":
return ssh.Device(ssh.Location(mgmtip, password))
if protocol == "restconf":
dev = restconf.Device(location=restconf.Location(cport,
mgmtip,
password),
mapping=mapping,
yangdir=self.args.yangdir)
mgmtip,
password),
mapping=mapping,
yangdir=self.args.yangdir)
if test_reset:
dev.test_reset()
return dev
test/infamy/neigh.py
+3 -3
View File
@@ -1,7 +1,7 @@
import json
import re
import subprocess
def ll6ping(ifname, flags=["-w60", "-c1", "-L", "-n"]):
argv = ["ping"] + flags + ["ff02::1%{}".format(ifname)]
@@ -13,6 +13,6 @@ def ll6ping(ifname, flags=["-w60", "-c1", "-L", "-n"]):
except subprocess.CalledProcessError:
return None
m = re.search("^\d+ bytes from ([:0-9a-f]+%\S+):", ping.stdout, re.MULTILINE)
m = re.search(r"^\d+ bytes from ([:0-9a-f]+%\S+):",
ping.stdout, re.MULTILINE)
return m.group(1) if m else None
test/infamy/netconf.py
+17 -4
View File
@@ -71,6 +71,7 @@ class Manager(netconf_client.ncclient.Manager):
self.set_logger_level(logging.DEBUG)
self.logger().addHandler(logging.StreamHandler(sys.stderr))
class NccGetDataReply:
"""Fold in to DataReply class when upstreaming"""
def __init__(self, raw, ele):
@@ -78,12 +79,14 @@ class NccGetDataReply:
self.data_xml = lxml.etree.tostring(self.data_ele)
self.raw_reply = raw
class NccGetSchemaReply:
def __init__(self, raw):
self.ele = lxml.etree.fromstring(raw.xml.decode())
self.ele = self.ele.find("{urn:ietf:params:xml:ns:yang:ietf-netconf-monitoring}data")
self.schema = self.ele.text
@dataclass
class Location:
interface: str
@@ -92,6 +95,7 @@ class Location:
username: str = "admin"
port: int = 830
class Device(Transport):
def __init__(self,
location: Location,
@@ -401,9 +405,18 @@ class Device(Transport):
newd = mod.parse_data_dict(new, no_state=True)
lyd = oldd.diff(newd)
return self.put_config(lyd.print_mem("xml", with_siblings=True, pretty=False))
return self.put_config(lyd.print_mem("xml", with_siblings=True,
pretty=False))
def get_current_time_with_offset(self):
root = self.get_data("/ietf-system:system-state/clock", parse=False).data_ele
current_datetime = root.find('.//{urn:ietf:params:xml:ns:yang:ietf-system}current-datetime').text
return current_datetime
"""
Return current datetime with offset.
This method retrieves the current datetime from the raw data
before it is passed through libyang. This is necessary because
libyang "adjusts" the time for the offset, and we need the
unadjusted time.
"""
data = self.get_data("/ietf-system:system-state/clock", parse=False)
xpath = './/{urn:ietf:params:xml:ns:yang:ietf-system}current-datetime'
return data.data_ele.find(xpath).text
test/infamy/restconf.py
+143 -104
View File
@@ -5,14 +5,11 @@ import os
import sys
import libyang
import re
import urllib.parse
import time
from requests.auth import HTTPBasicAuth
from urllib.parse import quote
from urllib3.exceptions import InsecureRequestWarning
from dataclasses import dataclass
from lxml import etree
from infamy.transport import Transport
# We know we have a self-signed certificate, silence warning about it
@@ -45,108 +42,121 @@ def xpath_to_uri(xpath, extra=None):
return uri_path
# Workaround for bug in requests 2.32.x: https://github.com/psf/requests/issues/6735
def requests_workaround(method, url, json, headers, auth, verify=False, retry=0):
# Create a session
session=requests.Session()
session = requests.Session()
# Prepare the request
request=requests.Request(method, url, json=json, headers=headers, auth=auth)
prepared_request=session.prepare_request(request)
prepared_request.url=prepared_request.url.replace('%25', '%')
response=session.send(prepared_request, verify=verify)
request = requests.Request(method, url, json=json, headers=headers,
auth=auth)
prepared_request = session.prepare_request(request)
prepared_request.url = prepared_request.url.replace('%25', '%')
response = session.send(prepared_request, verify=verify)
try:
response.raise_for_status() # Raise an exception for HTTP errors
# Raise exceptions for HTTP errors
response.raise_for_status()
except Exception as e:
if e.response.status_code == 502 and retry < 10: # most likely caused by nginx up, but not yet rousette
retry=retry+1
# most likely caused by nginx up, but not yet rousette
if e.response.status_code == 502 and retry < 10:
retry = retry + 1
print(f"{method} {url}: HTTP error 502, retrying({retry})")
time.sleep(1)
response=requests_workaround(method, url, json, headers, auth, verify, retry)
response = requests_workaround(method, url, json, headers, auth,
verify, retry)
else:
raise e
return response
def requests_workaround_put(url, json, headers, auth, verify=False):
return requests_workaround('PUT', url, json, headers, auth, verify=False)
def requests_workaround_delete(url, headers, auth, verify=False):
return requests_workaround('DELETE', url, None, headers, auth, verify=False)
def requests_workaround_post(url, json, headers, auth, verify=False):
return requests_workaround('POST', url, json, headers, auth, verify=False)
def requests_workaround_get(url, headers, auth, verify=False):
return requests_workaround('GET', url, None, headers, auth, verify=False)
def restconf_reachable(neigh, password):
try:
headers={
headers = {
'Content-Type': 'application/yang-data+json',
'Accept': 'application/yang-data+json'
}
url=f"https://[{neigh}]/restconf/data/ietf-system:system/hostname"
auth=HTTPBasicAuth("admin", password)
url = f"https://[{neigh}]/restconf/data/ietf-system:system/hostname"
auth = HTTPBasicAuth("admin", password)
response=requests_workaround_get(url, headers=headers, auth=auth, verify=False)
if response.status_code==200:
response = requests_workaround_get(url, headers=headers, auth=auth,
verify=False)
if response.status_code == 200:
print(f"{neigh} answers to TCP connections on port 443 (RESTCONF)")
return True
except:
return False
return False
class Device(Transport):
def __init__(self,
location: Location,
mapping: dict,
yangdir: None | str = None):
print("Testing using RESTCONF")
self.location=location
self.url_base=f"https://[{location.host}]:{location.port}"
self.restconf_url=f"{self.url_base}/restconf"
self.yang_url=f"{self.url_base}/yang"
self.rpc_url=f"{self.url_base}/restconf/operations"
self.headers={
self.location = location
self.url_base = f"https://[{location.host}]:{location.port}"
self.restconf_url = f"{self.url_base}/restconf"
self.yang_url = f"{self.url_base}/yang"
self.rpc_url = f"{self.url_base}/restconf/operations"
self.headers = {
'Content-Type': 'application/yang-data+json',
'Accept': 'application/yang-data+json'
}
self.auth=HTTPBasicAuth(location.username, location.password)
self.modules={}
self.auth = HTTPBasicAuth(location.username, location.password)
self.modules = {}
self.lyctx=libyang.Context(yangdir)
self.lyctx = libyang.Context(yangdir)
self._ly_bootstrap(yangdir)
self._ly_init(yangdir)
def get_schemas_list(self):
data=self.get_operational("/ietf-yang-library:modules-state").print_dict()
modules = self.get_operational("/ietf-yang-library:modules-state")
data = modules.print_dict()
return data["modules-state"]["module"]
def get_schema(self, name, revision, yangdir):
schema_name=f"{name}@{revision}.yang"
url=f"{self.yang_url}/{schema_name}"
data=self._get_raw(url=url, parse=False)
schema_name = f"{name}@{revision}.yang"
url = f"{self.yang_url}/{schema_name}"
data = self._get_raw(url=url, parse=False)
sys.stdout.write(f"Downloading YANG model {schema_name} ...\r\033[K")
data=data.decode('utf-8')
data = data.decode('utf-8')
with open(f"{yangdir}/{schema_name}", 'w') as json_file:
json_file.write(data)
def schema_exist(self, name, revision, yangdir):
schema_name=f"{name}@{revision}.yang"
schema_path=f"{yangdir}/{schema_name}"
schema_name = f"{name}@{revision}.yang"
schema_path = f"{yangdir}/{schema_name}"
return os.path.exists(schema_path)
def _ly_bootstrap(self, yangdir):
schemas=self.get_schemas_list()
schemas = self.get_schemas_list()
for schema in schemas:
if not self.schema_exist(schema["name"], schema["revision"],yangdir):
if not self.schema_exist(schema["name"], schema["revision"], yangdir):
self.get_schema(schema["name"], schema["revision"], yangdir)
if schema.get("submodule"):
for submodule in schema["submodule"]:
if not self.schema_exist(submodule["name"], submodule["revision"],yangdir):
if not self.schema_exist(submodule["name"], submodule["revision"], yangdir):
self.get_schema(submodule["name"], submodule["revision"], yangdir)
if not any("submodule" in x and schema["name"] in x["submodule"] for x in schemas):
@@ -155,14 +165,11 @@ class Device(Transport):
print("YANG models downloaded.")
def _ly_init(self, yangdir):
lib=self.lyctx.load_module("ietf-yang-library")
ns=libyang.util.c2str(lib.cdata.ns)
for ms in self.modules.values():
if ms["conformance-type"] != "implement":
continue
mod=self.lyctx.load_module(ms["name"])
mod = self.lyctx.load_module(ms["name"])
# TODO: ms["feature"] contains the list of enabled
# features, so ideally we should only enable the supported"
@@ -172,21 +179,24 @@ class Device(Transport):
def _get_raw(self, url, parse=True):
"""Actually send a GET to RESTCONF server"""
response=requests_workaround_get(url, headers=self.headers, auth=self.auth, verify=False)
response.raise_for_status() # Raise an exception for HTTP errors
response = requests_workaround_get(url, headers=self.headers,
auth=self.auth, verify=False)
# Raise exceptions for HTTP errors
response.raise_for_status()
if parse:
data=response.json()
data=self.lyctx.parse_data_mem(json.dumps(data), "json", parse_only=True)
data = response.json()
data = self.lyctx.parse_data_mem(json.dumps(data), "json",
parse_only=True)
return data
else:
return response.content
return response.content
def get_datastore(self, datastore="operational" , path="", parse=True):
def get_datastore(self, datastore="operational", path="", parse=True):
"""Get a datastore"""
dspath=f"/ds/ietf-datastores:{datastore}"
if not path is None:
dspath=f"{dspath}/{path}"
url=f"{self.restconf_url}{dspath}"
dspath = f"/ds/ietf-datastores:{datastore}"
if path is not None:
dspath = f"{dspath}/{path}"
url = f"{self.restconf_url}{dspath}"
return self._get_raw(url, parse)
def get_running(self, path=None):
@@ -203,58 +213,70 @@ class Device(Transport):
def post_datastore(self, datastore, data):
"""Actually send a POST to RESTCONF server"""
url=f"{self.restconf_url}/ds/ietf-datastores:{datastore}"
response=requests_workaround_post(url,
json=data, # Directly pass the dictionary without using json.dumps
headers=self.headers,
auth=self.auth,
verify=False
)
response.raise_for_status() # Raise an exception for HTTP errors
url = f"{self.restconf_url}/ds/ietf-datastores:{datastore}"
# Directly pass the dictionary without using json.dumps
response = requests_workaround_post(url,
json=data,
headers=self.headers,
auth=self.auth,
verify=False)
# Raise exceptions for HTTP errors
response.raise_for_status()
def put_datastore(self, datastore, data):
"""Actually send a PUT to RESTCONF server"""
response=requests_workaround_put(f"{self.restconf_url}/ds/ietf-datastores:{datastore}/",
json=data, # Directly pass the dictionary without using json.dumps
# Directly pass the dictionary without using json.dumps
response = requests_workaround_put(
f"{self.restconf_url}/ds/ietf-datastores:{datastore}/",
json=data,
headers=self.headers,
auth=self.auth,
verify=False
)
response.raise_for_status() # Raise an exception for HTTP errors
# Raise exceptions for HTTP errors
response.raise_for_status()
def get_config_dict(self, modname):
"""Get all configuration for module @modname as dictionary"""
ds=self.get_running(modname)
ds=json.loads(ds.print_mem("json", with_siblings=True, pretty=False))
model, container=modname.split(":")
ds = self.get_running(modname)
ds = json.loads(ds.print_mem("json", with_siblings=True, pretty=False))
model, container = modname.split(":")
for k, v in ds.items():
return {container: v}
def put_config_dict(self, modname, edit):
"""Add @edit to running config and put the whole configuration"""
# This is hacky, refactor when rousette have PATCH support.
running=self.get_running()
mod=self.lyctx.get_module(modname)
for k, v in edit.items():
module=modname+":"+k
data=v
# This is hacky, refactor when rousette have PATCH support.
running = self.get_running()
mod = self.lyctx.get_module(modname)
for k, _ in edit.items():
module = modname + ":" + k
break
# Ugly hack, but this function should be refactored when patch is avaible in rousette anyway.
rundict = json.loads(running.print_mem("json", with_siblings=True, pretty=False))
# Ugly hack, but this function should be refactored when patch
# is available in rousette anyway.
rundict = json.loads(running.print_mem("json", with_siblings=True,
pretty=False))
if rundict.get(module) is None:
rundict[module] = {}
running=self.lyctx.parse_data_mem(json.dumps(rundict), "json", parse_only=True)
running = self.lyctx.parse_data_mem(json.dumps(rundict), "json",
parse_only=True)
change=mod.parse_data_dict(edit, no_state=True, validate=False)
change = mod.parse_data_dict(edit, no_state=True, validate=False)
running.merge_module(change)
return self.put_datastore("running", json.loads(running.print_mem("json", with_siblings=True, pretty=False)))
data = json.loads(running.print_mem("json", with_siblings=True,
pretty=False))
return self.put_datastore("running", data)
def call_rpc(self, rpc):
url=f"{self.rpc_url}/{rpc}"
"""Actually send a POST to RESTCONF server"""
response=requests_workaround_post(
url = f"{self.rpc_url}/{rpc}"
response = requests_workaround_post(
url,
json=None,
headers=self.headers,
@@ -269,62 +291,79 @@ class Device(Transport):
def get_data(self, xpath=None, parse=True):
"""Get operational data"""
uri=xpath_to_uri(xpath) if xpath is not None else None
data=self.get_operational(uri, parse)
if parse==False:
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
data=json.loads(data.print_mem("json", with_siblings=True, pretty=False))
for k,v in data.items():
model, container=k.split(":")
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}
def copy(self, source, target):
factory=self.get_datastore(source)
data=factory.print_mem("json", with_siblings=True, pretty=False)
factory = self.get_datastore(source)
data = factory.print_mem("json", with_siblings=True, pretty=False)
self.put_datastore(target, json.loads(data))
def reboot(self):
self.call_rpc("ietf-system:system-restart")
def call_action(self, xpath):
path=xpath_to_uri(xpath)
url=f"{self.restconf_url}/data{path}"
response=requests_workaround_post(
path = xpath_to_uri(xpath)
url = f"{self.restconf_url}/data{path}"
response = requests_workaround_post(
url,
json=None,
headers=self.headers,
auth=self.auth,
verify=False)
response.raise_for_status() # Raise an exception for HTTP errors
verify=False
)
# Raise exceptions for HTTP errors
response.raise_for_status()
return response.content
def get_current_time_with_offset(self):
"""Parse the time in the raw reply, before it has been passed through libyang, there all offset is lost"""
data=self.get_data("/ietf-system:system-state/clock", parse=False)
data=json.loads(data)
"""
Return current datetime with offset.
This method retrieves the current datetime from the raw data
before it is passed through libyang. This is necessary because
libyang "adjusts" the time for the offset, and we need the
unadjusted time.
"""
data = self.get_data("/ietf-system:system-state/clock", parse=False)
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"""
content=self.get_data(f"/ietf-interfaces:interfaces/interface={iface}")
interface=content.get("interfaces", {}).get("interface", None)
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 (rousette) address by id and netconf (netopper2) address by name
# 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)}"
url=f"{self.restconf_url}{path}"
response=requests_workaround_delete(url, headers=self.headers, auth=self.auth, verify=False)
response.raise_for_status() # Raise an exception for HTTP errors
path = f"/ds/ietf-datastores:running/{xpath_to_uri(xpath)}"
url = f"{self.restconf_url}{path}"
response = requests_workaround_delete(url, headers=self.headers,
auth=self.auth, verify=False)
# Raise exceptions for HTTP errors
response.raise_for_status()
return True
test/infamy/ssh.py
+1 -1
View File
@@ -20,7 +20,7 @@ class Device(object):
return None
args = list(args)
if type(args[0]) == str:
if type(args[0]) is str:
if kwargs.get("shell"):
args[0] = ["/bin/sh", "-c", args[0]]
kwargs["shell"] = False
test/infamy/tap.py
+7
View File
@@ -85,16 +85,23 @@ class Test:
def fail(self):
raise TestFail()
class TestResult(Exception):
pass
class TestPass(TestResult):
pass
class TestFail(TestResult):
pass
class TestSkip(TestResult):
pass
class CommentWriter:
def __init__(self, f):
self.f = f
test/infamy/topology.py
+12 -6
View File
@@ -1,8 +1,9 @@
import networkx as nx
from networkx.algorithms import isomorphism
def _qstrip(text):
if text == None:
if text is None:
return None
if text.startswith("\"") and text.endswith("\""):
@@ -10,6 +11,7 @@ def _qstrip(text):
return text
def map_edges(les, pes):
acc = []
les = sorted(list(les.values()), key=lambda x: x.get("kind", ""), reverse=True)
@@ -23,11 +25,14 @@ def map_edges(les, pes):
return acc
def match_node(pn, ln):
return pn.get("kind") == ln.get("kind")
def match_edge(pes, les):
return map_edges(les, pes) != None
return map_edges(les, pes) is not None
class Topology:
def __init__(self, dotg):
@@ -47,7 +52,7 @@ class Topology:
sn, sp = e.get_source().split(":")
dn, dp = e.get_destination().split(":")
attrs = { _qstrip(k): _qstrip(v) for k, v in e.get_attributes().items() }
attrs = {_qstrip(k): _qstrip(v) for k, v in e.get_attributes().items()}
attrs[sn] = sp
attrs[dn] = dp
self.g.add_edge(sn, dn, **attrs)
@@ -95,7 +100,7 @@ class Topology:
return True
def xlate(self, lnode, lport=None):
assert(self.mapping)
assert self.mapping
if lnode not in self.mapping:
return None
@@ -121,7 +126,7 @@ class Topology:
def get_password(self, node):
n = self.dotg.get_node(node)
b = n[0] if n else {}
password=b.get("password")
password = b.get("password")
return qstrip(password) if password is not None else "admin"
@@ -138,7 +143,7 @@ class Topology:
def get_ctrl(self):
ns = self.get_nodes(lambda _, attrs: attrs.get("kind") == "controller")
assert(len(ns) == 1)
assert len(ns) == 1
return ns[0]
def get_infixen(self):
@@ -147,6 +152,7 @@ class Topology:
def get_attr(self, name, default=None):
return _qstrip(self.dotg.get_attributes().get(name, default))
# Support calling this script like so...
#
# python3 topology.py <physical> <logical>
test/infamy/transport.py
+13 -2
View File
@@ -1,39 +1,50 @@
from abc import ABC, abstractmethod
from infamy.neigh import ll6ping
class Transport(ABC):
"""Common functions for NETCONF/RESTCONF"""
@abstractmethod
def get_data(self, xpath=None, parse=True):
pass
@abstractmethod
def get_config_dict(self, modname):
pass
@abstractmethod
def put_config_dict(self, modname, edit):
pass
@abstractmethod
def get_dict(self, xpath=None):
pass
@abstractmethod
def delete_xpath(self, xpath):
pass
@abstractmethod
def copy(self, source, target):
pass
@abstractmethod
def reboot(self):
pass
@abstractmethod
def get_current_time_with_offset(self):
# This is needed since libyang is too nice and removes the original offset
"""Needed since libyang is too nice and removes the original offset"""
pass
@abstractmethod
def call_action(self, xpath):
pass
@abstractmethod
def get_iface(self, iface): # Should be common, but is not due to bug in rousette
def get_iface(self, iface):
"""Should be common, but is not due to bug in rousette"""
pass
def get_mgmt_ip(self):