Add configuration for running health checks without threads

health_check/conf.py

@@ -4,3 +4,4 @@
 HEALTH_CHECK.setdefault("DISK_USAGE_MAX", 90)
 HEALTH_CHECK.setdefault("MEMORY_MIN", 100)
 HEALTH_CHECK.setdefault("WARNINGS_AS_ERRORS", True)
+HEALTH_CHECK.setdefault("DISABLE_THREADING", False)

health_check/mixins.py

@@ -40,16 +40,22 @@ def _run(plugin):
 
                 connections.close_all()
 
-        with ThreadPoolExecutor(max_workers=len(self.plugins) or 1) as executor:
-            for plugin in executor.map(_run, self.plugins):
-                if plugin.critical_service:
-                    if not HEALTH_CHECK["WARNINGS_AS_ERRORS"]:
-                        errors.extend(
-                            e
-                            for e in plugin.errors
-                            if not isinstance(e, ServiceWarning)
-                        )
-                    else:
-                        errors.extend(plugin.errors)
+        def _collect_errors(plugin):
+            if plugin.critical_service:
+                if not HEALTH_CHECK["WARNINGS_AS_ERRORS"]:
+                    errors.extend(
+                        e for e in plugin.errors if not isinstance(e, ServiceWarning)
+                    )
+                else:
+                    errors.extend(plugin.errors)
+
+        if HEALTH_CHECK["DISABLE_THREADING"]:
+            for plugin in self.plugins:
+                _run(plugin)
+                _collect_errors(plugin)
+        else:
+            with ThreadPoolExecutor(max_workers=len(self.plugins) or 1) as executor:
+                for plugin in executor.map(_run, self.plugins):
+                    _collect_errors(plugin)
 
         return errors

tests/test_mixins.py

@@ -1,6 +1,9 @@
+from unittest.mock import patch
+
 import pytest
 
 from health_check.backends import BaseHealthCheckBackend
+from health_check.conf import HEALTH_CHECK
 from health_check.mixins import CheckMixin
 from health_check.plugins import plugin_dir
 
@@ -28,11 +31,42 @@ def setup(self):
         yield
         plugin_dir.reset()
 
-    def test_plugins(self):
+    @pytest.mark.parametrize("disable_threading", [(True,), (False,)])
+    def test_plugins(self, monkeypatch, disable_threading):
+        monkeypatch.setitem(HEALTH_CHECK, "DISABLE_THREADING", disable_threading)
+
         assert len(Checker().plugins) == 2
 
-    def test_errors(self):
+    @pytest.mark.parametrize("disable_threading", [(True,), (False,)])
+    def test_errors(self, monkeypatch, disable_threading):
+        monkeypatch.setitem(HEALTH_CHECK, "DISABLE_THREADING", disable_threading)
+
         assert len(Checker().errors) == 1
 
-    def test_run_check(self):
+    @pytest.mark.parametrize("disable_threading", [(True,), (False,)])
+    def test_run_check(self, monkeypatch, disable_threading):
+        monkeypatch.setitem(HEALTH_CHECK, "DISABLE_THREADING", disable_threading)
+
         assert len(Checker().run_check()) == 1
+
+    def test_run_check_threading_enabled(self, monkeypatch):
+        """Ensure threading used when not disabled."""
+
+        # Ensure threading is enabled.
+        monkeypatch.setitem(HEALTH_CHECK, "DISABLE_THREADING", False)
+
+        # Ensure ThreadPoolExecutor is used
+        with patch("health_check.mixins.ThreadPoolExecutor") as tpe:
+            Checker().run_check()
+            tpe.assert_called()
+
+    def test_run_check_threading_disabled(self, monkeypatch):
+        """Ensure threading not used when disabled."""
+
+        # Ensure threading is disabled.
+        monkeypatch.setitem(HEALTH_CHECK, "DISABLE_THREADING", True)
+
+        # Ensure ThreadPoolExecutor is not used
+        with patch("health_check.mixins.ThreadPoolExecutor") as tpe:
+            Checker().run_check()
+            tpe.assert_not_called()