|
| 1 | +"""Exercise readiness retries and failure reporting without a Docker daemon.""" |
| 2 | +import os |
| 3 | +from pathlib import Path |
| 4 | +import subprocess |
| 5 | +import sys |
| 6 | +import tempfile |
| 7 | +import unittest |
| 8 | + |
| 9 | + |
| 10 | +class DatabaseReadinessTests(unittest.TestCase): |
| 11 | + def run_wait(self, behavior, driver="postgres", wait_seconds="2"): |
| 12 | + with tempfile.TemporaryDirectory() as directory: |
| 13 | + root = Path(directory) |
| 14 | + docker = root / "docker" |
| 15 | + docker.write_text( |
| 16 | + f"#!{sys.executable}\n" + ''' |
| 17 | +import os |
| 18 | +from pathlib import Path |
| 19 | +import sys |
| 20 | +import time |
| 21 | +
|
| 22 | +state = Path(os.environ["PROBE_STATE"]) |
| 23 | +mode = os.environ["PROBE_BEHAVIOR"] |
| 24 | +command = sys.argv[1] |
| 25 | +if command == "exec": |
| 26 | + attempts = int(state.read_text()) if state.exists() else 0 |
| 27 | + state.write_text(str(attempts + 1)) |
| 28 | + if mode == "hang": |
| 29 | + time.sleep(60) |
| 30 | + sys.exit(0 if mode == "ready" or (mode == "retry" and attempts > 0) else 1) |
| 31 | +if command == "inspect": |
| 32 | + print("false" if mode == "stopped" else "true") |
| 33 | +if command == "logs": |
| 34 | + print("database startup diagnostics") |
| 35 | +''' |
| 36 | + ) |
| 37 | + docker.chmod(0o755) |
| 38 | + env = dict(os.environ, PATH=f"{root}{os.pathsep}{os.environ['PATH']}", |
| 39 | + PROBE_STATE=str(root / "attempts"), PROBE_BEHAVIOR=behavior) |
| 40 | + result = subprocess.run( |
| 41 | + ["bash", str(Path(__file__).with_name("wait-for-db.sh")), |
| 42 | + "test-container", driver, wait_seconds], |
| 43 | + env=env, text=True, capture_output=True, timeout=8, |
| 44 | + ) |
| 45 | + attempts = int((root / "attempts").read_text()) if (root / "attempts").exists() else 0 |
| 46 | + return result, attempts |
| 47 | + |
| 48 | + def test_supported_drivers_succeed_when_ready(self): |
| 49 | + for driver in ("postgres", "mysql", "mssql"): |
| 50 | + with self.subTest(driver=driver): |
| 51 | + result, attempts = self.run_wait("ready", driver) |
| 52 | + self.assertEqual(result.returncode, 0, result.stderr) |
| 53 | + self.assertEqual(attempts, 1) |
| 54 | + |
| 55 | + def test_retries_until_schema_is_ready(self): |
| 56 | + result, attempts = self.run_wait("retry", wait_seconds="5") |
| 57 | + self.assertEqual(result.returncode, 0, result.stderr) |
| 58 | + self.assertGreater(attempts, 1) |
| 59 | + |
| 60 | + def test_stopped_container_fails_with_logs(self): |
| 61 | + result, attempts = self.run_wait("stopped", wait_seconds="60") |
| 62 | + self.assertNotEqual(result.returncode, 0) |
| 63 | + self.assertEqual(attempts, 1) |
| 64 | + self.assertIn("database startup diagnostics", result.stderr) |
| 65 | + |
| 66 | + def test_hung_probe_is_bounded_and_reports_logs(self): |
| 67 | + result, _ = self.run_wait("hang", wait_seconds="1") |
| 68 | + self.assertNotEqual(result.returncode, 0) |
| 69 | + self.assertIn("database startup diagnostics", result.stderr) |
| 70 | + |
| 71 | + def test_invalid_arguments_never_probe_database(self): |
| 72 | + for driver, wait in (("unknown", "2"), ("postgres", "0")): |
| 73 | + with self.subTest(driver=driver, wait=wait): |
| 74 | + result, attempts = self.run_wait("ready", driver, wait) |
| 75 | + self.assertEqual(result.returncode, 2) |
| 76 | + self.assertEqual(attempts, 0) |
| 77 | + |
| 78 | + |
| 79 | +if __name__ == "__main__": |
| 80 | + unittest.main() |
0 commit comments