Add unit tests for wait_loop

tests/test_wait_loop.py

@@ -0,0 +1,230 @@
+import logging
+import shutil
+from tempfile import mkdtemp
+from unittest.mock import Mock, call
+
+import nginx_config_reloader
+from nginx_config_reloader import ListenTargetTerminated, wait_loop
+from tests.testcase import TestCase
+
+
+class TestWaitLoop(TestCase):
+    def setUp(self):
+        self.source = mkdtemp()
+        self.mock_logger = Mock(spec_set=logging.Logger)
+        self.nginx_config_reloader = self.set_up_patch(
+            "nginx_config_reloader.NginxConfigReloader"
+        )
+        self.mock_handler = Mock()
+        self.nginx_config_reloader.return_value = self.mock_handler
+        self.set_up_patch("nginx_config_reloader.SYSTEM_BUS")
+        self.set_up_patch("nginx_config_reloader.NginxConfigReloaderInterface")
+        self.set_up_patch("nginx_config_reloader.threading.Thread")
+        self.time_sleep = self.set_up_patch("nginx_config_reloader.time.sleep")
+        self.after_loop = self.set_up_patch("nginx_config_reloader.after_loop")
+
+    def tearDown(self):
+        shutil.rmtree(self.source, ignore_errors=True)
+
+    def test_wait_loop_creates_nginx_config_reloader_handler(self):
+        self._run_wait_loop_with_keyboard_interrupt()
+
+        self.nginx_config_reloader.assert_called_once_with(
+            logger=self.mock_logger,
+            no_magento_config=False,
+            no_custom_config=False,
+            dir_to_watch=self.source,
+            use_systemd=False,
+        )
+
+    def test_wait_loop_creates_handler_with_custom_arguments(self):
+        self._run_wait_loop_with_keyboard_interrupt(
+            no_magento_config=True,
+            no_custom_config=True,
+            use_systemd=True,
+        )
+
+        self.nginx_config_reloader.assert_called_once_with(
+            logger=self.mock_logger,
+            no_magento_config=True,
+            no_custom_config=True,
+            dir_to_watch=self.source,
+            use_systemd=True,
+        )
+
+    def test_wait_loop_sets_up_dbus_when_no_dbus_is_false(self):
+        system_bus = self.set_up_patch("nginx_config_reloader.SYSTEM_BUS")
+        interface_class = self.set_up_patch(
+            "nginx_config_reloader.NginxConfigReloaderInterface"
+        )
+        thread_class = self.set_up_patch("nginx_config_reloader.threading.Thread")
+
+        self._run_wait_loop_with_keyboard_interrupt(no_dbus=False)
+
+        interface_class.assert_called_once_with(self.mock_handler)
+        system_bus.publish_object.assert_called_once()
+        system_bus.register_service.assert_called_once()
+        thread_class.assert_called_once_with(
+            target=nginx_config_reloader.dbus_event_loop
+        )
+        thread_class.return_value.start.assert_called_once()
+
+    def test_wait_loop_skips_dbus_setup_when_no_dbus_is_true(self):
+        system_bus = self.set_up_patch("nginx_config_reloader.SYSTEM_BUS")
+        interface_class = self.set_up_patch(
+            "nginx_config_reloader.NginxConfigReloaderInterface"
+        )
+        thread_class = self.set_up_patch("nginx_config_reloader.threading.Thread")
+
+        self._run_wait_loop_with_keyboard_interrupt(no_dbus=True)
+
+        interface_class.assert_not_called()
+        system_bus.publish_object.assert_not_called()
+        system_bus.register_service.assert_not_called()
+        thread_class.assert_not_called()
+
+    def test_wait_loop_waits_for_directory_to_appear(self):
+        # Return False twice, then True on subsequent calls
+        exists_mock = self.set_up_patch(
+            "nginx_config_reloader.os.path.exists", side_effect=[False, False, True]
+        )
+
+        self._run_wait_loop_with_keyboard_interrupt()
+
+        self.assertEqual(exists_mock.call_count, 3)
+        # Should have called sleep(5) twice while waiting for directory
+        sleep_calls = [c for c in self.time_sleep.call_args_list if c == call(5)]
+        self.assertEqual(len(sleep_calls), 2)
+
+    def test_wait_loop_logs_warning_when_directory_not_found(self):
+        # Return False once, then True
+        self.set_up_patch(
+            "nginx_config_reloader.os.path.exists", side_effect=[False, True]
+        )
+
+        self._run_wait_loop_with_keyboard_interrupt()
+
+        self.mock_logger.warning.assert_any_call(
+            f"Configuration dir {self.source} not found, waiting..."
+        )
+
+    def test_wait_loop_calls_reload_with_send_signal_false(self):
+        self._run_wait_loop_with_keyboard_interrupt()
+
+        self.mock_handler.reload.assert_called_once_with(send_signal=False)
+
+    def test_wait_loop_starts_observer(self):
+        self._run_wait_loop_with_keyboard_interrupt()
+
+        self.mock_handler.start_observer.assert_called_once()
+
+    def test_wait_loop_calls_after_loop_in_loop(self):
+        loop_count = [0]
+
+        def mock_after_loop(handler):
+            loop_count[0] += 1
+            if loop_count[0] >= 3:
+                raise KeyboardInterrupt
+
+        self.after_loop.side_effect = mock_after_loop
+
+        wait_loop(
+            logger=self.mock_logger,
+            dir_to_watch=self.source,
+            no_dbus=True,
+        )
+
+        self.assertEqual(self.after_loop.call_count, 3)
+        self.after_loop.assert_called_with(self.mock_handler)
+
+    def test_wait_loop_sleeps_one_second_between_after_loop_calls(self):
+        loop_count = [0]
+
+        def mock_sleep(seconds):
+            if seconds == 1:
+                loop_count[0] += 1
+                if loop_count[0] >= 2:
+                    raise KeyboardInterrupt
+
+        self.time_sleep.side_effect = mock_sleep
+
+        wait_loop(
+            logger=self.mock_logger,
+            dir_to_watch=self.source,
+            no_dbus=True,
+        )
+
+        self.assertGreaterEqual(loop_count[0], 2)
+
+    def test_wait_loop_handles_listen_target_terminated(self):
+        call_count = [0]
+
+        def mock_start_observer():
+            call_count[0] += 1
+            if call_count[0] == 1:
+                raise ListenTargetTerminated
+            raise KeyboardInterrupt
+
+        self.mock_handler.start_observer.side_effect = mock_start_observer
+
+        wait_loop(
+            logger=self.mock_logger,
+            dir_to_watch=self.source,
+            no_dbus=True,
+        )
+
+        # Should have tried to start observer twice
+        self.assertEqual(self.mock_handler.start_observer.call_count, 2)
+        # Should have stopped observer after ListenTargetTerminated
+        self.mock_handler.stop_observer.assert_called()
+        # Should have logged warning
+        self.mock_logger.warning.assert_any_call(
+            "Configuration dir lost, waiting for it to reappear"
+        )
+
+    def test_wait_loop_stops_observer_on_keyboard_interrupt(self):
+        self._run_wait_loop_with_keyboard_interrupt()
+
+        self.mock_handler.stop_observer.assert_called_once()
+
+    def test_wait_loop_logs_info_on_keyboard_interrupt(self):
+        self._run_wait_loop_with_keyboard_interrupt()
+
+        self.mock_logger.info.assert_any_call("Shutting down observer.")
+
+    def test_wait_loop_logs_info_when_starting_to_listen(self):
+        self._run_wait_loop_with_keyboard_interrupt()
+
+        self.mock_logger.info.assert_any_call(f"Listening for changes to {self.source}")
+
+    def test_wait_loop_reloads_config_after_listen_target_terminated(self):
+        call_count = [0]
+
+        def mock_start_observer():
+            call_count[0] += 1
+            if call_count[0] == 1:
+                raise ListenTargetTerminated
+            raise KeyboardInterrupt
+
+        self.mock_handler.start_observer.side_effect = mock_start_observer
+
+        wait_loop(
+            logger=self.mock_logger,
+            dir_to_watch=self.source,
+            no_dbus=True,
+        )
+
+        # reload should be called twice - once initially and once after recovery
+        self.assertEqual(self.mock_handler.reload.call_count, 2)
+
+    def _run_wait_loop_with_keyboard_interrupt(self, **kwargs):
+        """Helper to run wait_loop that exits on first after_loop call."""
+        self.after_loop.side_effect = KeyboardInterrupt
+
+        default_kwargs = {
+            "dir_to_watch": self.source,
+            "no_dbus": True,
+        }
+        default_kwargs.update(kwargs)
+
+        wait_loop(logger=self.mock_logger, **default_kwargs)