business-assistant/scripts/tests/test_deploy_signals.py

"""Synthetic process identities and signals; never signal a real process."""
from contextlib import nullcontext
import errno
from pathlib import Path
import signal
from tempfile import TemporaryDirectory
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch

from scripts import deploy


class SignalCompatibilityTests(unittest.TestCase):
    def setUp(self):
        temporary = TemporaryDirectory()
        self.addCleanup(temporary.cleanup)
        self.root = Path(temporary.name)
        self.deployment = deploy.Deployment(self.root)
        self.identity = {'pid': 4321, 'start': '100', 'boot': 'synthetic-boot'}
        self.deployment.identity = Mock(return_value=self.identity)

    def test_missing_both_or_one_function_uses_checked_pid(self):
        for available in ('neither', 'open_only', 'send_only'):
            with self.subTest(available=available):
                opened, sent, killed = Mock(), Mock(), Mock()
                fake_os = SimpleNamespace(kill=killed)
                fake_signal = SimpleNamespace()
                if available == 'open_only':
                    fake_os.pidfd_open = opened
                if available == 'send_only':
                    fake_signal.pidfd_send_signal = sent
                self.deployment.identity.reset_mock()
                with patch.object(deploy, 'os', fake_os), patch.object(deploy, 'signal', fake_signal):
                    self.deployment.send(self.identity, 15)
                killed.assert_called_once_with(4321, 15)
                self.assertEqual(self.deployment.identity.call_count, 2)
                opened.assert_not_called()
                sent.assert_not_called()

    def test_identity_change_exit_or_other_checkout_never_signals_in_fallback(self):
        for replacement in (None, {**self.identity, 'start': '101'}, {**self.identity, 'boot': 'new-boot'}):
            for stage in ('initial', 'final'):
                with self.subTest(replacement=replacement, stage=stage):
                    self.deployment.identity = Mock(side_effect=[replacement] if stage == 'initial'
                                                    else [self.identity, replacement])
                    kill = Mock()
                    with patch.object(deploy, 'os', SimpleNamespace(kill=kill)), \
                            patch.object(deploy, 'signal', SimpleNamespace()):
                        self.deployment.send(self.identity, 15)
                    kill.assert_not_called()

    def test_pidfd_remains_preferred_and_fd_is_closed(self):
        with patch.object(deploy.os, 'pidfd_open', return_value=88, create=True) as opened, \
                patch.object(deploy.signal, 'pidfd_send_signal', create=True) as send, \
                patch.object(deploy.os, 'close') as close, patch.object(deploy.os, 'kill') as kill:
            self.deployment.send(self.identity, signal.SIGTERM)
        opened.assert_called_once_with(4321)
        send.assert_called_once_with(88, signal.SIGTERM)
        close.assert_called_once_with(88)
        kill.assert_not_called()

    def test_kernel_unsupported_open_or_send_falls_back_and_closes_fd(self):
        for stage in ('open', 'send'):
            for code in (errno.ENOSYS, errno.ENOTSUP):
                with self.subTest(stage=stage, code=code):
                    with patch.object(deploy.os, 'pidfd_open', return_value=88, create=True) as opened, \
                            patch.object(deploy.signal, 'pidfd_send_signal', create=True) as send, \
                            patch.object(deploy.os, 'close') as close, patch.object(deploy.os, 'kill') as kill:
                        (opened if stage == 'open' else send).side_effect = OSError(code, 'unsupported')
                        self.deployment.send(self.identity, 15)
                    kill.assert_called_once_with(4321, 15)
                    if stage == 'send':
                        close.assert_called_once_with(88)
                    else:
                        close.assert_not_called()

    def test_permission_and_other_errors_do_not_fall_back(self):
        for stage in ('open', 'send'):
            for code in (errno.EPERM, errno.EBADF, errno.EINVAL):
                with self.subTest(stage=stage, code=code):
                    with patch.object(deploy.os, 'pidfd_open', return_value=88, create=True) as opened, \
                            patch.object(deploy.signal, 'pidfd_send_signal', create=True) as send, \
                            patch.object(deploy.os, 'close') as close, patch.object(deploy.os, 'kill') as kill:
                        (opened if stage == 'open' else send).side_effect = OSError(code, 'synthetic')
                        with self.assertRaises(OSError):
                            self.deployment.send(self.identity, 15)
                    kill.assert_not_called()
                    self.assertEqual(close.call_count, int(stage == 'send'))

    def test_disappearing_process_during_signalling_is_already_stopped(self):
        for stage in ('open', 'send', 'kill'):
            with self.subTest(stage=stage):
                with patch.object(deploy.os, 'pidfd_open', return_value=88, create=True) as opened, \
                        patch.object(deploy.signal, 'pidfd_send_signal', create=True) as send, \
                        patch.object(deploy.os, 'close') as close, patch.object(deploy.os, 'kill') as kill:
                    if stage == 'kill':
                        opened.side_effect = OSError(errno.ENOSYS, 'unsupported')
                        kill.side_effect = ProcessLookupError()
                    else:
                        (opened if stage == 'open' else send).side_effect = ProcessLookupError()
                    self.deployment.send(self.identity, 15)
                self.assertEqual(kill.call_count, int(stage == 'kill'))
                self.assertEqual(close.call_count, int(stage == 'send'))

    def test_fallback_permission_failure_is_not_reported_as_success(self):
        kill = Mock(side_effect=PermissionError())
        with patch.object(deploy, 'os', SimpleNamespace(kill=kill)), \
                patch.object(deploy, 'signal', SimpleNamespace()):
            with self.assertRaises(PermissionError):
                self.deployment.send(self.identity, 15)

    def test_nonpositive_or_special_pids_are_rejected(self):
        for pid in (-1, 0, 1, '4321', True):
            with self.subTest(pid=pid), patch.object(deploy.os, 'kill') as kill:
                with self.assertRaisesRegex(RuntimeError, 'invalid_api_process_identity'):
                    self.deployment.send({**self.identity, 'pid': pid}, 15)
                kill.assert_not_called()

    def test_restart_without_pidfd_stops_old_process_before_starting(self):
        (self.root / '.env').write_text('# synthetic only\n')
        running = [self.identity]
        self.deployment.identity = Mock(side_effect=lambda pid: running[0] if running else None)
        self.deployment.processes = Mock(side_effect=lambda: list(running))
        self.deployment.systemd_managed = Mock(return_value=False)
        self.deployment.lock = Mock(return_value=nullcontext())
        self.deployment.settings = Mock()
        self.deployment.sync = Mock()
        self.deployment.activate_runtime = Mock()
        self.deployment.pidfile.write_text('synthetic old identity')

        def killed(pid, sig):
            self.assertEqual((pid, sig), (4321, signal.SIGTERM))
            running.clear()

        def started():
            self.assertFalse(running)
            self.assertFalse(self.deployment.pidfile.exists())

        self.deployment.start = Mock(side_effect=started)
        with patch.object(deploy.os, 'pidfd_open', None, create=True), \
                patch.object(deploy.signal, 'pidfd_send_signal', None, create=True), \
                patch.object(deploy.signal, 'SIGKILL', 9, create=True), \
                patch.object(deploy.os, 'kill', side_effect=killed) as kill:
            self.assertEqual(self.deployment.execute('restart'), 0)
        kill.assert_called_once_with(4321, signal.SIGTERM)
        self.deployment.start.assert_called_once()