mirror of
https://github.com/frappe/gunicorn.git
synced 2026-07-01 10:11:30 +08:00
88e8ef0f36
Add a command-line client for the companion control socket so operators do not have to hand-craft JSON. gunicorn.companion.ctl speaks the manager's newline-delimited JSON protocol: status, start, stop, restart, and reread. The socket path comes from --socket or . Exit status is 0 when the manager reports ok, 1 when it reports a failure, and 2 for a usage error or an unreachable socket. Registered as the gunicorn-companion console script. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
#
|
|
# This file is part of gunicorn released under the MIT license.
|
|
# See the NOTICE for more information.
|
|
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from gunicorn.companion import ctl
|
|
|
|
|
|
def test_run_status_prints_and_returns_zero(capsys):
|
|
with mock.patch.object(
|
|
ctl, "send_command", return_value={"ok": True, "companions": []}
|
|
) as send:
|
|
code = ctl.run(["--socket", "/tmp/x.sock", "status"])
|
|
assert code == 0
|
|
send.assert_called_once_with("/tmp/x.sock", {"cmd": "status"})
|
|
assert "ok" in capsys.readouterr().out
|
|
|
|
|
|
def test_run_per_name_command_sends_name():
|
|
with mock.patch.object(
|
|
ctl, "send_command", return_value={"ok": True, "message": "x"}
|
|
) as send:
|
|
code = ctl.run(["--socket", "/tmp/x.sock", "stop", "ticker"])
|
|
assert code == 0
|
|
send.assert_called_once_with("/tmp/x.sock", {"cmd": "stop", "name": "ticker"})
|
|
|
|
|
|
def test_run_failure_response_returns_one():
|
|
with mock.patch.object(
|
|
ctl, "send_command", return_value={"ok": False, "error": "bad"}
|
|
):
|
|
assert ctl.run(["--socket", "/tmp/x.sock", "status"]) == 1
|
|
|
|
|
|
def test_run_per_name_command_requires_name():
|
|
with pytest.raises(SystemExit):
|
|
ctl.run(["--socket", "/tmp/x.sock", "stop"])
|
|
|
|
|
|
def test_run_requires_socket(monkeypatch):
|
|
monkeypatch.delenv("GUNICORN_COMPANION_SOCKET", raising=False)
|
|
with pytest.raises(SystemExit):
|
|
ctl.run(["status"])
|
|
|
|
|
|
def test_run_unreachable_socket_returns_two():
|
|
with mock.patch.object(ctl, "send_command", side_effect=OSError("nope")):
|
|
assert ctl.run(["--socket", "/tmp/x.sock", "status"]) == 2
|
|
|
|
|
|
def test_send_command_round_trip():
|
|
client = mock.Mock()
|
|
client.recv.side_effect = [b'{"ok": true}\n']
|
|
with mock.patch("socket.socket", return_value=client):
|
|
result = ctl.send_command("/tmp/x.sock", {"cmd": "status"})
|
|
client.connect.assert_called_once_with("/tmp/x.sock")
|
|
assert client.sendall.call_args.args[0] == b'{"cmd": "status"}\n'
|
|
assert result == {"ok": True}
|