software/tests/test_signals.py

1"""Tests for graceful shutdown signal handling."""23from __future__ import annotations45import signal6import threading7import time89import pytest1011from src.orchestration.signals import (12    GracefulShutdown,13    ShutdownCallback,14    ignore_signals_in_child,15)161718class TestGracefulShutdown:19    """Tests for GracefulShutdown context manager."""2021    def test_initial_state(self) -> None:22        """Test initial state is not stopped."""23        shutdown = GracefulShutdown()24        assert shutdown.should_stop is False25        assert shutdown.signal_received is None2627    def test_context_manager_entry(self) -> None:28        """Test context manager returns self."""29        shutdown = GracefulShutdown()30        with shutdown as s:31            assert s is shutdown32            assert shutdown.should_stop is False3334    def test_request_stop(self) -> None:35        """Test programmatic stop request."""36        shutdown = GracefulShutdown()37        assert shutdown.should_stop is False3839        shutdown.request_stop()4041        assert shutdown.should_stop is True4243    def test_wait_for_stop_timeout(self) -> None:44        """Test wait_for_stop with timeout."""45        shutdown = GracefulShutdown()4647        start = time.time()48        result = shutdown.wait_for_stop(timeout=0.1)49        elapsed = time.time() - start5051        assert result is False52        assert elapsed >= 0.153        assert elapsed < 0.5  # Should not take too long5455    def test_wait_for_stop_signaled(self) -> None:56        """Test wait_for_stop when stop is signaled."""57        shutdown = GracefulShutdown()5859        # Signal stop in a separate thread60        def signal_stop() -> None:61            time.sleep(0.05)62            shutdown.request_stop()6364        thread = threading.Thread(target=signal_stop)65        thread.start()6667        result = shutdown.wait_for_stop(timeout=1.0)6869        thread.join()70        assert result is True7172    def test_context_manager_restores_handlers(self) -> None:73        """Test that context manager restores original handlers."""74        original_sigint = signal.getsignal(signal.SIGINT)75        original_sigterm = signal.getsignal(signal.SIGTERM)7677        with GracefulShutdown():78            pass7980        assert signal.getsignal(signal.SIGINT) == original_sigint81        assert signal.getsignal(signal.SIGTERM) == original_sigterm8283    def test_nested_context_managers(self) -> None:84        """Test that nested context managers work correctly."""85        with GracefulShutdown() as outer:86            with GracefulShutdown() as inner:87                inner.request_stop()88                assert inner.should_stop is True89                assert outer.should_stop is False9091    def test_multiple_stop_requests(self) -> None:92        """Test that multiple stop requests are idempotent."""93        shutdown = GracefulShutdown()9495        shutdown.request_stop()96        shutdown.request_stop()97        shutdown.request_stop()9899        assert shutdown.should_stop is True100101102class TestShutdownCallback:103    """Tests for ShutdownCallback utility."""104105    def test_register_callback(self) -> None:106        """Test registering a callback."""107        shutdown = GracefulShutdown()108        callbacks = ShutdownCallback(shutdown)109110        called = []111        callbacks.register(lambda: called.append(1))112113        callbacks.run_all()114115        assert called == [1]116117    def test_multiple_callbacks(self) -> None:118        """Test registering and running multiple callbacks."""119        shutdown = GracefulShutdown()120        callbacks = ShutdownCallback(shutdown)121122        results = []123        callbacks.register(lambda: results.append("first"))124        callbacks.register(lambda: results.append("second"))125        callbacks.register(lambda: results.append("third"))126127        callbacks.run_all()128129        assert results == ["first", "second", "third"]130131    def test_callback_exception_handling(self) -> None:132        """Test that callback exceptions are collected, not raised."""133        shutdown = GracefulShutdown()134        callbacks = ShutdownCallback(shutdown)135136        results = []137        callbacks.register(lambda: results.append("first"))138        callbacks.register(lambda: (_ for _ in ()).throw(ValueError("error")))139        callbacks.register(lambda: results.append("third"))140141        exceptions = callbacks.run_all()142143        # First and third should still run144        assert results == ["first", "third"]145        # Exception should be collected146        assert len(exceptions) == 1147        assert isinstance(exceptions[0], ValueError)148149    def test_run_all_empty(self) -> None:150        """Test run_all with no callbacks."""151        shutdown = GracefulShutdown()152        callbacks = ShutdownCallback(shutdown)153154        exceptions = callbacks.run_all()155156        assert exceptions == []157158159class TestIgnoreSignalsInChild:160    """Tests for ignore_signals_in_child function."""161162    def test_sets_signal_handlers(self) -> None:163        """Test that function sets SIG_IGN handlers."""164        original_sigint = signal.getsignal(signal.SIGINT)165        original_sigterm = signal.getsignal(signal.SIGTERM)166167        try:168            ignore_signals_in_child()169170            assert signal.getsignal(signal.SIGINT) == signal.SIG_IGN171            assert signal.getsignal(signal.SIGTERM) == signal.SIG_IGN172        finally:173            # Restore original handlers174            signal.signal(signal.SIGINT, original_sigint)175            signal.signal(signal.SIGTERM, original_sigterm)176177178class TestThreadSafety:179    """Tests for thread safety of shutdown handling."""180181    def test_concurrent_stop_requests(self) -> None:182        """Test concurrent stop requests from multiple threads."""183        shutdown = GracefulShutdown()184        threads = []185186        for _ in range(10):187            t = threading.Thread(target=shutdown.request_stop)188            threads.append(t)189            t.start()190191        for t in threads:192            t.join()193194        assert shutdown.should_stop is True195196    def test_concurrent_should_stop_reads(self) -> None:197        """Test concurrent reads of should_stop."""198        shutdown = GracefulShutdown()199        results = []200201        def reader() -> None:202            for _ in range(100):203                results.append(shutdown.should_stop)204                time.sleep(0.001)205206        threads = [threading.Thread(target=reader) for _ in range(5)]207208        for t in threads:209            t.start()210211        time.sleep(0.05)212        shutdown.request_stop()213214        for t in threads:215            t.join()216217        # Should have some False and some True218        assert False in results219        assert True in results220