Merge pull request #1166 from kernelkit/infamy

Mix of test framework fixes and improvements

Author: troglobit

test/infamy/env.py

@@ -117,13 +117,13 @@ def get_password(self, node):
         return self.ptop.get_password(node)
 
     def is_reachable(self, node, port):
-       ip = neigh.ll6ping(port)
-       if not ip:
-           return False
+        ip = neigh.ll6ping(port)
+        if not ip:
+            return False
 
-       return util.is_reachable(ip, self, self.get_password(node))
+        return util.is_reachable(ip, self, self.get_password(node))
 
-    def attach(self, node, port="mgmt", protocol=None, test_reset=True, username = None, password = None):
+    def attach(self, node, port="mgmt", protocol=None, test_reset=True, username=None, password=None):
         """Attach to node on port using protocol."""
 
         name = node
@@ -133,7 +133,6 @@ def attach(self, node, port="mgmt", protocol=None, test_reset=True, username = N
         else:
             mapping = None
 
-
         # Precedence:
         # 1. Caller specifies `protocol`
         # 2. User specifies `-t` when executing test
@@ -159,8 +158,10 @@ def attach(self, node, port="mgmt", protocol=None, test_reset=True, username = N
 
         if protocol == "netconf":
             dev = netconf.Device(name,
-                                 location=netconf.Location(cport, mgmtip,
-                                                           username,password),
+                                 location=netconf.Location(cport,
+                                                           mgmtip,
+                                                           username,
+                                                           password),
                                  mapping=mapping,
                                  yangdir=self.args.yangdir)
             if test_reset:

test/infamy/restconf.py

@@ -26,16 +26,14 @@ class Location:
 
 def xpath_to_uri(xpath, extra=None):
     """Convert xpath to HTTP URI"""
-    # If the xpath has a
     pattern = r'\[(.*?)=["\'](.*?)["\']\]'
     matches = re.findall(pattern, xpath)
 
+    uri_path = xpath
     if matches:
         for key, value in matches:
             # replace [key=value] with =value
-            uri_path = re.sub(rf'\[{re.escape(key)}=["\']{re.escape(value)}["\']\]', f'={value}', xpath)
-    else:
-        uri_path = xpath
+            uri_path = re.sub(rf'\[{re.escape(key)}=["\']{re.escape(value)}["\']\]', f'={value}', uri_path)
 
     # Append extra if provided
     if extra is not None:

test/infamy/tap.py

@@ -30,7 +30,7 @@ def __enter__(self):
         self.out.flush()
         return self
 
-    def __exit__(self, _, e, __):
+    def __exit__(self, t, e, tb):
         now = datetime.datetime.now().strftime("%F %T")
         self.out.write(f"# Exiting ({now})\n")
         self.out.flush()
@@ -40,12 +40,16 @@ def __exit__(self, _, e, __):
         if not e:
             self._not_ok("Missing explicit test result\n")
         else:
-            if type(e) in (TestPass, TestSkip):
+            if t in (TestPass, TestSkip):
                 self.out.write(f"{self.steps}..{self.steps}\n")
                 self.out.flush()
                 raise SystemExit(0)
-
-            traceback.print_exception(e, file=self.commenter)
+            if t is AssertionError:
+                traceback.print_tb(tb, file=self.commenter)
+                self.out.write(f"{str(e)}\n")
+                self.out.flush()
+            else:
+                traceback.print_exception(e, file=self.commenter)
 
             if type(e) is subprocess.CalledProcessError:
                 print("Failing subprocess stdout:\n", e.stdout)
@@ -95,7 +99,10 @@ def succeed(self):
     def skip(self):
         raise TestSkip()
 
-    def fail(self):
+    def fail(self, message=None):
+        if message:
+            self.out.write(f"# {message}\n")
+            self.out.flush()
         raise TestFail()