- Add local_test_runner script

- add delay to graph_test expect - increases reliability

Author: macgyver13

test/graph_test.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 
 import json
-import os
+import os, time
 
 import pexpect
 from test_base import TestBase
@@ -39,21 +39,29 @@ def directory_exists(self):
             self.sut = pexpect.spawn("warnet create")
             self.sut.expect("name", timeout=10)
             self.sut.sendline("ANewNetwork")
+            time.sleep(.1)
             self.sut.expect("many", timeout=10)
             self.sut.sendline("")
+            time.sleep(.1)
             self.sut.expect("connections", timeout=10)
             self.sut.sendline("")
+            time.sleep(.1)
             self.sut.expect("version", timeout=10)
             self.sut.sendline("")
+            time.sleep(.1)
             self.sut.expect("enable fork-observer", timeout=10)
             self.sut.sendline("")
+            time.sleep(.1)
             self.sut.expect("seconds", timeout=10)
             self.sut.sendline("")
+            time.sleep(.1)
             self.sut.expect("enable grafana", timeout=10)
             self.sut.sendline("")
+            time.sleep(.1)
             self.sut.expect("successfully", timeout=50)
         except Exception as e:
             print(f"\nReceived prompt text:\n  {self.sut.before.decode('utf-8')}\n")
+            print(os.listdir(self.tmpdir))
             raise e
 
     def run_created_network(self):

test/local_test_runner.py

@@ -0,0 +1,56 @@
+import yaml
+import subprocess, sys
+
+# developers can use this tool to verify all repo workflow test execute to completion
+#
+# execute from repo root like this to execute all tests
+#  python test/local_test_runner.py
+#
+# add optional argument to just execute any matching tests
+#  python test/local_test_runner.py [ln_ | graph]
+
+
+def has_key_path(d, key_path, separator="."):
+    """Check if a nested key path (dotted notation) exists in a dictionary."""
+    keys = key_path.split(separator)
+    for key in keys:
+        if not isinstance(d, dict) or key not in d:
+            return False
+        d = d[key]
+    return True
+
+# Load the workflow file
+with open('.github/workflows/test.yml', 'r') as file:
+    workflow = yaml.safe_load(file)
+
+tests_total=0
+tests_skipped=0
+tests_completed=0
+
+for job_name, job_details in workflow.get("jobs", {}).items():
+    if has_key_path(job_details, "strategy.matrix.test"):
+        print("Found test strategy job, starting serial execution of each test")
+        tests = job_details["strategy"]["matrix"]["test"]
+    
+        for test in tests:
+            tests_total+=1
+            if len(sys.argv) > 1 and sys.argv[1] not in test:
+                print("skipping test as requested:",test)
+                tests_skipped+=1
+                continue
+            command = f"python test/{test}"
+            print("###################################################################################################")
+            print("############## executing:", command)
+            print("###################################################################################################")
+            process = subprocess.run(command, shell=True)
+            if process.returncode != 0:
+                print("******** testing failed")
+                if process.stdout: print("stdout:", process.stdout)
+                if process.stderr: print("stderr:", process.stderr)
+                sys.exit(1)
+            tests_completed+=1
+
+print("###################################################################################################")
+print("testing complete")
+print(f"{tests_completed} of {tests_total} complete - skipped: {tests_skipped} tests")
+print("###################################################################################################")