add optional command line arg to run_all_fuzz_tests (#1188)

lib/agent0/agent0/hyperdrive/interactive/interactive_hyperdrive_test.py

@@ -246,11 +246,13 @@ def test_advance_time_with_checkpoints(chain: LocalChain):
     # in which case 1 checkpoint will be made. Hence, we can't be certain on how many checkpoints
     # were made per advance time.
 
+    min_time_error = 4  # seconds
+
     # Advance time lower than a checkpoint duration
     pre_time = hyperdrive_interface.get_block_timestamp(hyperdrive_interface.get_current_block())
     checkpoint_events = chain.advance_time(600, create_checkpoints=True)
     post_time = hyperdrive_interface.get_block_timestamp(hyperdrive_interface.get_current_block())
-    assert abs(post_time - pre_time - 600) <= 3
+    assert abs(post_time - pre_time - 600) <= min_time_error
     # assert 0 or 1 checkpoints made
     assert len(checkpoint_events[interactive_hyperdrive]) in {0, 1}
 
@@ -259,7 +261,7 @@ def test_advance_time_with_checkpoints(chain: LocalChain):
     checkpoint_events = chain.advance_time(3600, create_checkpoints=True)
     post_time = hyperdrive_interface.get_block_timestamp(hyperdrive_interface.get_current_block())
     # Advancing time equal to checkpoint duration results in time being off by few second
-    assert abs(post_time - pre_time - 3600) <= 3
+    assert abs(post_time - pre_time - 3600) <= min_time_error
     # assert one checkpoint made
     assert len(checkpoint_events[interactive_hyperdrive]) in {1, 2}
 
@@ -268,7 +270,7 @@ def test_advance_time_with_checkpoints(chain: LocalChain):
     checkpoint_events = chain.advance_time(datetime.timedelta(hours=3), create_checkpoints=True)
     post_time = hyperdrive_interface.get_block_timestamp(hyperdrive_interface.get_current_block())
     # Advancing time equal to checkpoint duration results in time being off by few second
-    assert abs(post_time - pre_time - 3600 * 3) <= 3
+    assert abs(post_time - pre_time - 3600 * 3) <= min_time_error
     # assert 3 checkpoints made
     assert len(checkpoint_events[interactive_hyperdrive]) in {3, 4}
 
@@ -278,7 +280,7 @@ def test_advance_time_with_checkpoints(chain: LocalChain):
     checkpoint_events = chain.advance_time(4000, create_checkpoints=True)
     post_time = hyperdrive_interface.get_block_timestamp(hyperdrive_interface.get_current_block())
     # Advancing time equal to checkpoint duration results in time being off by few second
-    assert abs(post_time - pre_time - 4000) <= 3
+    assert abs(post_time - pre_time - 4000) <= min_time_error
     # assert 1 checkpoint made
     assert len(checkpoint_events[interactive_hyperdrive]) in {1, 2}
 

lib/agent0/agent0/interactive_fuzz/run_all_fuzz_tests.py

@@ -1,6 +1,10 @@
 """Runs all fuzz tests forever."""
 from __future__ import annotations
 
+import argparse
+import sys
+from typing import NamedTuple, Sequence
+
 from hyperlogs.rollbar_utilities import initialize_rollbar
 
 from agent0.hyperdrive.interactive.chain import LocalChain
@@ -12,13 +16,22 @@
 )
 
 
-def main():
-    """Runs all fuzz tests"""
+def main(argv: Sequence[str] | None = None):
+    """Runs all fuzz tests.
+
+    Arguments
+    ---------
+    argv: Sequence[str]
+        The argv values returned from argparser.
+    """
+    parsed_args = parse_arguments(argv)
+
     initialize_rollbar("interactivefuzz")
 
     num_trades = 10
     num_paths_checked = 10
 
+    num_checks = 0
     while True:
         try:
             print("Running hyperdrive balance test")
@@ -47,6 +60,59 @@ def main():
             fuzz_profit_check(chain_config)
         except AssertionError:
             pass
+        num_checks += 1
+        if parsed_args.number_of_runs > 0 and num_checks > parsed_args.number_of_runs:
+            break
+
+
+class Args(NamedTuple):
+    """Command line arguments for the invariant checker."""
+
+    number_of_runs: int
+
+
+def namespace_to_args(namespace: argparse.Namespace) -> Args:
+    """Converts argprase.Namespace to Args.
+
+    Arguments
+    ---------
+    namespace: argparse.Namespace
+        Object for storing arg attributes.
+
+    Returns
+    -------
+    Args
+        Formatted arguments
+    """
+    return Args(
+        number_of_runs=namespace.number_of_runs,
+    )
+
+
+def parse_arguments(argv: Sequence[str] | None = None) -> Args:
+    """Parses input arguments.
+
+    Arguments
+    ---------
+    argv: Sequence[str]
+        The argv values returned from argparser.
+
+    Returns
+    -------
+    Args
+        Formatted arguments
+    """
+    parser = argparse.ArgumentParser(description="Runs a loop to check Hyperdrive invariants at each block.")
+    parser.add_argument(
+        "--number_of_runs",
+        type=int,
+        default=0,
+        help="The number times to run the tests. If not set, will run forever.",
+    )
+    # Use system arguments if none were passed
+    if argv is None:
+        argv = sys.argv
+    return namespace_to_args(parser.parse_args())
 
 
 if __name__ == "__main__":