sandcastle-ng

test_sandcastle_wait.py (6805B)

---

 # This file is in the public domain.
 
 import os
 import signal
 import subprocess
 import tempfile
 import time
 import unittest
 from pathlib import Path
 
 REPOSITORY = Path(__file__).resolve().parents[1]
 WAIT_SCRIPT = REPOSITORY / "sandcastle-wait"
 
 FAKE_PODMAN = r"""#!/usr/bin/env python3
 import os
 import signal
 import sys
 import time
 from pathlib import Path
 
 
 arguments = sys.argv[1:]
 with Path(os.environ["FAKE_PODMAN_LOG"]).open("a", encoding="utf-8") as log:
     log.write(" ".join(arguments) + "\n")
 
 if arguments[0] == "inspect":
     running = os.environ.get("FAKE_CONTAINER_RUNNING", "true")
     if running == "missing":
         raise SystemExit(125)
     print(running)
     raise SystemExit(0)
 
 if arguments[:2] != ["exec", "taler-sandcastle"]:
     raise SystemExit(125)
 
 command = arguments[2:]
 if command[:2] == ["systemctl", "show"]:
     states = os.environ["FAKE_UNIT_STATES"].split(",")
     state_index = Path(os.environ["FAKE_STATE_INDEX"])
     try:
         index = int(state_index.read_text(encoding="utf-8"))
     except FileNotFoundError:
         index = 0
     state_index.write_text(str(index + 1), encoding="utf-8")
     state = states[min(index, len(states) - 1)]
     if state == "ERROR":
         raise SystemExit(125)
     print(state)
     raise SystemExit(0)
 
 if command[:2] == ["systemctl", "status"]:
     print("diagnostic unit status")
     raise SystemExit(3)
 
 if command[0] == "journalctl":
     stopped = Path(os.environ["FAKE_JOURNAL_STOPPED"])
 
     def terminate(_signum, _frame):
         stopped.write_text("stopped\n", encoding="utf-8")
         raise SystemExit(0)
 
     signal.signal(signal.SIGTERM, terminate)
     print("provisioning journal output", flush=True)
     while True:
         time.sleep(0.1)
 
 raise SystemExit(125)
 """
 
 
 class SandcastleWaitTests(unittest.TestCase):
     def setUp(self):
         self.temporary = tempfile.TemporaryDirectory()
         self.root = Path(self.temporary.name)
         self.bin_dir = self.root / "bin"
         self.bin_dir.mkdir()
         podman = self.bin_dir / "podman"
         podman.write_text(FAKE_PODMAN, encoding="utf-8")
         podman.chmod(0o755)
 
     def tearDown(self):
         self.temporary.cleanup()
 
     def environment(self, states="active", running="true"):
         environment = os.environ.copy()
         environment.update(
             {
                 "PATH": f"{self.bin_dir}:{environment['PATH']}",
                 "FAKE_CONTAINER_RUNNING": running,
                 "FAKE_UNIT_STATES": states,
                 "FAKE_STATE_INDEX": str(self.root / "state-index"),
                 "FAKE_PODMAN_LOG": str(self.root / "podman-log"),
                 "FAKE_JOURNAL_STOPPED": str(self.root / "journal-stopped"),
             }
         )
         return environment
 
     def run_wait(self, states="active", running="true"):
         return subprocess.run(
             [str(WAIT_SCRIPT)],
             env=self.environment(states, running),
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True,
             timeout=5,
         )
 
     def podman_log(self):
         try:
             return (self.root / "podman-log").read_text(encoding="utf-8")
         except FileNotFoundError:
             return ""
 
     def test_missing_or_stopped_container_fails_immediately(self):
         for running in ("missing", "false"):
             with self.subTest(running=running):
                 result = self.run_wait(running=running)
                 self.assertNotEqual(0, result.returncode)
                 self.assertIn("does not exist or is not running", result.stderr)
 
     def test_already_successful_unit_exits_without_following_logs(self):
         result = self.run_wait()
 
         self.assertEqual(0, result.returncode)
         self.assertIn("provisioning succeeded", result.stdout)
         self.assertNotIn("journalctl", self.podman_log())
 
     def test_transient_exec_race_is_retried(self):
         result = self.run_wait(states="ERROR,active")
 
         self.assertEqual(0, result.returncode)
         self.assertEqual(2, self.podman_log().count("systemctl show"))
 
     def test_already_failed_unit_prints_diagnostics(self):
         result = self.run_wait(states="failed")
 
         self.assertNotEqual(0, result.returncode)
         self.assertIn("provisioning failed", result.stderr)
         self.assertIn("diagnostic unit status", result.stderr)
         self.assertIn("systemctl status", self.podman_log())
 
     def test_unexpected_unit_state_fails(self):
         result = self.run_wait(states="inactive")
 
         self.assertNotEqual(0, result.returncode)
         self.assertIn("unexpected state 'inactive'", result.stderr)
         self.assertIn("diagnostic unit status", result.stderr)
 
     def test_running_unit_follows_logs_and_succeeds(self):
         result = self.run_wait(states="activating,active")
 
         self.assertEqual(0, result.returncode)
         self.assertIn("provisioning journal output", result.stdout)
         self.assertIn("provisioning succeeded", result.stdout)
         self.assertTrue((self.root / "journal-stopped").is_file())
 
     def test_running_unit_follows_logs_and_reports_failure(self):
         result = self.run_wait(states="activating,failed")
 
         self.assertNotEqual(0, result.returncode)
         self.assertIn("provisioning journal output", result.stdout)
         self.assertIn("provisioning failed", result.stderr)
         self.assertIn("diagnostic unit status", result.stderr)
         self.assertTrue((self.root / "journal-stopped").is_file())
 
     def test_losing_contact_while_waiting_fails_and_stops_journal(self):
         result = self.run_wait(states="activating,ERROR")
 
         self.assertNotEqual(0, result.returncode)
         self.assertIn("Lost contact", result.stderr)
         self.assertTrue((self.root / "journal-stopped").is_file())
 
     def test_termination_signal_stops_journal(self):
         process = subprocess.Popen(
             [str(WAIT_SCRIPT)],
             env=self.environment(states="activating"),
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True,
         )
         try:
             deadline = time.monotonic() + 5
             while "journalctl" not in self.podman_log():
                 self.assertIsNone(process.poll())
                 if time.monotonic() >= deadline:
                     self.fail("journal follower did not start")
                 time.sleep(0.01)
 
             process.send_signal(signal.SIGTERM)
             process.communicate(timeout=5)
         finally:
             if process.poll() is None:
                 process.terminate()
                 process.communicate(timeout=5)
 
         self.assertEqual(128 + signal.SIGTERM, process.returncode)
         self.assertTrue((self.root / "journal-stopped").is_file())
 
 
 if __name__ == "__main__":
     unittest.main()