asda?‰PNG  IHDR ? f ??C1 sRGB ??é gAMA ±? üa pHYs ? ??o¨d GIDATx^íüL”÷e÷Y?a?("Bh?_ò???¢§?q5k?*:t0A-o??¥]VkJ¢M??f?±8\k2íll£1]q?ù???T test_login.py000064400000002106151030026100007250 0ustar00from __future__ import print_function import select import contextlib import errno from systemd import login import pytest @contextlib.contextmanager def skip_oserror(code): try: yield except (OSError, IOError) as e: if e.errno == code: pytest.skip() raise def test_seats(): # just check that we get some sequence back with skip_oserror(errno.ENOENT): seats = login.seats() assert len(seats) >= 0 def test_sessions(): with skip_oserror(errno.ENOENT): sessions = login.sessions() assert len(sessions) >= 0 def test_machine_names(): with skip_oserror(errno.ENOENT): machine_names = login.machine_names() assert len(machine_names) >= 0 def test_uids(): with skip_oserror(errno.ENOENT): uids = login.uids() assert len(uids) >= 0 def test_monitor(): p = select.poll() with skip_oserror(errno.ENOENT): m = login.Monitor("machine") p.register(m, m.get_events()) login.machine_names() p.poll(1) login.machine_names() test_daemon.py000064400000024032151030026100007405 0ustar00import sys import os import posix import socket import contextlib import errno from systemd.daemon import (booted, is_fifo, _is_fifo, is_socket, _is_socket, is_socket_inet, _is_socket_inet, is_socket_unix, _is_socket_unix, is_socket_sockaddr, _is_socket_sockaddr, is_mq, _is_mq, listen_fds, notify) import pytest @contextlib.contextmanager def skip_enosys(): try: yield except OSError as e: if e.errno == errno.ENOSYS: pytest.skip() raise @contextlib.contextmanager def closing_socketpair(family): pair = socket.socketpair(family) try: yield pair finally: pair[0].close() pair[1].close() def test_booted(): if os.path.exists('/run/systemd/system'): # assume we are running under systemd assert booted() else: # don't assume anything assert booted() in {False, True} def test__is_fifo(tmpdir): path = tmpdir.join('test.fifo').strpath posix.mkfifo(path) fd = os.open(path, os.O_RDONLY|os.O_NONBLOCK) assert _is_fifo(fd, None) assert _is_fifo(fd, path) def test__is_fifo_file(tmpdir): file = tmpdir.join('test.fifo') file.write('boo') path = file.strpath fd = os.open(path, os.O_RDONLY|os.O_NONBLOCK) assert not _is_fifo(fd, None) assert not _is_fifo(fd, path) def test__is_fifo_bad_fd(tmpdir): path = tmpdir.join('test.fifo').strpath with pytest.raises(OSError): assert not _is_fifo(-1, None) with pytest.raises(OSError): assert not _is_fifo(-1, path) def test_is_fifo(tmpdir): path = tmpdir.join('test.fifo').strpath posix.mkfifo(path) fd = os.open(path, os.O_RDONLY|os.O_NONBLOCK) file = os.fdopen(fd, 'r') assert is_fifo(file, None) assert is_fifo(file, path) assert is_fifo(fd, None) assert is_fifo(fd, path) def test_is_fifo_file(tmpdir): file = tmpdir.join('test.fifo') file.write('boo') path = file.strpath fd = os.open(path, os.O_RDONLY|os.O_NONBLOCK) file = os.fdopen(fd, 'r') assert not is_fifo(file, None) assert not is_fifo(file, path) assert not is_fifo(fd, None) assert not is_fifo(fd, path) def test_is_fifo_bad_fd(tmpdir): path = tmpdir.join('test.fifo').strpath with pytest.raises(OSError): assert not is_fifo(-1, None) with pytest.raises(OSError): assert not is_fifo(-1, path) def is_mq_wrapper(arg): try: return is_mq(arg) except OSError as error: # systemd < 227 compatibility assert error.errno == errno.EBADF return False def _is_mq_wrapper(arg): try: return _is_mq(arg) except OSError as error: # systemd < 227 compatibility assert error.errno == errno.EBADF return False def test_no_mismatch(): with closing_socketpair(socket.AF_UNIX) as pair: for sock in pair: assert not is_fifo(sock) assert not is_mq_wrapper(sock) assert not is_socket_inet(sock) with skip_enosys(): assert not is_socket_sockaddr(sock, '127.0.0.1:2000') fd = sock.fileno() assert not is_fifo(fd) assert not is_mq_wrapper(fd) assert not is_socket_inet(fd) with skip_enosys(): assert not is_socket_sockaddr(fd, '127.0.0.1:2000') assert not _is_fifo(fd) assert not _is_mq_wrapper(fd) assert not _is_socket_inet(fd) with skip_enosys(): assert not _is_socket_sockaddr(fd, '127.0.0.1:2000') def test_is_socket(): with closing_socketpair(socket.AF_UNIX) as pair: for sock in pair: for arg in (sock, sock.fileno()): assert is_socket(arg) assert is_socket(arg, socket.AF_UNIX) assert not is_socket(arg, socket.AF_INET) assert is_socket(arg, socket.AF_UNIX, socket.SOCK_STREAM) assert not is_socket(arg, socket.AF_INET, socket.SOCK_DGRAM) with skip_enosys(): assert not is_socket_sockaddr(arg, '8.8.8.8:2000', socket.SOCK_DGRAM, 0, 0) assert _is_socket(arg) assert _is_socket(arg, socket.AF_UNIX) assert not _is_socket(arg, socket.AF_INET) assert _is_socket(arg, socket.AF_UNIX, socket.SOCK_STREAM) assert not _is_socket(arg, socket.AF_INET, socket.SOCK_DGRAM) with skip_enosys(): assert not _is_socket_sockaddr(arg, '8.8.8.8:2000', socket.SOCK_DGRAM, 0, 0) def test_is_socket_sockaddr(): with contextlib.closing(socket.socket(socket.AF_INET)) as sock: sock.bind(('127.0.0.1', 0)) addr, port = sock.getsockname() port = ':{}'.format(port) for listening in (0, 1): for arg in (sock, sock.fileno()): with skip_enosys(): assert is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_STREAM) with skip_enosys(): assert is_socket_sockaddr(arg, '127.0.0.1' + port, socket.SOCK_STREAM) with skip_enosys(): assert is_socket_sockaddr(arg, '127.0.0.1' + port, listening=listening) with skip_enosys(): assert is_socket_sockaddr(arg, '127.0.0.1' + port, listening=-1) with skip_enosys(): assert not is_socket_sockaddr(arg, '127.0.0.1' + port, listening=not listening) with pytest.raises(ValueError): is_socket_sockaddr(arg, '127.0.0.1', flowinfo=123456) with skip_enosys(): assert not is_socket_sockaddr(arg, '129.168.11.11:23', socket.SOCK_STREAM) with skip_enosys(): assert not is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_DGRAM) with pytest.raises(ValueError): _is_socket_sockaddr(arg, '127.0.0.1', 0, 123456) with skip_enosys(): assert not _is_socket_sockaddr(arg, '129.168.11.11:23', socket.SOCK_STREAM) with skip_enosys(): assert not _is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_DGRAM) sock.listen(11) def test__is_socket(): with closing_socketpair(socket.AF_UNIX) as pair: for sock in pair: fd = sock.fileno() assert _is_socket(fd) assert _is_socket(fd, socket.AF_UNIX) assert not _is_socket(fd, socket.AF_INET) assert _is_socket(fd, socket.AF_UNIX, socket.SOCK_STREAM) assert not _is_socket(fd, socket.AF_INET, socket.SOCK_DGRAM) assert _is_socket(fd) assert _is_socket(fd, socket.AF_UNIX) assert not _is_socket(fd, socket.AF_INET) assert _is_socket(fd, socket.AF_UNIX, socket.SOCK_STREAM) assert not _is_socket(fd, socket.AF_INET, socket.SOCK_DGRAM) def test_is_socket_unix(): with closing_socketpair(socket.AF_UNIX) as pair: for sock in pair: for arg in (sock, sock.fileno()): assert is_socket_unix(arg) assert not is_socket_unix(arg, path="/no/such/path") assert is_socket_unix(arg, socket.SOCK_STREAM) assert not is_socket_unix(arg, socket.SOCK_DGRAM) def test__is_socket_unix(): with closing_socketpair(socket.AF_UNIX) as pair: for sock in pair: fd = sock.fileno() assert _is_socket_unix(fd) assert not _is_socket_unix(fd, 0, -1, "/no/such/path") assert _is_socket_unix(fd, socket.SOCK_STREAM) assert not _is_socket_unix(fd, socket.SOCK_DGRAM) def test_listen_fds_no_fds(): # make sure we have no fds to listen to os.unsetenv('LISTEN_FDS') os.unsetenv('LISTEN_PID') assert listen_fds() == [] assert listen_fds(True) == [] assert listen_fds(False) == [] def test_listen_fds(): os.environ['LISTEN_FDS'] = '3' os.environ['LISTEN_PID'] = str(os.getpid()) assert listen_fds(False) == [3, 4, 5] assert listen_fds(True) == [3, 4, 5] assert listen_fds() == [] def test_listen_fds_default_unset(): os.environ['LISTEN_FDS'] = '1' os.environ['LISTEN_PID'] = str(os.getpid()) assert listen_fds(False) == [3] assert listen_fds() == [3] assert listen_fds() == [] def test_notify_no_socket(): assert notify('READY=1') is False with skip_enosys(): assert notify('FDSTORE=1', fds=[]) is False assert notify('FDSTORE=1', fds=[1, 2]) is False assert notify('FDSTORE=1', pid=os.getpid()) is False assert notify('FDSTORE=1', pid=os.getpid(), fds=(1,)) is False if sys.version_info >= (3,): connection_error = ConnectionRefusedError else: connection_error = OSError def test_notify_bad_socket(): os.environ['NOTIFY_SOCKET'] = '/dev/null' with pytest.raises(connection_error): notify('READY=1') with pytest.raises(connection_error): with skip_enosys(): notify('FDSTORE=1', fds=[]) with pytest.raises(connection_error): notify('FDSTORE=1', fds=[1, 2]) with pytest.raises(connection_error): notify('FDSTORE=1', pid=os.getpid()) with pytest.raises(connection_error): 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) try: sock.bind(path) except socket.error as e: pytest.xfail('failed to bind socket (%s)' % e) # SO_PASSCRED is not defined in python2.7 SO_PASSCRED = getattr(socket, 'SO_PASSCRED', 16) sock.setsockopt(socket.SOL_SOCKET, SO_PASSCRED, 1) os.environ['NOTIFY_SOCKET'] = path assert notify('READY=1') with skip_enosys(): assert notify('FDSTORE=1', fds=[]) assert notify('FDSTORE=1', fds=[1, 2]) assert notify('FDSTORE=1', pid=os.getpid()) assert notify('FDSTORE=1', pid=os.getpid(), fds=(1,)) __pycache__/test_daemon.cpython-36.opt-1.pyc000064400000014443151030026100014635 0ustar003 X(@sZddlZddlZddlZddlZddlZddlZddlmZmZm Z m Z m Z m Z m Z mZmZmZmZmZmZmZmZddlZejddZejddZddZd d Zd d Zd dZddZddZddZ ddZ!ddZ"ddZ#ddZ$ddZ%dd Z&d!d"Z'd#d$Z(d%d&Z)d'd(Z*d)d*Z+d+d,Z,ej-d2krBe.Z/ne0Z/d.d/Z1d0d1Z2dS)3N)bootedis_fifo_is_fifo is_socket _is_socketis_socket_inet_is_socket_inetis_socket_unix_is_socket_unixis_socket_sockaddr_is_socket_sockaddris_mq_is_mq listen_fdsnotifyccsHy dVWn8tk rB}z|jtjkr0tjWYdd}~XnXdS)N)OSErrorerrnoZENOSYSpytestskip)er#/usr/lib64/python3.6/test_daemon.py skip_enosyss   rc cs4tj|}z |VWd|dj|djXdS)Nr)socketZ socketpairclose)Zfamilypairrrrclosing_socketpairs    rcCstjjdrndS)Nz/run/systemd/system)ospathexistsrrrr test_booted&s r!cCs.|jdj}tj|tj|tjtjB}dS)Nz test.fifo)joinstrpathposixmkfiforopenO_RDONLY O_NONBLOCK)tmpdirrfdrrr test__is_fifo.s  r+cCs2|jd}|jd|j}tj|tjtjB}dS)Nz test.fifoboo)r"writer#rr&r'r()r)filerr*rrrtest__is_fifo_file6s   r/cCs<|jdj}tjtWdQRXtjtWdQRXdS)Nz test.fifo)r"r#rraisesr)r)rrrrtest__is_fifo_bad_fd?s     r1cCs:|jdj}tj|tj|tjtjB}tj|d}dS)Nz test.fifor) r"r#r$r%rr&r'r(fdopen)r)rr*r.rrr test_is_fifoHs    r4cCs>|jd}|jd|j}tj|tjtjB}tj|d}dS)Nz test.fifor,r2)r"r-r#rr&r'r(r3)r)r.rr*rrrtest_is_fifo_fileSs    r5cCs<|jdj}tjtWdQRXtjtWdQRXdS)Nz test.fifo)r"r#rr0r)r)rrrrtest_is_fifo_bad_fd_s     r6cCs.yt|Stk r(}zdSd}~XnXdS)NF)r r)argerrorrrr is_mq_wrapperhsr9cCs.yt|Stk r(}zdSd}~XnXdS)NF)rr)r7r8rrr_is_mq_wrapperpsr:cCsfttjR}xJ|D]B}tWdQRX|j}tWdQRXtWdQRXqWWdQRXdS)N)rrAF_UNIXrfileno)rsockr*rrrtest_no_mismatchxs    r>cCsbttjN}xF|D]>}x&||jfD]}tWdQRXq$WtWdQRXqWWdQRXdS)N)rrr;r<r)rr=r7rrrtest_is_sockets  r?cCCs`tjtjtjB}|jd|j\}}dj|}xd D]}x||jfD]}tWdQRXtWdQRXtWdQRXtWdQRXtWdQRXt j t t |dddWdQRXtWdQRXtWdQRXqRWt j t t |dddWdQRXtWdQRXtWdQRX|jdq>WWdQRXdS) N 127.0.0.1rz:{}ri@)Zflowinfo )r@r)rr) contextlibclosingrZAF_INETbindZ getsocknameformatr<rrr0 ValueErrorr r Zlisten)r=ZaddrZportZ listeningr7rrrtest_is_socket_sockaddrs:             rGc Cs0ttj}x|D] }|j}qWWdQRXdS)N)rrr;r<)rr=r*rrrtest__is_sockets   rHc Cs>ttj*}x"|D]}x||jfD]}q$WqWWdQRXdS)N)rrr;r<)rr=r7rrrtest_is_socket_unixs  rIc Cs0ttj}x|D] }|j}qWWdQRXdS)N)rrr;r<)rr=r*rrrtest__is_socket_unixs  rJcCstjdtjddS)N LISTEN_FDS LISTEN_PID)runsetenvrrrrtest_listen_fds_no_fdss  rNcCs dtjd<ttjtjd<dS)N3rKrL)renvironstrgetpidrrrrtest_listen_fdss rScCs dtjd<ttjtjd<dS)N1rKrL)rrPrQrRrrrrtest_listen_fds_default_unsets rUc CstWdQRXdS)N)rrrrrtest_notify_no_sockets rVc/Csdtjd<tjttdWdQRXtjt$ttdgdWdQRXWdQRXtjttdddgdWdQRXtjttdtjdWdQRXtjttdtjd d WdQRXdS) Nz /dev/null NOTIFY_SOCKETzREADY=1z FDSTORE=1)fdsr)pid)r[rY)r)rrPrr0connection_errorrrrRrrrrtest_notify_bad_sockets       r]cCs|jdj}tjtjtj}y|j|Wn2tjk r\}ztjd|WYdd}~XnXt tdd}|j tj |d|t j d<tWdQRXdS)Nrzfailed to bind socket (%s) SO_PASSCREDrrX)r"r#rr;Z SOCK_DGRAMrDr8rZxfailgetattrZ setsockoptZ SOL_SOCKETrrPr)r)rr=rr^rrrtest_notify_with_sockets     ra)rW)3sysrr$rrBrZsystemd.daemonrrrrrrrr r r r r rrrrcontextmanagerrrr!r+r/r1r4r5r6r9r:r>r?rGrHrIrJrNrSrUrV version_infoConnectionRefusedErrorr\rr]rarrrrsBD        &    __pycache__/test_login.cpython-36.opt-1.pyc000064400000002671151030026100014502 0ustar003 XF@srddlmZddlZddlZddlZddlmZddlZejddZ ddZ dd Z d d Z d d Z ddZdS))print_functionN)loginccsJy dVWn:ttfk rD}z|j|kr2tjWYdd}~XnXdS)N)OSErrorIOErrorerrnopytestskip)codeer "/usr/lib64/python3.6/test_login.py skip_oserror s   r c Cs"ttjtj}WdQRXdS)N)r rENOENTrseats)rr r r test_seatss rc Cs"ttjtj}WdQRXdS)N)r rrrsessions)rr r r test_sessionss rc Cs"ttjtj}WdQRXdS)N)r rrr machine_names)rr r r test_machine_namess rc Cs"ttjtj}WdQRXdS)N)r rrruids)rr r r test_uids#s rc CsVtj}ttj:tjd}|j||jtj |jdtj WdQRXdS)Nmachine) selectZpollr rrrZMonitorregisterZ get_eventsr)pmr r r test_monitor(s   r)Z __future__rr contextlibrZsystemdrrcontextmanagerr rrrrrr r r r s   __pycache__/test_daemon.cpython-36.pyc000064400000021105151030026100013667 0ustar003 X(@sZddlZddlZddlZddlZddlZddlZddlmZmZm Z m Z m Z m Z m Z mZmZmZmZmZmZmZmZddlZejddZejddZddZd d Zd d Zd dZddZddZddZ ddZ!ddZ"ddZ#ddZ$ddZ%dd Z&d!d"Z'd#d$Z(d%d&Z)d'd(Z*d)d*Z+d+d,Z,ej-d2krBe.Z/ne0Z/d.d/Z1d0d1Z2dS)3N)bootedis_fifo_is_fifo is_socket _is_socketis_socket_inet_is_socket_inetis_socket_unix_is_socket_unixis_socket_sockaddr_is_socket_sockaddris_mq_is_mq listen_fdsnotifyccsHy dVWn8tk rB}z|jtjkr0tjWYdd}~XnXdS)N)OSErrorerrnoZENOSYSpytestskip)er#/usr/lib64/python3.6/test_daemon.py skip_enosyss   rc cs4tj|}z |VWd|dj|djXdS)Nr)socketZ socketpairclose)Zfamilypairrrrclosing_socketpairs    rcCs*tjjdrts&tntdks&tdS)Nz/run/systemd/systemFT>FT)ospathexistsrAssertionErrorrrrr test_booted&s  r"cCsJ|jdj}tj|tj|tjtjB}t|ds8t t||sFt dS)Nz test.fifo) joinstrpathposixmkfiforopenO_RDONLY O_NONBLOCKrr!)tmpdirrfdrrr test__is_fifo.s   r,cCsR|jd}|jd|j}tj|tjtjB}t|d s>tt|| sNtdS)Nz test.fifoboo) r#writer$rr'r(r)rr!)r*filerr+rrrtest__is_fifo_file6s   r0cCs\|jdj}tjttdd s(tWdQRXtjttd| sNtWdQRXdS)Nz test.fiforr1)r#r$rraisesrrr!)r*rrrrtest__is_fifo_bad_fd?s    r3cCsr|jdj}tj|tj|tjtjB}tj|d}t |dsDt t ||sRt t |ds`t t ||snt dS)Nz test.fifor) r#r$r%r&rr'r(r)fdopenrr!)r*rr+r/rrr test_is_fifoHs   r6cCs~|jd}|jd|j}tj|tjtjB}tj|d}t|d sJt t|| sZt t|d sjt t|| szt dS)Nz test.fifor-r4) r#r.r$rr'r(r)r5rr!)r*r/rr+rrrtest_is_fifo_fileSs   r7cCs\|jdj}tjttdd s(tWdQRXtjttd| sNtWdQRXdS)Nz test.fiforr1r1)r#r$rr2rrr!)r*rrrrtest_is_fifo_bad_fd_s    r8cCs>yt|Stk r8}z|jtjks*tdSd}~XnXdS)NF)r rrEBADFr!)argerrorrrr is_mq_wrapperhs r<cCs>yt|Stk r8}z|jtjks*tdSd}~XnXdS)NF)rrrr9r!)r:r;rrr_is_mq_wrapperps r=c Csttj}x|D]}t| s&tt| s4tt| sBttt|d sZtWdQRX|j }t| sztt| stt| sttt|d stWdQRXt | stt | stt | sttt |d stWdQRXqWWdQRXdS)Nz127.0.0.1:2000)rrAF_UNIXrr!r<rrr filenorr=rr )rsockr+rrrtest_no_mismatchxs$ rAcCsLttj6}x,|D]"}x||jfD]}t|s:tt|tjsJtt|tj s\tt|tjtjsptt|tjtj stt t |dtjdd stWdQRXq*Wt |stt |tjstt |tj stt |tjtjstt |tjtj stt t |dtjdd s0tWdQRXqWWdQRXdS)Nz 8.8.8.8:2000r) rrr>r?rr!AF_INET SOCK_STREAM SOCK_DGRAMrr rr )rr@r:rrrtest_is_sockets" & rEcECs,tjtjtj}|jd |j\}}dj|}xd D]}xT||jfD]B}tt |dtj stt WdQRXtt |d|tj st WdQRXtt |d||dst WdQRXtt |d|d dst WdQRXt"t |d|| d st WdQRXt j tt |dddWdQRXtt |dtj  sft WdQRXtt |dtj st WdQRXqTWt j tt|dddWdQRXtt|dtj  st WdQRXtt|dtj st WdQRX|jd q>WWdQRXdS) N 127.0.0.1rz:{}r) listeningi@)Zflowinfoz129.168.11.11:23 )rFr)rrr1) contextlibclosingrrBbindZ getsocknameformatr?rr rCr!rr2 ValueErrorrDr Zlisten)r@ZaddrZportrGr:rrrtest_is_socket_sockaddrs:      &  $   rNc Csttj}x|D]}|j}t|s*tt|tjs:tt|tj sLtt|tjtjs`tt|tjtj svtt|stt|tjstt|tj stt|tjtjstt|tjtj stqWWdQRXdS)N) rrr>r?rr!rBrCrD)rr@r+rrrtest__is_sockets    rOc Cs~ttjj}xb|D]Z}xT||jfD]D}t|s4tt|dd sFtt|tjsVtt|tj s$tq$WqWWdQRXdS)Nz /no/such/path)r)rrr>r?r r!rCrD)rr@r:rrrtest_is_socket_unixs   rPc Csrttj^}xV|D]N}|j}t|s*tt|ddd s>tt|tjsNtt|tj stqWWdQRXdS)Nrrz /no/such/pathr1)rrr>r?r r!rCrD)rr@r+rrrtest__is_socket_unixs   rQcCsFtjdtjdtgks"ttdgks2ttdgksBtdS)N LISTEN_FDS LISTEN_PIDTF)runsetenvrr!rrrrtest_listen_fds_no_fdss   rUcCsZdtjd<ttjtjd<tddddgks2ttddddgksHttgksVtdS) N3rRrSFT)renvironstrgetpidrr!rrrrtest_listen_fdss  r]cCsPdtjd<ttjtjd<tddgks.ttdgks>ttgksLtdS)N1rRrSFrW)rrZr[r\rr!rrrrtest_listen_fds_default_unsets  r_c Cstddkstttdgddks,tWdQRXtdddgddksNttdtjddksfttdtjd ddkstdS) NzREADY=1Fz FDSTORE=1)fdsr)pid)rbr`)r)rr!rrr\rrrrtest_notify_no_sockets rcrWc/Csdtjd<tjttdWdQRXtjt$ttdgdWdQRXWdQRXtjttdddgdWdQRXtjttdtjdWdQRXtjttdtjd d WdQRXdS) Nz /dev/null NOTIFY_SOCKETzREADY=1z FDSTORE=1)r`rra)rb)rbr`)r)rrZrr2connection_errorrrr\rrrrtest_notify_bad_sockets       rfcCs|jdj}tjtjtj}y|j|Wn2tjk r\}ztjd|WYdd}~XnXt tdd}|j tj |d|t j d<tdstttdgd stWdQRXtddd gd sttdt jd sttdt jd d stdS)Nrzfailed to bind socket (%s) SO_PASSCREDrrdzREADY=1z FDSTORE=1)r`ra)rb)rbr`)r)r#r$rr>rDrKr;rZxfailgetattrZ setsockoptZ SOL_SOCKETrrZrr!rr\)r*rr@rrgrrrtest_notify_with_sockets     rj)rW)3sysrr%rrIrZsystemd.daemonrrrrrrrr r r r r rrrrcontextmanagerrrr"r,r0r3r6r7r8r<r=rArErNrOrPrQrUr]r_rc version_infoConnectionRefusedErrorrerrfrjrrrrsBD        &    __pycache__/test_journal.cpython-36.pyc000064400000025312151030026100014102 0ustar003 X]'@sddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z m Z ddlmZddlZejdZejdZGdddZejd d Zejd d Zd dZddZddZddZddZddZddZddZddZ dd Z!d!d"Z"d#d$Z#d%d&Z$d'd(Z%d)d*Z&d+d,Z'd-d.Z(d/d0Z)d1d2Z*d3d4Z+d5d6Z,d7d8Z-d9d:Z.d;d<Z/d=d>Z0d?d@Z1dAdBZ2dS)C)print_functionN)journalid128) _make_lineZ 8441372f8dca4ca98694a6091fd8519fZ 8441370000000000000000001fd85000c@seZdZddZdddZdS) MockSendercCs g|_dS)N)buf)selfr $/usr/lib64/python3.6/test_journal.py__init__szMockSender.__init__Nc Ksd|g}|dk r,t|d|}|jd|||koH|koHdknrltjddddd\}}}|dk r|jd||dk r|jd j||dk r|jd ||jd d |jD|jj|dS) NzMESSAGE=hexz MESSAGE_ID=)limitrz CODE_FILE=zCODE_LINE={:d}z CODE_FUNC=css|]\}}t||VqdS)N)r).0keyvalr r r )sz"MockSender.send..)getattrappend _traceback extract_stackformatextenditemsr) rZMESSAGE MESSAGE_IDZ CODE_FILEZ CODE_LINEZ CODE_FUNCkwargsargsidr r r sends  "zMockSender.send)NNNN)__name__ __module__ __qualname__r rr r r r rsrccsJy dVWn:ttfk rD}z|j|kr2tjWYdd}~XnXdS)N)OSErrorIOErrorerrnopytestskip)codeer r r skip_oserror,s   r*c cs,y dVWntk r&tjYnXdS)N) ValueErrorr&r'r r r r skip_valueerror5s r,cCs|tjj}|tjtjkst|tjtjks0t|tjdtjksHt|tjdtjks`t|tj dtjksxt|tj tjkst|tj dtj kst|tj dtj kst|tj tj kst|tj dtj kst|tj dtj kst|tj tj kst|tj dtjks2t|tjtjksHt|tjtjks^t|tjdtjksxtdS)N)rJournalHandlerZ map_priorityloggingZNOTSETZ LOG_DEBUGAssertionErrorDEBUGZLOG_INFOINFOZ LOG_WARNINGZWARNZLOG_ERRZERRORZLOG_CRITZFATALZCRITICALZ LOG_ALERT)pr r r test_priorities<s"r4c Cs.ddi}tjttjf|WdQRXdS)Nz X r)r&raisesr+rr.)kwr r r "test_journalhandler_init_exceptionQs r7cCsddd}tjtjf|dS)Nr)XX3)rr.r/r2)r6r r r test_journalhandler_initVs r;cCstjdtjddddd}t}dd|jd}tjtjf|}|j|t|j dksXt d|j d ksjt d |j d ks|t dS) Nz test-loggertestpathr-testrr8)r9r:sender_functionzX=3rzX3=4) r/ LogRecordr2rrrr.emitlenrr0)recordsenderr6handlerr r r test_journalhandler_infoZs rEcCsltjdtjddddd}t}tjtj|jd}|j|t|j dksLt t dd|j dDsht dS) Nz test-loggerr<r-r=)r>css|]}|jd VqdS)z MESSAGE_ID=N) startswith)rmr r r rksz4test_journalhandler_no_message_id..r) r/r?r2rrr.rr@rArr0all)rBrCrDr r r !test_journalhandler_no_message_ides  rIcCsjtjdtjddddd}t}tjtj|jtd}|j|t |j dksNt dtj |j dksft dS)Nz test-loggerr<r-r=)r>rz MESSAGE_ID=r) r/r?r2rrr.rTEST_MIDr@rArr0r )rBrCrDr r r )test_journalhandler_message_id_on_handlerms  rKcCsltjdtjddddd}t}tjtj|jtjd}|j |t |j dksPt dtj|j dksht dS)Nz test-loggerr<r-r=)r>rz MESSAGE_ID=r) r/r?r2rrr.rrJr r@rArr0)rBrCrDr r r -test_journalhandler_message_id_on_handler_hexvs   rLcCsttjdtjddddd}t|jd<t}tjtj|jt d}|j |t |j dksXt dtj|j dkspt dS) Nz test-loggerr<r-r=r)r>rz MESSAGE_ID=r)r/r?r2 TEST_MID2__dict__rrr.rrJr@rArr0r )rBrCrDr r r )test_journalhandler_message_id_on_messages   rOcCsvtjdtjddddd}tj|jd<t}tjtj|j t d}|j |t |j dksZtdtj|j dksrtdS) Nz test-loggerr<r-r=r)r>rz MESSAGE_ID=r)r/r?r2rMr rNrrr.rrJr@rArr0)rBrCrDr r r -test_journalhandler_message_id_on_message_hexs   rPcCsRtj}tjtj}tjtj}tjtj}tjtjtjBtjB}tjd}dS)Nr)rReaderZ LOCAL_ONLYZ RUNTIME_ONLY SYSTEM_ONLY)j1j2j3Zj4Zj5Zj6r r r test_reader_init_flagss    rVcCstjttjtjWdQRXttj|jtjd}WdQRXttj|jtjtjBd}WdQRXtj|jtjtj Bd}dS)N)pathflags) r&r5r+rrQZOS_ROOTr,strpath CURRENT_USERrR)tmpdirrSrTrUr r r test_reader_os_roots r\cCs>tj|jd}tjd|jdtj|jd}tj|jddS)N)rWr)rrQrY)r[rSrTr r r test_reader_init_pathsr]c Cs(tjttjdddWdQRXdS)Nrr-)rW)r&r5r#rrQr r r r test_reader_init_path_invalid_fds r_c Cs(tjttjdddWdQRXdS)Nr)rW)r&r5r#rrQr r r r %test_reader_init_path_nondirectory_fds r`cCstj|jtj}ttjtj|d}WdQRXt |gksBt t tjtj |d}WdQRXt |gkstt tjtj |d}t |gkst dS)N)rW)osopenrYO_RDONLYr*r%ENOSYSrrQlistr0r,ZSYSTEMrZ)r[fdrSrTrUr r r test_reader_init_path_fds rgcCsVtj|jd}||j s tWdQRX|js4ttjtt|WdQRXdS)N)rW) rrQrYclosedr0r&r5r#next)r[jr r r test_reader_as_cms   rkc Cs<tj|jd}| |jtj|jtjjWdQRXdS)N)rW)rrQrYZmessageid_matchrZSD_MESSAGE_JOURNAL_STARTZSD_MESSAGE_JOURNAL_STOPr )r[rjr r r test_reader_messageid_matchs rlc Cs@tj|jd}|$|j|jt|jtjWdQRXdS)N)rW)rrQrYZ this_bootrJr )r[rjr r r test_reader_this_boots  rmc Cs@tj|jd}|$|j|jt|jtjWdQRXdS)N)rW)rrQrYZ this_machinerJr )r[rjr r r test_reader_this_machines  rncCs^tj|jd}|&ttj|jd}WdQRXWdQRXt|tsLt |tksZt dS)N)rWZFOOBAR) rrQrYr*r%rdZ query_unique isinstancesetr0)r[rjansr r r test_reader_query_uniques  rrcCs\tj|jd}|$ttj|j}WdQRXWdQRXt|tsJt |tksXt dS)N)rW) rrQrYr*r%rdZenumerate_fieldsrorpr0)r[rjrqr r r test_reader_enumerate_fieldss  rscCsLtj|jd}|$ttj|j}WdQRXWdQRX|dksHtdS)N)rWF)rrQrYr*r%rdhas_runtime_filesr0)r[rjrqr r r test_reader_has_runtime_filess  rucCsLtj|jd}|$ttj|j}WdQRXWdQRX|dksHtdS)N)rWF)rrQrYr*r%rdrtr0)r[rjrqr r r test_reader_has_persistent_filess  rvcCsPdddi}tj|j|d}|jdd}|dks4t|jdd}|dksLtdS) NZxxxcSsdS)Nyyyr )argr r r sz(test_reader_converters..)rW converterssabcrwZzzzs)rrQrYZ_convert_fieldr0)r[rzrjrr r r test_reader_converterss     r{cCs`ddddd}tj|j|d}|jddddgdd gd }|d dd d gdd gd ks\tdS) NcSsdS)Nrwr )rxr r r ry sz+test_reader_convert_entry..cSsdS)NYYYr )rxr r r ry s)x1x2)rWrzsabcssdefs)r}Zy1r~Zy2rwr|)rrQrYZ_convert_entryr0)r[rzrjrr r r test_reader_convert_entry s rcCsFtj|jd}tj}|j||jdtjddd}|j|dS)N)rWi90ir8)rrQrYtimeZ seek_realtimedatetime)r[rjZnowZlong_agor r r test_seek_realtimes   rc CsFttjtjd}WdQRX|jd}|dks6ttd|ddS)Nztest_journal.pyz message... zprinted message...)file)rN)r*r%ENOENTrstreamwriter0print)rresr r r test_journal_stream"s    r)3Z __future__r contextlibrr%r/rarZuuid tracebackrZsystemdrrZsystemd.journalrr&ZUUIDrJrMrcontextmanagerr*r,r4r7r;rErIrKrLrOrPrVr\r]r_r`rgrkrlrmrnrrrsrurvr{rrrr r r r sV              __pycache__/test_journal.cpython-36.opt-1.pyc000064400000022277151030026100015050 0ustar003 X]'@sddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z m Z ddlmZddlZejdZejdZGdddZejd d Zejd d Zd dZddZddZddZddZddZddZddZddZ dd Z!d!d"Z"d#d$Z#d%d&Z$d'd(Z%d)d*Z&d+d,Z'd-d.Z(d/d0Z)d1d2Z*d3d4Z+d5d6Z,d7d8Z-d9d:Z.d;d<Z/d=d>Z0d?d@Z1dAdBZ2dS)C)print_functionN)journalid128) _make_lineZ 8441372f8dca4ca98694a6091fd8519fZ 8441370000000000000000001fd85000c@seZdZddZdddZdS) MockSendercCs g|_dS)N)buf)selfr $/usr/lib64/python3.6/test_journal.py__init__szMockSender.__init__Nc Ksd|g}|dk r,t|d|}|jd|||koH|koHdknrltjddddd\}}}|dk r|jd||dk r|jd j||dk r|jd ||jd d |jD|jj|dS) NzMESSAGE=hexz MESSAGE_ID=)limitrz CODE_FILE=zCODE_LINE={:d}z CODE_FUNC=css|]\}}t||VqdS)N)r).0keyvalr r r )sz"MockSender.send..)getattrappend _traceback extract_stackformatextenditemsr) rZMESSAGE MESSAGE_IDZ CODE_FILEZ CODE_LINEZ CODE_FUNCkwargsargsidr r r sends  "zMockSender.send)NNNN)__name__ __module__ __qualname__r rr r r r rsrccsJy dVWn:ttfk rD}z|j|kr2tjWYdd}~XnXdS)N)OSErrorIOErrorerrnopytestskip)codeer r r skip_oserror,s   r*c cs,y dVWntk r&tjYnXdS)N) ValueErrorr&r'r r r r skip_valueerror5s r,cCs tjj}dS)N)rJournalHandlerZ map_priority)pr r r test_priorities<sr/c Cs.ddi}tjttjf|WdQRXdS)Nz X r)r&raisesr+rr-)kwr r r "test_journalhandler_init_exceptionQs r2cCsddd}tjtjf|dS)Nr)XX3)rr-loggingINFO)r1r r r test_journalhandler_initVs r8cCsJtjdtjddddd}t}dd|jd}tjtjf|}|j|dS)Nz test-loggertestpathtestrr3)r4r5sender_function)r6 LogRecordr7rrrr-emit)recordsenderr1handlerr r r test_journalhandler_infoZs  rBcCs>tjdtjddddd}t}tjtj|jd}|j|dS)Nz test-loggerr9r:r;)r<)r6r=r7rrr-rr>)r?r@rAr r r !test_journalhandler_no_message_ides  rCcCs@tjdtjddddd}t}tjtj|jtd}|j|dS)Nz test-loggerr9r:r;)r<r) r6r=r7rrr-rTEST_MIDr>)r?r@rAr r r )test_journalhandler_message_id_on_handlerms   rEcCsBtjdtjddddd}t}tjtj|jtjd}|j |dS)Nz test-loggerr9r:r;)r<r) r6r=r7rrr-rrDr r>)r?r@rAr r r -test_journalhandler_message_id_on_handler_hexvs    rFcCsJtjdtjddddd}t|jd<t}tjtj|jt d}|j |dS)Nz test-loggerr9r:r;r)r<r) r6r=r7 TEST_MID2__dict__rrr-rrDr>)r?r@rAr r r )test_journalhandler_message_id_on_messages   rIcCsLtjdtjddddd}tj|jd<t}tjtj|j t d}|j |dS)Nz test-loggerr9r:r;r)r<r) r6r=r7rGr rHrrr-rrDr>)r?r@rAr r r -test_journalhandler_message_id_on_message_hexs   rJcCsRtj}tjtj}tjtj}tjtj}tjtjtjBtjB}tjd}dS)Nr)rReaderZ LOCAL_ONLYZ RUNTIME_ONLY SYSTEM_ONLY)j1j2j3Zj4Zj5Zj6r r r test_reader_init_flagss    rPcCstjttjtjWdQRXttj|jtjd}WdQRXttj|jtjtjBd}WdQRXtj|jtjtj Bd}dS)N)pathflags) r&r0r+rrKZOS_ROOTr,strpath CURRENT_USERrL)tmpdirrMrNrOr r r test_reader_os_roots rVcCs>tj|jd}tjd|jdtj|jd}tj|jddS)N)rQr)rrKrS)rUrMrNr r r test_reader_init_pathsrWc Cs(tjttjdddWdQRXdS)Nrr:)rQ)r&r0r#rrKr r r r test_reader_init_path_invalid_fds rYc Cs(tjttjdddWdQRXdS)Nr)rQ)r&r0r#rrKr r r r %test_reader_init_path_nondirectory_fds rZcCshtj|jtj}ttjtj|d}WdQRXt tjtj |d}WdQRXtjtj |d}dS)N)rQ) osopenrSO_RDONLYr*r%ENOSYSrrKr,ZSYSTEMrT)rUfdrMrNrOr r r test_reader_init_path_fds r`cCs@tj|jd}|WdQRXtjtt|WdQRXdS)N)rQ)rrKrSr&r0r#next)rUjr r r test_reader_as_cms   rcc Cs<tj|jd}| |jtj|jtjjWdQRXdS)N)rQ)rrKrSZmessageid_matchrZSD_MESSAGE_JOURNAL_STARTZSD_MESSAGE_JOURNAL_STOPr )rUrbr r r test_reader_messageid_matchs rdc Cs@tj|jd}|$|j|jt|jtjWdQRXdS)N)rQ)rrKrSZ this_bootrDr )rUrbr r r test_reader_this_boots  rec Cs@tj|jd}|$|j|jt|jtjWdQRXdS)N)rQ)rrKrSZ this_machinerDr )rUrbr r r test_reader_this_machines  rfcCsBtj|jd}|&ttj|jd}WdQRXWdQRXdS)N)rQZFOOBAR)rrKrSr*r%r^Z query_unique)rUrbansr r r test_reader_query_uniques  rhcCs@tj|jd}|$ttj|j}WdQRXWdQRXdS)N)rQ)rrKrSr*r%r^Zenumerate_fields)rUrbrgr r r test_reader_enumerate_fieldss  ricCs@tj|jd}|$ttj|j}WdQRXWdQRXdS)N)rQ)rrKrSr*r%r^has_runtime_files)rUrbrgr r r test_reader_has_runtime_filess  rkcCs@tj|jd}|$ttj|j}WdQRXWdQRXdS)N)rQ)rrKrSr*r%r^rj)rUrbrgr r r test_reader_has_persistent_filess  rlcCs8dddi}tj|j|d}|jdd}|jdd}dS)NZxxxcSsdS)Nyyyr )argr r r sz(test_reader_converters..)rQ converterssabcZzzzs)rrKrSZ_convert_field)rUrprbrr r r test_reader_converterss    rqcCsBddddd}tj|j|d}|jddddgdd gd }dS) NcSsdS)Nrmr )rnr r r ro sz+test_reader_convert_entry..cSsdS)NZYYYr )rnr r r ro s)x1x2)rQrpsabcssdefs)rrZy1rsZy2)rrKrSZ_convert_entry)rUrprbrr r r test_reader_convert_entry s rtcCsFtj|jd}tj}|j||jdtjddd}|j|dS)N)rQi90ir3)rrKrStimeZ seek_realtimedatetime)rUrbZnowZlong_agor r r test_seek_realtimes   rxc Cs:ttjtjd}WdQRX|jd}td|ddS)Nztest_journal.pyz message... zprinted message...)file)r*r%ENOENTrstreamwriteprint)r{resr r r test_journal_stream"s  r)3Z __future__r contextlibrwr%r6r[rvZuuid tracebackrZsystemdrrZsystemd.journalrr&ZUUIDrDrGrcontextmanagerr*r,r/r2r8rBrCrErFrIrJrPrVrWrYrZr`rcrdrerfrhrirkrlrqrtrxrr r r r sV              __pycache__/test_login.cpython-36.pyc000064400000003100151030026100013527 0ustar003 XF@srddlmZddlZddlZddlZddlmZddlZejddZ ddZ dd Z d d Z d d Z ddZdS))print_functionN)loginccsJy dVWn:ttfk rD}z|j|kr2tjWYdd}~XnXdS)N)OSErrorIOErrorerrnopytestskip)codeer "/usr/lib64/python3.6/test_login.py skip_oserror s   r c Cs2ttjtj}t|dks$tWdQRXdS)Nr)r rENOENTrseatslenAssertionError)rr r r test_seatss rc Cs2ttjtj}t|dks$tWdQRXdS)Nr)r rrrsessionsrr)rr r r test_sessionss rc Cs2ttjtj}t|dks$tWdQRXdS)Nr)r rrr machine_namesrr)rr r r test_machine_namess rc Cs2ttjtj}t|dks$tWdQRXdS)Nr)r rrruidsrr)rr r r test_uids#s rc CsVtj}ttj:tjd}|j||jtj |jdtj WdQRXdS)Nmachine) selectZpollr rrrZMonitorregisterZ get_eventsr)pmr r r test_monitor(s   r)Z __future__rr contextlibrZsystemdrrcontextmanagerr rrrrrr r r r s   test_journal.py000064400000023535151030026100007623 0ustar00from __future__ import print_function import contextlib import datetime import errno import logging import os import time import uuid import traceback as _traceback from systemd import journal, id128 from systemd.journal import _make_line import pytest TEST_MID = uuid.UUID('8441372f8dca4ca98694a6091fd8519f') TEST_MID2 = uuid.UUID('8441370000000000000000001fd85000') class MockSender: def __init__(self): self.buf = [] def send(self, MESSAGE, MESSAGE_ID=None, CODE_FILE=None, CODE_LINE=None, CODE_FUNC=None, **kwargs): args = ['MESSAGE=' + MESSAGE] if MESSAGE_ID is not None: id = getattr(MESSAGE_ID, 'hex', MESSAGE_ID) args.append('MESSAGE_ID=' + id) if CODE_LINE is CODE_FILE is CODE_FUNC is None: CODE_FILE, CODE_LINE, CODE_FUNC = _traceback.extract_stack(limit=2)[0][:3] if CODE_FILE is not None: args.append('CODE_FILE=' + CODE_FILE) if CODE_LINE is not None: args.append('CODE_LINE={:d}'.format(CODE_LINE)) if CODE_FUNC is not None: args.append('CODE_FUNC=' + CODE_FUNC) args.extend(_make_line(key, val) for key, val in kwargs.items()) self.buf.append(args) @contextlib.contextmanager def skip_oserror(code): try: yield except (OSError, IOError) as e: if e.errno == code: pytest.skip() raise @contextlib.contextmanager def skip_valueerror(): try: yield except ValueError: pytest.skip() def test_priorities(): p = journal.JournalHandler.map_priority assert p(logging.NOTSET) == journal.LOG_DEBUG assert p(logging.DEBUG) == journal.LOG_DEBUG assert p(logging.DEBUG - 1) == journal.LOG_DEBUG assert p(logging.DEBUG + 1) == journal.LOG_INFO assert p(logging.INFO - 1) == journal.LOG_INFO assert p(logging.INFO) == journal.LOG_INFO assert p(logging.INFO + 1) == journal.LOG_WARNING assert p(logging.WARN - 1) == journal.LOG_WARNING assert p(logging.WARN) == journal.LOG_WARNING assert p(logging.WARN + 1) == journal.LOG_ERR assert p(logging.ERROR - 1) == journal.LOG_ERR assert p(logging.ERROR) == journal.LOG_ERR assert p(logging.ERROR + 1) == journal.LOG_CRIT assert p(logging.FATAL) == journal.LOG_CRIT assert p(logging.CRITICAL) == journal.LOG_CRIT assert p(logging.CRITICAL + 1) == journal.LOG_ALERT def test_journalhandler_init_exception(): kw = {' X ':3} with pytest.raises(ValueError): journal.JournalHandler(**kw) def test_journalhandler_init(): kw = {'X':3, 'X3':4} journal.JournalHandler(logging.INFO, **kw) def test_journalhandler_info(): record = logging.LogRecord('test-logger', logging.INFO, 'testpath', 1, 'test', None, None) sender = MockSender() kw = {'X':3, 'X3':4, 'sender_function': sender.send} handler = journal.JournalHandler(logging.INFO, **kw) handler.emit(record) assert len(sender.buf) == 1 assert 'X=3' in sender.buf[0] assert 'X3=4' in sender.buf[0] def test_journalhandler_no_message_id(): record = logging.LogRecord('test-logger', logging.INFO, 'testpath', 1, 'test', None, None) sender = MockSender() handler = journal.JournalHandler(logging.INFO, sender_function=sender.send) handler.emit(record) assert len(sender.buf) == 1 assert all(not m.startswith('MESSAGE_ID=') for m in sender.buf[0]) def test_journalhandler_message_id_on_handler(): record = logging.LogRecord('test-logger', logging.INFO, 'testpath', 1, 'test', None, None) sender = MockSender() handler = journal.JournalHandler(logging.INFO, sender_function=sender.send, MESSAGE_ID=TEST_MID) handler.emit(record) assert len(sender.buf) == 1 assert 'MESSAGE_ID=' + TEST_MID.hex in sender.buf[0] def test_journalhandler_message_id_on_handler_hex(): record = logging.LogRecord('test-logger', logging.INFO, 'testpath', 1, 'test', None, None) sender = MockSender() handler = journal.JournalHandler(logging.INFO, sender_function=sender.send, MESSAGE_ID=TEST_MID.hex) handler.emit(record) assert len(sender.buf) == 1 assert 'MESSAGE_ID=' + TEST_MID.hex in sender.buf[0] def test_journalhandler_message_id_on_message(): record = logging.LogRecord('test-logger', logging.INFO, 'testpath', 1, 'test', None, None) record.__dict__['MESSAGE_ID'] = TEST_MID2 sender = MockSender() handler = journal.JournalHandler(logging.INFO, sender_function=sender.send, MESSAGE_ID=TEST_MID) handler.emit(record) assert len(sender.buf) == 1 assert 'MESSAGE_ID=' + TEST_MID2.hex in sender.buf[0] def test_journalhandler_message_id_on_message_hex(): record = logging.LogRecord('test-logger', logging.INFO, 'testpath', 1, 'test', None, None) record.__dict__['MESSAGE_ID'] = TEST_MID2.hex sender = MockSender() handler = journal.JournalHandler(logging.INFO, sender_function=sender.send, MESSAGE_ID=TEST_MID) handler.emit(record) assert len(sender.buf) == 1 assert 'MESSAGE_ID=' + TEST_MID2.hex in sender.buf[0] def test_reader_init_flags(): j1 = journal.Reader() j2 = journal.Reader(journal.LOCAL_ONLY) j3 = journal.Reader(journal.RUNTIME_ONLY) j4 = journal.Reader(journal.SYSTEM_ONLY) j5 = journal.Reader(journal.LOCAL_ONLY | journal.RUNTIME_ONLY | journal.SYSTEM_ONLY) j6 = journal.Reader(0) def test_reader_os_root(tmpdir): with pytest.raises(ValueError): journal.Reader(journal.OS_ROOT) with skip_valueerror(): j1 = journal.Reader(path=tmpdir.strpath, flags=journal.OS_ROOT) with skip_valueerror(): j2 = journal.Reader(path=tmpdir.strpath, flags=journal.OS_ROOT | journal.CURRENT_USER) j3 = journal.Reader(path=tmpdir.strpath, flags=journal.OS_ROOT | journal.SYSTEM_ONLY) def test_reader_init_path(tmpdir): j1 = journal.Reader(path=tmpdir.strpath) journal.Reader(0, path=tmpdir.strpath) j2 = journal.Reader(path=tmpdir.strpath) journal.Reader(path=tmpdir.strpath) def test_reader_init_path_invalid_fd(): with pytest.raises(OSError): journal.Reader(0, path=-1) def test_reader_init_path_nondirectory_fd(): with pytest.raises(OSError): journal.Reader(0, path=0) def test_reader_init_path_fd(tmpdir): fd = os.open(tmpdir.strpath, os.O_RDONLY) with skip_oserror(errno.ENOSYS): j1 = journal.Reader(path=fd) assert list(j1) == [] with skip_valueerror(): j2 = journal.Reader(journal.SYSTEM, path=fd) assert list(j2) == [] j3 = journal.Reader(journal.CURRENT_USER, path=fd) assert list(j3) == [] def test_reader_as_cm(tmpdir): j = journal.Reader(path=tmpdir.strpath) with j: assert not j.closed assert j.closed # make sure that operations on the Reader fail with pytest.raises(OSError): next(j) def test_reader_messageid_match(tmpdir): j = journal.Reader(path=tmpdir.strpath) with j: j.messageid_match(id128.SD_MESSAGE_JOURNAL_START) j.messageid_match(id128.SD_MESSAGE_JOURNAL_STOP.hex) def test_reader_this_boot(tmpdir): j = journal.Reader(path=tmpdir.strpath) with j: j.this_boot() j.this_boot(TEST_MID) j.this_boot(TEST_MID.hex) def test_reader_this_machine(tmpdir): j = journal.Reader(path=tmpdir.strpath) with j: j.this_machine() j.this_machine(TEST_MID) j.this_machine(TEST_MID.hex) def test_reader_query_unique(tmpdir): j = journal.Reader(path=tmpdir.strpath) with j: with skip_oserror(errno.ENOSYS): ans = j.query_unique('FOOBAR') assert isinstance(ans, set) assert ans == set() def test_reader_enumerate_fields(tmpdir): j = journal.Reader(path=tmpdir.strpath) with j: with skip_oserror(errno.ENOSYS): ans = j.enumerate_fields() assert isinstance(ans, set) assert ans == set() def test_reader_has_runtime_files(tmpdir): j = journal.Reader(path=tmpdir.strpath) with j: with skip_oserror(errno.ENOSYS): ans = j.has_runtime_files() assert ans is False def test_reader_has_persistent_files(tmpdir): j = journal.Reader(path=tmpdir.strpath) with j: with skip_oserror(errno.ENOSYS): ans = j.has_runtime_files() assert ans is False def test_reader_converters(tmpdir): converters = {'xxx' : lambda arg: 'yyy'} j = journal.Reader(path=tmpdir.strpath, converters=converters) val = j._convert_field('xxx', b'abc') assert val == 'yyy' val = j._convert_field('zzz', b'\200\200') assert val == b'\200\200' def test_reader_convert_entry(tmpdir): converters = {'x1' : lambda arg: 'yyy', 'x2' : lambda arg: 'YYY'} j = journal.Reader(path=tmpdir.strpath, converters=converters) val = j._convert_entry({'x1' : b'abc', 'y1' : b'\200\200', 'x2' : [b'abc', b'def'], 'y2' : [b'\200\200', b'\200\201']}) assert val == {'x1' : 'yyy', 'y1' : b'\200\200', 'x2' : ['YYY', 'YYY'], 'y2' : [b'\200\200', b'\200\201']} def test_seek_realtime(tmpdir): j = journal.Reader(path=tmpdir.strpath) now = time.time() j.seek_realtime(now) j.seek_realtime(12345) long_ago = datetime.datetime(1970, 5, 4) j.seek_realtime(long_ago) def test_journal_stream(): # This will fail when running in a bare chroot without /run/systemd/journal/stdout with skip_oserror(errno.ENOENT): stream = journal.stream('test_journal.py') res = stream.write('message...\n') assert res in (11, None) # Python2 returns None print('printed message...', file=stream)