feat(supervisor): implement and verify automated tunnel recovery supervisor with test coverage

2026-10-09 12:57:28 +00:00
parent 6d2cbabe34
commit 87be6ae4fc
2 changed files with 167 additions and 0 deletions
@@ -0,0 +1,37 @@
+import os
+import subprocess
+import tempfile
+import unittest
+
+REPO_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+SCRIPT_PATH = os.path.join(REPO_DIR, "cloud-uptime", "uptime-watcher.sh")
+
+class TestUptimeWatcher(unittest.TestCase):
+    def test_script_exists_and_executable(self):
+        self.assertTrue(os.path.exists(SCRIPT_PATH), f"Script missing: {SCRIPT_PATH}")
+        self.assertTrue(os.access(SCRIPT_PATH, os.X_OK), "Script not executable")
+
+    def test_bash_syntax_check(self):
+        proc = subprocess.run(["bash", "-n", SCRIPT_PATH], capture_output=True, text=True)
+        self.assertEqual(proc.returncode, 0, f"Bash syntax error: {proc.stderr}")
+
+    def test_watcher_execution_in_sandbox(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            hooks_state = os.path.join(tmpdir, "hooks", "state", "uptime-watcher")
+            os.makedirs(hooks_state, exist_ok=True)
+            ws_tunnel = os.path.join(tmpdir, "workspace", "tunnel")
+            os.makedirs(ws_tunnel, exist_ok=True)
+            
+            env_file = os.path.join(ws_tunnel, "machine.env")
+            with open(env_file, "w") as f:
+                f.write("MUSE_MACHINE=test-node\nSSH_PORT=2224\nTERM_PORT=7681\n")
+                
+            env = os.environ.copy()
+            env["HOME"] = tmpdir
+            # Running with dry environment should safely exit (fail count tracked)
+            proc = subprocess.run([SCRIPT_PATH], env=env, capture_output=True, text=True)
+            # The script exits 0 even on fail unless fatal crash, logging status
+            self.assertTrue(os.path.exists(os.path.join(hooks_state, "consec_failures")))
+
+if __name__ == "__main__":
+    unittest.main()