96 lines
3.6 KiB
Python
96 lines
3.6 KiB
Python
import pytest
|
|
|
|
import logging
|
|
|
|
from hobo.logger import RequestContextFilter
|
|
|
|
from tenant_schemas.utils import tenant_context
|
|
|
|
from django.contrib.auth.models import User
|
|
|
|
|
|
def test_request_context_filter(caplog, settings, tenants, client):
|
|
root_logger = logging.getLogger()
|
|
assert len(root_logger.handlers) == 1
|
|
root_logger.handlers[0].addFilter(RequestContextFilter())
|
|
for tenant in tenants:
|
|
with tenant_context(tenant):
|
|
user = User.objects.create(first_name='John', last_name='Doe', username='john.doe',
|
|
email='jodn.doe@example.com')
|
|
user.set_password('john.doe')
|
|
user.save()
|
|
user.saml_identifiers.create(name_id='ab' * 16, issuer='https://idp.example.com')
|
|
|
|
for tenant in tenants:
|
|
settings.ALLOWED_HOSTS.append(tenant.domain_url)
|
|
with tenant_context(tenant):
|
|
client.login(username='john.doe', password='john.doe')
|
|
client.get('/', SERVER_NAME=tenant.domain_url,
|
|
HTTP_X_FORWARDED_FOR='99.99.99.99, 127.0.0.1')
|
|
records = [record for record in caplog.records if record.levelno >= logging.INFO]
|
|
assert len(records) == 2
|
|
for tenant, record in zip(tenants, records):
|
|
assert record.ip == '99.99.99.99'
|
|
assert record.tenant == tenant.domain_url
|
|
assert record.path == '/'
|
|
assert record.request_id.startswith('r:')
|
|
assert record.user == user.username
|
|
assert record.user_email == user.email
|
|
assert record.user_name == user.username
|
|
assert record.user_display_name == 'John Doe'
|
|
assert record.user_uuid == 'ab' * 16
|
|
assert record.application == 'fake-agent'
|
|
|
|
|
|
@pytest.fixture
|
|
def journald_handler():
|
|
from hobo.journal import JournalHandler
|
|
|
|
root_logger = logging.getLogger()
|
|
journald_handler = JournalHandler()
|
|
root_logger.handlers.append(journald_handler)
|
|
try:
|
|
yield journald_handler
|
|
finally:
|
|
root_logger.handlers.remove(journald_handler)
|
|
|
|
|
|
def test_systemd(settings, tenants, client, journald_handler):
|
|
root_logger = logging.getLogger()
|
|
assert len(root_logger.handlers) == 2
|
|
journald_handler.addFilter(RequestContextFilter())
|
|
|
|
for tenant in tenants:
|
|
with tenant_context(tenant):
|
|
user = User.objects.create(first_name='John', last_name='Doe', username='john.doe',
|
|
email='jodn.doe@example.com')
|
|
user.set_password('john.doe')
|
|
user.save()
|
|
user.saml_identifiers.create(name_id='ab' * 16, issuer='https://idp.example.com')
|
|
|
|
for tenant in tenants:
|
|
settings.ALLOWED_HOSTS.append(tenant.domain_url)
|
|
with tenant_context(tenant):
|
|
client.login(username='john.doe', password='john.doe')
|
|
client.get('/', SERVER_NAME=tenant.domain_url,
|
|
HTTP_X_FORWARDED_FOR='99.99.99.99, 127.0.0.1')
|
|
|
|
from systemd.journal import Reader
|
|
import time
|
|
|
|
reader = Reader()
|
|
reader.seek_realtime(time.time() - 10)
|
|
records = [l for l in reader if l['MESSAGE'] == 'wat!']
|
|
assert len(records) == 2
|
|
for tenant, record in zip(tenants, records):
|
|
assert record['IP'] == '99.99.99.99'
|
|
assert record['TENANT'] == tenant.domain_url
|
|
assert record['PATH'] == '/'
|
|
assert record['REQUEST_ID'].startswith('r:')
|
|
assert record['USER'] == user.username
|
|
assert record['USER_EMAIL'] == user.email
|
|
assert record['USER_NAME'] == user.username
|
|
assert record['USER_DISPLAY_NAME'] == 'John Doe'
|
|
assert record['USER_UUID'] == 'ab' * 16
|
|
assert record['APPLICATION'] == 'fake-agent'
|