[PATCH net-next v7 1/3] selftests: net: py: support cmd verifying expected failure

[Willem de Bruijn <willemdebruijn.kernel@gmail.com>]

From: Willem de Bruijn <willemb@google.com>

Support negative tests, where cmd raises an exception if the command
succeeded.

Add optional argument expect_fail to cmd and bkg. Where fail fails the
test on unexpected error, expect_fail fails it on unexpected success.

Both fail on negative return code. Python subprocess may set a
negative return code on process crash or timeout. Those are never
anticipated failures.

Signed-off-by: Willem de Bruijn <willemb@google.com>

---

In this design fail and expect_fail are two complementary tests.
If both are set to true, an exception would always be raised.

This is contrast to the previous approach, where fail enables
fail_on_unexpected_returncode and expect_fail modifies what that
unexpected_returncode value range is.

---

v6 -> v7
  - convert from 'verify_failed' value for fail, to separate variable

v4 -> v5
  - initial version of standalone patch
---
 tools/testing/selftests/net/lib/py/utils.py | 39 +++++++++++++++------
 1 file changed, 29 insertions(+), 10 deletions(-)

diff --git a/tools/testing/selftests/net/lib/py/utils.py b/tools/testing/selftests/net/lib/py/utils.py
index 6c44a3d2bbf7..b68d4607114d 100644
--- a/tools/testing/selftests/net/lib/py/utils.py
+++ b/tools/testing/selftests/net/lib/py/utils.py
@@ -23,6 +23,10 @@ class CmdExitFailure(Exception):
         self.cmd = cmd_obj
 
 
+class CmdExitZeroFailure(CmdExitFailure):
+    """ Command succeeded (returned zero exit code), but expected failure. """
+
+
 def fd_read_timeout(fd, timeout):
     rlist, _, _ = select.select([fd], [], [], timeout)
     if rlist:
@@ -39,8 +43,9 @@ class cmd:
 
     Use bkg() instead to run a command in the background.
     """
-    def __init__(self, comm, shell=None, fail=True, ns=None, background=False,
-                 host=None, timeout=5, ksft_ready=None, ksft_wait=None):
+    def __init__(self, comm, shell=None, fail=True, expect_fail=False, ns=None,
+                 background=False, host=None, timeout=5, ksft_ready=None,
+                 ksft_wait=None):
         if ns:
             comm = f'ip netns exec {ns} ' + comm
 
@@ -88,7 +93,8 @@ class cmd:
                     self._process_terminate(terminate=terminate, timeout=1)
                     raise CmdInitFailure("Did not receive ready message", self)
         if not background:
-            self.process(terminate=False, fail=fail, timeout=timeout)
+            self.process(terminate=False, fail=fail, expect_fail=expect_fail,
+                         timeout=timeout)
 
     def _process_terminate(self, terminate, timeout):
         if terminate:
@@ -102,7 +108,7 @@ class cmd:
 
         return stdout, stderr
 
-    def process(self, terminate=True, fail=None, timeout=5):
+    def process(self, terminate=True, fail=None, expect_fail=False, timeout=5):
         if fail is None:
             fail = not terminate
 
@@ -111,10 +117,19 @@ class cmd:
 
         stdout, stderr = self._process_terminate(terminate=terminate,
                                                  timeout=timeout)
-        if self.proc.returncode != 0 and fail:
+
+        # Fail on unexpected test failure if fail.
+        # Fail on unexpected test success if expect_fail.
+        # Fail on negative returncode if either:
+        # Set by subprocess on crash or signal, this is never expected failure.
+        if (self.proc.returncode != 0 and fail or
+            (self.proc.returncode < 0 and expect_fail)):
             if len(stderr) > 0 and stderr[-1] == "\n":
                 stderr = stderr[:-1]
             raise CmdExitFailure("Command failed", self)
+        elif self.proc.returncode == 0 and expect_fail:
+            raise CmdExitZeroFailure("Command succeeded (expected fail)", self)
+
 
     def __repr__(self):
         def str_fmt(name, s):
@@ -157,14 +172,17 @@ class bkg(cmd):
 
         with bkg("my_binary", ksft_wait=5):
     """
-    def __init__(self, comm, shell=None, fail=None, ns=None, host=None,
-                 exit_wait=False, ksft_ready=None, ksft_wait=None):
+    def __init__(self, comm, shell=None, fail=None, expect_fail=None,
+                 ns=None, host=None, exit_wait=False, ksft_ready=None,
+                 ksft_wait=None):
         super().__init__(comm, background=True,
-                         shell=shell, fail=fail, ns=ns, host=host,
-                         ksft_ready=ksft_ready, ksft_wait=ksft_wait)
+                         shell=shell, fail=fail, expect_fail=expect_fail,
+                         ns=ns, host=host, ksft_ready=ksft_ready,
+                         ksft_wait=ksft_wait)
         self.terminate = not exit_wait and not ksft_wait
         self._exit_wait = exit_wait
         self.check_fail = fail
+        self.expect_fail = expect_fail
 
         if shell and self.terminate:
             print("# Warning: combining shell and terminate is risky!")
@@ -179,7 +197,8 @@ class bkg(cmd):
         # since forcing termination silences failures with fail=None
         if self.proc.poll() is None:
             terminate = terminate or (self._exit_wait and ex_type is not None)
-        return self.process(terminate=terminate, fail=self.check_fail)
+        return self.process(terminate=terminate, fail=self.check_fail,
+                            expect_fail=self.expect_fail)
 
 
 GLOBAL_DEFER_QUEUE = []
-- 
2.54.0.545.g6539524ca2-goog