Ensure shutdown does not deadlock (#49282)

tests/util/test_thread.py

@@ -1,9 +1,11 @@
 """Test Home Assistant thread utils."""
 
 import asyncio
+from unittest.mock import Mock, patch
 
 import pytest
 
+from homeassistant.util import thread
 from homeassistant.util.async_ import run_callback_threadsafe
 from homeassistant.util.thread import ThreadWithException
 
@@ -53,3 +55,57 @@ async def test_thread_fails_raise(hass):
 
 class _EmptyClass:
     """An empty class."""
+
+
+async def test_deadlock_safe_shutdown_no_threads():
+    """Test we can shutdown without deadlock without any threads to join."""
+
+    dead_thread_mock = Mock(
+        join=Mock(), daemon=False, is_alive=Mock(return_value=False)
+    )
+    daemon_thread_mock = Mock(
+        join=Mock(), daemon=True, is_alive=Mock(return_value=True)
+    )
+    mock_threads = [
+        dead_thread_mock,
+        daemon_thread_mock,
+    ]
+
+    with patch("homeassistant.util.threading.enumerate", return_value=mock_threads):
+        thread.deadlock_safe_shutdown()
+
+    assert not dead_thread_mock.join.called
+    assert not daemon_thread_mock.join.called
+
+
+async def test_deadlock_safe_shutdown():
+    """Test we can shutdown without deadlock."""
+
+    normal_thread_mock = Mock(
+        join=Mock(), daemon=False, is_alive=Mock(return_value=True)
+    )
+    dead_thread_mock = Mock(
+        join=Mock(), daemon=False, is_alive=Mock(return_value=False)
+    )
+    daemon_thread_mock = Mock(
+        join=Mock(), daemon=True, is_alive=Mock(return_value=True)
+    )
+    exception_thread_mock = Mock(
+        join=Mock(side_effect=Exception), daemon=False, is_alive=Mock(return_value=True)
+    )
+    mock_threads = [
+        normal_thread_mock,
+        dead_thread_mock,
+        daemon_thread_mock,
+        exception_thread_mock,
+    ]
+
+    with patch("homeassistant.util.threading.enumerate", return_value=mock_threads):
+        thread.deadlock_safe_shutdown()
+
+    expected_timeout = thread.THREADING_SHUTDOWN_TIMEOUT / 2
+
+    assert normal_thread_mock.join.call_args[0] == (expected_timeout,)
+    assert not dead_thread_mock.join.called
+    assert not daemon_thread_mock.join.called
+    assert exception_thread_mock.join.call_args[0] == (expected_timeout,)