1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202
|
import contextlib
import datetime
import errno
import logging
import os
import time
import uuid
from systemd import journal, id128
import pytest
TEST_MID = uuid.UUID('8441372f8dca4ca98694a6091fd8519f')
@contextlib.contextmanager
def skip_enosys():
try:
yield
except OSError as e:
if e.errno == errno.ENOSYS:
pytest.skip()
raise
@contextlib.contextmanager
def skip_valueerror():
try:
yield
except ValueError:
pytest.skip()
def test_priorities():
p = journal.JournalHandler.mapPriority
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_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_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_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_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_enosys():
ans = j.has_runtime_files()
assert ans == False
def test_reader_has_persistent_files(tmpdir):
j = journal.Reader(path=tmpdir.strpath)
with j:
with skip_enosys():
ans = j.has_runtime_files()
assert ans == 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)
|