Add Local Test Runner Script

test/graph_test.py

@@ -2,6 +2,7 @@
 
 import json
 import os
+import time
 
 import pexpect
 from test_base import TestBase
@@ -39,21 +40,29 @@ def directory_exists(self):
             self.sut = pexpect.spawn("warnet create")
             self.sut.expect("name", timeout=10)
             self.sut.sendline("ANewNetwork")
+            time.sleep(0.1)
             self.sut.expect("many", timeout=10)
             self.sut.sendline("")
+            time.sleep(0.1)
             self.sut.expect("connections", timeout=10)
             self.sut.sendline("")
+            time.sleep(0.1)
             self.sut.expect("version", timeout=10)
             self.sut.sendline("")
+            time.sleep(0.1)
             self.sut.expect("enable fork-observer", timeout=10)
             self.sut.sendline("")
+            time.sleep(0.1)
             self.sut.expect("seconds", timeout=10)
             self.sut.sendline("")
+            time.sleep(0.1)
             self.sut.expect("enable grafana", timeout=10)
             self.sut.sendline("")
+            time.sleep(0.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,69 @@
+import subprocess
+import sys
+
+import yaml
+
+# 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") as file:
+    workflow = yaml.safe_load(file)
+
+tests_total = 0
+tests_skipped = 0
+tests_completed = 0
+
+for job_details in workflow.get("jobs", {}).values():
+    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(
+    "###################################################################################################"
+)