This repository has been archived on 2023-02-21. You can view files and clone it, but cannot push or open issues or pull requests.
logtracker/tests/test_journalstream.py

113 lines
3.2 KiB
Python

import io
from django.test.utils import override_settings
from logtracker.journal.journalstream import get_journal_entries
raw_entries = b'''__CURSOR=s=6f1be3fd9e534674878d6ce7394e035f;i=f3b60;b=f8595485f17145e89bf6d8068dfbf140;m=cb448036;t=5ab93a50ad4e5;x=99d88bd273a65ed7
__REALTIME_TIMESTAMP=1596025501045989
__MONOTONIC_TIMESTAMP=3410264118
_BOOT_ID=f8595485f17145e89bf6d8068dfbf140
PRIORITY=6
_TRANSPO\r
fff4\r
RT=journal
_SELINUX_CONTEXT
^K^@^@^@^@^@^@^@unconfined
__CURSOR=s=92043378025640f3989a22d75fc7422d;i=f5ce1;b=c2a67631459e4eb4bef62a31641efe64;m=cdec3b6f;t=5abf656e3cfe2;x=b34e4efef33912b1
__REALTIME_TIMESTAMP=1596449391628258
__MONOTONIC_TIMESTAMP=3454810991
_BOOT_ID=c2a67631459e4eb4bef62a31641efe64
_UID=1000
_GID=1000
_CAP_EFFECTIVE=0
_SELINUX_CONTEXT
^K^@^@^@^@^@^@^@unconfined
_MACHINE_ID=332e6893061344b3bf9e76f4f066f008
_HOSTNAME=pad
_TRANSPORT=journal
_AUDIT_LOGINUID=1000
_SYSTEMD_OWNER_UID=1000
_SYSTEMD_SLICE=user-1000.slice
_SYSTEMD_USER_SLICE=-.slice
_AUDIT_SESSION=2
_SYSTEMD_CGROUP=/user.slice/user-1000.slice/session-2.scope
_SYSTEMD_SESSION=2
_SYSTEMD_UNIT=session-2.scope
_SYSTEMD_INVOCATION_ID=56dcd19554ef4153b7a409267bac34d7
PRIORITY=4
GLIB_OLD_LOG_API=1
_PID=1886
_COMM=gnome-shell
_EXE=/usr/bin/gnome-shell
_CMDLINE=/usr/bin/gnome-shell
GLIB_DOMAIN=Gjs
MESSAGE
^@^@^@^@^@^@^@JS ERROR: TypeError: this._trackedWindows.get(...) is undefined
_onWindowActorRemoved@resource:///org/gnome/shell/ui/panel.js:843:9
wrapper@resource:///org/gnome/gjs/modules/_legacy.js:82:22
_SOURCE_REALTIME_TIMESTAMP=1596449391615304
__CURSOR=s=92043378025640f3989a22d75fc7422d;i=f5ce2;b=c2a67631459e4eb4bef62a31641efe64;m=cdec3da9;t=5abf656e3d21b;x=43f26d5c857dddba
__REALTIME_TIMESTAMP=1596449391628827
__MONOTONIC_TIMESTAMP=3454811561
_BOOT_ID=c2a67631459e4eb4bef62a31641efe64
_UID=1000
_GID=1000
_CAP_EFFECTIVE=0
_SELINUX_CONTEXT
^K^@^@^@^@^@^@^@unconfined
_MACHINE_ID=332e6893061344b3bf9e76f4f066f008
_HOSTNAME=pad
PRIORITY=6
SYSLOG_FACILITY=3
SYSLOG_IDENTIFIER=systemd
_TRANSPORT=journal
_PID=1827
_COMM=systemd
_EXE=/lib/systemd/systemd
_CMDLINE=/lib/systemd/systemd --user
_AUDIT_SESSION=3
_AUDIT_LOGINUID=1000
_SYSTEMD_CGROUP=/user.slice/user-1000.slice/user@1000.service/init.scope
_SYSTEMD_OWNER_UID=1000
_SYSTEMD_UNIT=user@1000.service
_SYSTEMD_USER_UNIT=init.scope
_SYSTEMD_SLICE=user-1000.slice
_SYSTEMD_USER_SLICE=-.slice
_SYSTEMD_INVOCATION_ID=8ce5033623b845dfae8ea4666504d836
USER_UNIT=at-spi-dbus-bus.service
USER_INVOCATION_ID=cab95ef28cc14d17a509f096120f7d19
CODE_FILE=src/core/unit.c
CODE_LINE=5745
CODE_FUNC=unit_log_success
MESSAGE_ID=7ad2d189f7e94e70a38c781354912448
MESSAGE=at-spi-dbus-bus.service: Succeeded.
_SOURCE_REALTIME_TIMESTAMP=1596449391625441
0\r
'''
def test_journal_stream_auth(client):
page = client.post('/upload')
assert page.status_code == 403
@override_settings(DEBUG=True)
def test_journal_stream_auth_debug(client):
page = client.post('/upload')
assert page.status_code == 200
def test_journal_stream():
bf = io.BytesIO(raw_entries)
entries = list(get_journal_entries(bf))
assert len(entries) == 3
assert ('_TRANSPORT', 'journal') in entries[0]
assert '_legacy.js:82' in [v for k, v in entries[1] if k == 'MESSAGE'][0]