diff options
author | Zbigniew Jędrzejewski-Szmek <zbyszek@in.waw.pl> | 2015-10-26 23:38:23 -0400 |
---|---|---|
committer | Zbigniew Jędrzejewski-Szmek <zbyszek@in.waw.pl> | 2015-10-26 23:38:58 -0400 |
commit | a947f32b310005c4cc3e65a39504e0b32fb66bab (patch) | |
tree | 99d665087599dbc5b6e54681399505611fe6b97e /systemd | |
parent | 6baa4b417be73269ba6d596fd793ea39e0a4db18 (diff) | |
download | python-systemd-a947f32b310005c4cc3e65a39504e0b32fb66bab.tar.gz |
tests: add tests for notify()
Diffstat (limited to 'systemd')
-rw-r--r-- | systemd/test/test_daemon.py | 37 |
1 files changed, 36 insertions, 1 deletions
diff --git a/systemd/test/test_daemon.py b/systemd/test/test_daemon.py index af14f94..d3d3d6d 100644 --- a/systemd/test/test_daemon.py +++ b/systemd/test/test_daemon.py @@ -10,7 +10,8 @@ from systemd.daemon import (booted, is_socket_inet, _is_socket_inet, is_socket_unix, _is_socket_unix, is_mq, _is_mq, - listen_fds) + listen_fds, + notify) import pytest @@ -196,3 +197,37 @@ def test_listen_fds_default_unset(): assert listen_fds(False) == [3] assert listen_fds() == [3] assert listen_fds() == [] + +def test_notify_no_socket(): + assert notify('READY=1') == False + assert notify('FDSTORE=1', fds=[]) == False + assert notify('FDSTORE=1', fds=[1,2]) == False + assert notify('FDSTORE=1', pid=os.getpid()) == False + assert notify('FDSTORE=1', pid=os.getpid(), fds=(1,)) == False + +def test_notify_bad_socket(): + os.environ['NOTIFY_SOCKET'] = '/dev/null' + + with pytest.raises(ConnectionRefusedError): + notify('READY=1') + with pytest.raises(ConnectionRefusedError): + notify('FDSTORE=1', fds=[]) + with pytest.raises(ConnectionRefusedError): + notify('FDSTORE=1', fds=[1,2]) + with pytest.raises(ConnectionRefusedError): + notify('FDSTORE=1', pid=os.getpid()) + with pytest.raises(ConnectionRefusedError): + notify('FDSTORE=1', pid=os.getpid(), fds=(1,)) + +def test_notify_with_socket(tmpdir): + path = tmpdir.join('socket').strpath + sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + sock.bind(path) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) + os.environ['NOTIFY_SOCKET'] = path + + assert notify('READY=1') == True + assert notify('FDSTORE=1', fds=[]) == True + assert notify('FDSTORE=1', fds=[1,2]) == True + assert notify('FDSTORE=1', pid=os.getpid()) == True + assert notify('FDSTORE=1', pid=os.getpid(), fds=(1,)) == True |