def test_process_count_max():
    """
    Tests that the _handle_decoded_payload function does not spawn more than the configured amount of processes,
    as per process_count_max.
    """
    with patch("salt.minion.Minion.ctx", MagicMock(return_value={})), patch(
        "salt.utils.process.SignalHandlingProcess.start",
        MagicMock(return_value=True),
    ), patch(
        "salt.utils.process.SignalHandlingProcess.join",
        MagicMock(return_value=True),
    ), patch(
        "salt.utils.minion.running", MagicMock(return_value=[])
    ), patch(
        "salt.ext.tornado.gen.sleep",
        MagicMock(return_value=salt.ext.tornado.concurrent.Future()),
    ):
        process_count_max = 10
        mock_opts = salt.config.DEFAULT_MINION_OPTS.copy()
        mock_opts["__role"] = "minion"
        mock_opts["minion_jid_queue_hwm"] = 100
        mock_opts["process_count_max"] = process_count_max

        io_loop = salt.ext.tornado.ioloop.IOLoop()
        minion = salt.minion.Minion(mock_opts, jid_queue=[], io_loop=io_loop)
        try:

            # mock gen.sleep to throw a special Exception when called, so that we detect it
            class SleepCalledException(Exception):
                """Thrown when sleep is called"""

            salt.ext.tornado.gen.sleep.return_value.set_exception(
                SleepCalledException()
            )

            # up until process_count_max: gen.sleep does not get called, processes are started normally
            for i in range(process_count_max):
                mock_data = {"fun": "foo.bar", "jid": i}
                io_loop.run_sync(
                    lambda data=mock_data: minion._handle_decoded_payload(data)
                )
                assert (
                    salt.utils.process.SignalHandlingProcess.start.call_count == i + 1
                )
                assert len(minion.jid_queue) == i + 1
                salt.utils.minion.running.return_value += [i]

            # above process_count_max: gen.sleep does get called, JIDs are created but no new processes are started
            mock_data = {"fun": "foo.bar", "jid": process_count_max + 1}

            pytest.raises(
                SleepCalledException,
                lambda: io_loop.run_sync(
                    lambda: minion._handle_decoded_payload(mock_data)
                ),
            )
            assert (
                salt.utils.process.SignalHandlingProcess.start.call_count
                == process_count_max
            )
            assert len(minion.jid_queue) == process_count_max + 1
        finally:
            minion.destroy()