Merge pull request #64 from LabNConsulting/chopps/watchlog-and-retry-assert-ok

Watchlog and @Retry features

Author: choppsv1

munet/testing/util.py

@@ -59,7 +59,9 @@ def pause_test(desc=""):
     asyncio.run(async_pause_test(desc))
 
 
-def retry(retry_timeout, initial_wait=0, retry_sleep=2, expected=True):
+def retry(
+    retry_timeout, initial_wait=0, retry_sleep=2, expected=True, assert_is_except=True
+):
     """Retry decorated function until it returns None, raises an exception, or timeout.
 
     * `retry_timeout`: Retry for at least this many seconds; after waiting
@@ -68,12 +70,19 @@ def retry(retry_timeout, initial_wait=0, retry_sleep=2, expected=True):
     * `retry_sleep`: The time to sleep between retries.
     * `expected`: if False then the return logic is inverted, except for exceptions,
                   (i.e., a non None ends the retry loop, and returns that value)
+    * `assert_is_except`: If True (the default) then an AssertionError raised by the
+                         wrapped function will be treated as an excpetion. If False then
+                         an assertion raised by the wrapped fucntion is treated as
+                         non-None result it is treated as an exception. This is
+                         important for handling the expected=False case. Exceptions are
+                         always treated as failures even when expected is False.
     """
 
     def _retry(func):
         @functools.wraps(func)
         def func_retry(*args, **kwargs):
             # Allow the wrapped function's args to override the fixtures
+            _assert_is_except = kwargs.pop("assert_is_except", assert_is_except)
             _retry_sleep = float(kwargs.pop("retry_sleep", retry_sleep))
             _retry_timeout = kwargs.pop("retry_timeout", retry_timeout)
             _expected = kwargs.pop("expected", expected)
@@ -90,13 +99,19 @@ def func_retry(*args, **kwargs):
                 seconds_left = (retry_until - datetime.datetime.now()).total_seconds()
                 try:
                     try:
-                        ret = func(*args, seconds_left=seconds_left, **kwargs)
-                    except TypeError as error:
-                        if "seconds_left" not in str(error):
+                        try:
+                            ret = func(*args, seconds_left=seconds_left, **kwargs)
+                        except TypeError as error:
+                            if "seconds_left" not in str(error):
+                                raise
+                            ret = func(*args, **kwargs)
+                    except AssertionError as error:
+                        if _assert_is_except:
                             raise
-                        ret = func(*args, **kwargs)
-
-                    logging.debug("Function returned %s", ret)
+                        logging.info('Function returned assertion: "%s"', error)
+                        ret = error
+                    else:
+                        logging.debug("Function returned %s", ret)
 
                     positive_result = ret is None
                     if _expected == positive_result:

munet/watchlog.py

@@ -15,6 +15,7 @@
 
 class MatchFoundError(Exception):
     """An error raised when a match is not found."""
+
     def __init__(self, watchlog, match):
         self.watchlog = watchlog
         self.match = match
@@ -139,12 +140,16 @@ def from_mark(self, mark=None):
 
         If ``mark`` is None then return content since last ``set_mark`` was called.
 
+        If the file has been replaced (inode changes) then the marks will reset to 0.
+
         Args:
             mark: the mark in the content to return file content from.
 
         Return:
             returns the content between ``mark`` and the end of content.
         """
+        if mark is None:
+            mark = self.last_user_mark
         return self.content[mark:]
 
     def set_mark(self):
@@ -156,8 +161,8 @@ def set_mark(self):
     def snapshot(self):
         """Update the file content and return new text.
 
-        Returns any new text added since the last snapshot,
-        also updates the snapshot mark.
+        Returns any new text added since the last snapshot and updates the snapshot
+        mark.
 
         Return:
             Newly added text.
@@ -168,3 +173,15 @@ def snapshot(self):
         last_mark = self.last_snap_mark
         self.last_snap_mark = len(self.content)
         return self.content[last_mark:]
+
+    def snapshot_refresh(self):
+        """Update the file content and return the snapshot with this new content.
+
+        If the file has been replaced (inode changes) then the marks will reset to 0.
+
+        Return:
+            returns the content from the last snapshot plus any new content
+            added since ``snapshot()`` was called.
+        """
+        self.update_content()
+        return self.from_mark(self.last_snap_mark)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "munet"
-version = "0.15.5"
+version = "0.15.6"
 description = "A package to facilitate network simulations"
 authors = ["Christian Hopps <chopps@labn.net>"]
 license = "GPL-2.0-or-later"

tests/testing/test_testing.py

@@ -27,16 +27,22 @@ def test_stepfunction(stepf):
     stepf("the second step")
 
 
+# Assert twice then succeed
 @retry(retry_timeout=2, retry_sleep=0.1)
 def retrying_test_assert():
     if not hasattr(retrying_test_assert, "count"):
         retrying_test_assert.count = 0
     else:
         retrying_test_assert.count += 1
     assert retrying_test_assert.count == 2, "count not 2"
-    retrying_test_assert.count = 0
 
 
+def test_retry_assert(caplog):
+    retrying_test_assert()
+    assert caplog.text.count("Sleeping") == 2
+
+
+# Fail twice then succeed
 @retry(retry_timeout=2, retry_sleep=0.1)
 def retrying_test_string():
     if not hasattr(retrying_test_string, "count"):
@@ -53,24 +59,71 @@ def test_retry_string(caplog):
     assert caplog.text.count("Sleeping") == 2
 
 
-def test_retry_fail(caplog):
+# Succeed twice then fail
+@retry(retry_timeout=2, retry_sleep=0.1)
+def retrying_test_fail_first():
+    if not hasattr(retrying_test_fail_first, "count"):
+        retrying_test_fail_first.count = 0
+    else:
+        retrying_test_fail_first.count += 1
+    if retrying_test_fail_first.count != 2:
+        return None
+    return "count is 2"
+
+
+def test_retry_expected_fail_first(caplog):
+    retrying_test_fail_first(expected=False)
+    assert caplog.text.count("Sleeping") == 2
+
+
+def test_retry_expected_fail(caplog):
     @retry(retry_timeout=1, retry_sleep=0.1)
     def retrying_fail():
         return "Fail"
 
-    retrying_fail(expected=False)
+    ret = retrying_fail(expected=False)
+    assert ret == "Fail"
     assert caplog.text.count("Sleeping") == 0
 
 
-def test_retry_assert_fail(caplog):
+def test_retry_assert_exception(caplog):
     @retry(retry_timeout=1, retry_sleep=0.2)
     def retrying_assert():
         assert False, "Fail"
 
     try:
-        # Expected does not consider assert an expected failure
+        # Expected does not consider assert an expected failure by default
         retrying_assert(expected=False)
     except AssertionError:
-        assert caplog.text.count("Sleeping") == 5
+        # Should have retried before ultimately failing
+        assert caplog.text.count("Sleeping") > 0
+    else:
+        assert False, "Failed b/c no exception raised"
+
+
+def test_retry_assert_expected(caplog):
+    @retry(retry_timeout=1, retry_sleep=0.2, assert_is_except=False)
+    def retrying_assert():
+        assert False, "Fail"
+
+    # Expected does not consider assert an expected failure by default
+    ret = retrying_assert(expected=False)
+    assert isinstance(ret, AssertionError)
+    assert caplog.text.count("Sleeping") == 0
+
+
+# Succeed twice then assert
+@retry(retry_timeout=2, retry_sleep=0.1, assert_is_except=False)
+def retrying_test_fail_first_assert():
+    if not hasattr(retrying_test_fail_first_assert, "count"):
+        retrying_test_fail_first_assert.count = 0
     else:
-        assert False, "Failed b/c succeeded"
+        retrying_test_fail_first_assert.count += 1
+    assert retrying_test_fail_first_assert.count != 2, "count is 2"
+
+
+def test_retry_assert_expected_succeed_first(caplog):
+    # Expected does not consider assert an expected failure by default
+    ret = retrying_test_fail_first_assert(expected=False)
+    assert isinstance(ret, AssertionError)
+    assert caplog.text.count("Sleeping") == 2