paperscout/tests/test_shutdown.py at develop · cppalliance/paperscout · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
"""Tests for paperscout.shutdown."""

from __future__ import annotations

import logging
from http.server import HTTPServer
from unittest.mock import MagicMock

from paperscout.shutdown import shutdown_services, stop_bolt_server


class TestShutdownServices:
    def test_shutdown_services_drains_mq_and_logs(self, caplog):
        mq = MagicMock()
        mq.drain.return_value = 2
        with caplog.at_level(logging.INFO, logger="paperscout"):
            drained = shutdown_services(
                reason="SIGTERM",
                mq=mq,
                health_server=None,
                health_thread=None,
                app=None,
                bolt_thread=None,
                mq_drain_timeout=30.0,
                thread_join_timeout=5.0,
            )
        assert drained == 2
        mq.drain.assert_called_once_with(timeout=30.0)
        assert any("SIGTERM" in r.message and "drained 2" in r.message for r in caplog.records)

    def test_shutdown_services_skips_none_handles(self):
        shutdown_services(
            reason="unknown",
            mq=None,
            health_server=None,
            health_thread=None,
            app=None,
            bolt_thread=None,
            mq_drain_timeout=30.0,
            thread_join_timeout=5.0,
        )

    def test_stop_bolt_server_calls_shutdown(self):
        app = MagicMock()
        server = MagicMock()
        app._development_server = MagicMock(_server=server)
        stop_bolt_server(app)
        server.shutdown.assert_called_once()

    def test_shutdown_services_stops_health_server(self):
        health_server = MagicMock(spec=HTTPServer)
        shutdown_services(
            reason="SIGINT",
            mq=None,
            health_server=health_server,
            health_thread=None,
            app=None,
            bolt_thread=None,
            mq_drain_timeout=30.0,
            thread_join_timeout=5.0,
        )
        health_server.shutdown.assert_called_once()

    def test_shutdown_services_continues_after_mq_drain_failure(self, caplog):
        mq = MagicMock()
        mq.drain.side_effect = RuntimeError("drain boom")
        health_server = MagicMock(spec=HTTPServer)
        with caplog.at_level(logging.INFO, logger="paperscout"):
            drained = shutdown_services(
                reason="SIGTERM",
                mq=mq,
                health_server=health_server,
                health_thread=None,
                app=None,
                bolt_thread=None,
                mq_drain_timeout=30.0,
                thread_join_timeout=5.0,
            )
        assert drained == 0
        health_server.shutdown.assert_called_once()
        assert any("drained 0" in r.message for r in caplog.records)

    def test_shutdown_services_closes_pool(self, caplog):
        pool = MagicMock()
        with caplog.at_level(logging.INFO, logger="paperscout"):
            shutdown_services(
                reason="SIGTERM",
                mq=None,
                health_server=None,
                health_thread=None,
                app=None,
                bolt_thread=None,
                mq_drain_timeout=30.0,
                thread_join_timeout=5.0,
                pool=pool,
            )
        pool.closeall.assert_called_once()
        assert any("DB pool closed" in r.message for r in caplog.records)

    def test_shutdown_services_pool_close_raises_continues(self, caplog):
        mq = MagicMock()
        mq.drain.return_value = 1
        pool = MagicMock()
        pool.closeall.side_effect = RuntimeError("pool close boom")
        with caplog.at_level(logging.ERROR, logger="paperscout"):
            drained = shutdown_services(
                reason="SIGTERM",
                mq=mq,
                health_server=None,
                health_thread=None,
                app=None,
                bolt_thread=None,
                mq_drain_timeout=30.0,
                thread_join_timeout=5.0,
                pool=pool,
            )
        assert drained == 1
        pool.closeall.assert_called_once()

    def test_shutdown_services_skips_pool_when_none(self):
        shutdown_services(
            reason="unknown",
            mq=None,
            health_server=None,
            health_thread=None,
            app=None,
            bolt_thread=None,
            mq_drain_timeout=30.0,
            thread_join_timeout=5.0,
            pool=None,
        )