|
10 | 10 | is_socket_inet, _is_socket_inet, |
11 | 11 | is_socket_unix, _is_socket_unix, |
12 | 12 | is_mq, _is_mq, |
13 | | - listen_fds) |
| 13 | + listen_fds, |
| 14 | + notify) |
14 | 15 |
|
15 | 16 | import pytest |
16 | 17 |
|
@@ -196,3 +197,37 @@ def test_listen_fds_default_unset(): |
196 | 197 | assert listen_fds(False) == [3] |
197 | 198 | assert listen_fds() == [3] |
198 | 199 | assert listen_fds() == [] |
| 200 | + |
| 201 | +def test_notify_no_socket(): |
| 202 | + assert notify('READY=1') == False |
| 203 | + assert notify('FDSTORE=1', fds=[]) == False |
| 204 | + assert notify('FDSTORE=1', fds=[1,2]) == False |
| 205 | + assert notify('FDSTORE=1', pid=os.getpid()) == False |
| 206 | + assert notify('FDSTORE=1', pid=os.getpid(), fds=(1,)) == False |
| 207 | + |
| 208 | +def test_notify_bad_socket(): |
| 209 | + os.environ['NOTIFY_SOCKET'] = '/dev/null' |
| 210 | + |
| 211 | + with pytest.raises(ConnectionRefusedError): |
| 212 | + notify('READY=1') |
| 213 | + with pytest.raises(ConnectionRefusedError): |
| 214 | + notify('FDSTORE=1', fds=[]) |
| 215 | + with pytest.raises(ConnectionRefusedError): |
| 216 | + notify('FDSTORE=1', fds=[1,2]) |
| 217 | + with pytest.raises(ConnectionRefusedError): |
| 218 | + notify('FDSTORE=1', pid=os.getpid()) |
| 219 | + with pytest.raises(ConnectionRefusedError): |
| 220 | + notify('FDSTORE=1', pid=os.getpid(), fds=(1,)) |
| 221 | + |
| 222 | +def test_notify_with_socket(tmpdir): |
| 223 | + path = tmpdir.join('socket').strpath |
| 224 | + sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) |
| 225 | + sock.bind(path) |
| 226 | + sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) |
| 227 | + os.environ['NOTIFY_SOCKET'] = path |
| 228 | + |
| 229 | + assert notify('READY=1') == True |
| 230 | + assert notify('FDSTORE=1', fds=[]) == True |
| 231 | + assert notify('FDSTORE=1', fds=[1,2]) == True |
| 232 | + assert notify('FDSTORE=1', pid=os.getpid()) == True |
| 233 | + assert notify('FDSTORE=1', pid=os.getpid(), fds=(1,)) == True |
0 commit comments