passerelle/tests/test_jsonresponse.py

import json
import logging

import pytest
from django.core.exceptions import ObjectDoesNotExist, PermissionDenied
from django.http import Http404
from django.test.client import RequestFactory

from passerelle.utils import to_json


class WrappedException(Exception):
    pass


class LogAsWarningException(Exception):
    err_code = 'logaswarningexception'
    http_status = 488
    log_error = False

    def __str__(self):
        return 'log as warning exception'


@to_json()
def wrapped_exception(req, *args, **kwargs):
    raise WrappedException


@to_json()
def log_as_warning_exception(req, *args, **kwargs):
    raise LogAsWarningException


@to_json()
def http404_exception(req, *args, **kwargs):
    raise Http404


@to_json()
def doesnotexist_exception(req, *args, **kwargs):
    raise ObjectDoesNotExist


@to_json()
def permissiondenied_exception(req, *args, **kwargs):
    raise PermissionDenied


def test_jsonresponselog_get(rf, caplog, settings):
    caplog.set_level(logging.ERROR)

    wrapped_exception(rf.get('/'))
    assert len(caplog.records) == 1
    record = caplog.records[0]
    assert record.name == 'passerelle.jsonresponse'
    assert record.levelno == logging.ERROR
    assert hasattr(record, 'method')
    if record.method == 'POST':
        assert hasattr(record, 'body')
    assert 'Error occurred while processing request' in record.message

    caplog.clear()

    post_payload = {'data': 'plop'}
    with pytest.raises(WrappedException):
        wrapped_exception(rf.post('/?raise=1', post_payload))
    assert len(caplog.records) == 1
    caplog.clear()
    wrapped_exception(rf.post('/?raise=0', post_payload))
    wrapped_exception(rf.post('/?raise=blah', post_payload))
    wrapped_exception(rf.post('/?raise', post_payload))
    assert len(caplog.records) == 3


def test_jsonresponselog_http404(caplog):
    request = RequestFactory()
    http404_exception(request.get('/'))
    assert caplog.records == []


def test_jsonresponselog_doesnotexist(caplog):
    request = RequestFactory()
    doesnotexist_exception(request.get('/'))
    for record in caplog.records:
        assert record.name == 'passerelle.jsonresponse'
        assert record.levelno == logging.WARNING
        assert 'object not found' in record.message


def test_jsonresponselog_permissiondenied(caplog):
    request = RequestFactory()
    permissiondenied_exception(request.get('/'))
    for record in caplog.records:
        assert record.name == 'passerelle.jsonresponse'
        assert record.levelno == logging.WARNING
        assert 'Permission denied' in record.message


def test_jsonresponse_log_as_warning_exception(caplog):
    request = RequestFactory()
    response = log_as_warning_exception(request.get('/'))
    records = caplog.records
    assert len(records) == 1
    record = records[0]
    assert record.name == 'passerelle.jsonresponse'
    assert record.levelno == logging.WARNING
    assert hasattr(record, 'method')
    assert record.method == 'GET'
    assert 'Error occurred while processing request' in record.message
    assert response.status_code == 488
    data = json.loads(response.content)
    assert data['err'] == 'logaswarningexception'
    assert data['err_desc'] == 'log as warning exception'


def test_jsonresponse_error_header():
    request = RequestFactory()
    req = request.get('/')

    @to_json()
    def test_func(req):
        return {'test': 'un test'}

    result = test_func(req)
    assert result.status_code == 200
    data = json.loads(result.content)
    assert data == {'test': 'un test', 'err': 0}

    @to_json()
    def test_func2(req):
        class CustomException(Exception):
            http_status = 200

        raise CustomException

    result = test_func2(req)
    data = json.loads(result.content)
    assert 'err_class' in data
    assert 'err' in data
    assert data['err'] == 1
    assert data['err_class'] == 'tests.test_jsonresponse.CustomException'
    assert result.status_code == 200


def test_jsonresponse_with_http4O4_exception():
    request = RequestFactory()
    response = http404_exception(request.get('/'))
    assert response.status_code == 404


def test_jsonresponse_with_callback():
    request = RequestFactory()
    req = request.get('/?callback=myfunc')

    @to_json()
    def test_func(req):
        return {'foo': 'bar'}

    result = test_func(req)
    content_type = result.get('Content-Type')
    assert 'application/javascript' in content_type
    assert result.content.startswith(b'myfunc(')
    args = json.loads(result.content[7:-2])
    assert args == {'foo': 'bar', 'err': 0}


def test_jsonresponse_with_wrong_callback():
    request = RequestFactory()
    req = request.get('/?callback=myfunc()')

    @to_json()
    def test_func(req):
        return {'foo': 'bar'}

    result = test_func(req)
    assert result.status_code == 400