91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
import datetime
|
|
import pytest
|
|
from mock import patch
|
|
|
|
from django.utils import timezone
|
|
|
|
from passerelle.base.models import ResourceLog
|
|
from passerelle.apps.opengis.models import OpenGIS
|
|
from passerelle.apps.clicrdv.models import ClicRdv
|
|
|
|
|
|
def test_get_description_url_fields(db):
|
|
connector = OpenGIS(slug='plop', wms_service_url='http://www.example.net')
|
|
assert 'http://www.example.net' in [x[1] for x in connector.get_description_fields()]
|
|
|
|
connector = OpenGIS(slug='plop', wms_service_url='http://username:secret@www.example.net')
|
|
assert 'http://***:***@www.example.net' in [x[1] for x in connector.get_description_fields()]
|
|
|
|
connector = OpenGIS(slug='plop', wms_service_url='http://username@example.net:secret@www.example.net')
|
|
assert 'http://***:***@www.example.net' in [x[1] for x in connector.get_description_fields()]
|
|
|
|
|
|
def test_get_description_secret_fields(db):
|
|
connector = ClicRdv(slug='plop', apikey='secret1', username='plop', password='secret2')
|
|
assert not 'secret1' in [x[1] for x in connector.get_description_fields()]
|
|
assert not 'secret2' in [x[1] for x in connector.get_description_fields()]
|
|
|
|
|
|
def test_log_cleaning(db):
|
|
ResourceLog.objects.all().delete()
|
|
connector = OpenGIS(slug='plop', wms_service_url='http://www.example.net')
|
|
connector.logger.error('hello1')
|
|
connector.logger.error('hello2')
|
|
|
|
assert ResourceLog.objects.all().count() == 2
|
|
|
|
ResourceLog.objects.update(timestamp=timezone.now() - datetime.timedelta(days=10))
|
|
connector.logger.error('hello3')
|
|
assert ResourceLog.objects.all().count() == 3
|
|
|
|
connector.daily()
|
|
assert ResourceLog.objects.all().count() == 1
|
|
|
|
@pytest.fixture
|
|
def email_handler():
|
|
import logging
|
|
from django.utils.log import AdminEmailHandler
|
|
|
|
root = logging.getLogger()
|
|
handler = AdminEmailHandler(include_html=True)
|
|
handler.level = logging.ERROR
|
|
root.handlers.append(handler)
|
|
try:
|
|
yield
|
|
finally:
|
|
root.handlers.remove(handler)
|
|
|
|
|
|
def test_trace_emails(app, settings, dummy_csv_datasource, email_handler, mailoutbox):
|
|
from .utils import generic_endpoint_url
|
|
|
|
settings.ADMINS = [('admin', 'admin@example.net')]
|
|
|
|
logging_parameters = dummy_csv_datasource.logging_parameters
|
|
logging_parameters.save()
|
|
|
|
assert not mailoutbox
|
|
|
|
with patch.object(dummy_csv_datasource.__class__,
|
|
'execute_query',
|
|
side_effect=ValueError('coin'), autospec=True):
|
|
|
|
app.get(generic_endpoint_url(connector='csvdatasource',
|
|
endpoint='query/dummy-query/',
|
|
slug=dummy_csv_datasource.slug),
|
|
status=500)
|
|
|
|
assert len(mailoutbox) == 1
|
|
assert mailoutbox[0].to == ['admin@example.net']
|
|
|
|
logging_parameters.trace_emails = 'john.doe@example.net'
|
|
logging_parameters.save()
|
|
app.get(generic_endpoint_url(connector='csvdatasource',
|
|
endpoint='query/dummy-query/',
|
|
slug=dummy_csv_datasource.slug),
|
|
status=500)
|
|
|
|
assert len(mailoutbox) == 2
|
|
assert mailoutbox[0].to == ['admin@example.net']
|
|
assert mailoutbox[1].to == ['john.doe@example.net']
|