add validation API (fixes #10375)

- a serializer for referring to an user factorized with the push document API
- add common mixin for handling validation errors and successfull results
- add common fixtures for writing tests
This commit is contained in:
Benjamin Dauvergne 2016-03-28 03:06:59 +02:00
parent ad6b0fcac4
commit a302008b4b
11 changed files with 386 additions and 62 deletions

18
fargo/fargo/api_errors.py Normal file
View File

@ -0,0 +1,18 @@
from django.utils.translation import ugettext_lazy as _
from rest_framework.serializers import ValidationError
class APIError(ValidationError):
MISSING_USER = {'code': 'missing-user', 'msg': _('an user must be specified')}
UNKNOWN_USER = {'code': 'unknown-user', 'msg': _('user is unknown')}
NOT_BASE64 = {'code': 'not-base64', 'msg': _('data is not base64 encoded')}
NOT_STRING = {'code': 'not-string', 'msg': _('data is not a string')}
TOO_BIG = {'code': 'too-big', 'msg': _('file is too big (limit is {limit})')}
def __init__(self, kind, **kwargs):
assert hasattr(self, kind), 'error %s is not defined' % kind
detail = getattr(self, kind).copy()
detail.update(kwargs)
detail['msg'] = detail['msg'].format(**kwargs)
super(APIError, self).__init__([detail])

31
fargo/fargo/api_fields.py Normal file
View File

@ -0,0 +1,31 @@
import base64
import uuid
import six
from django.core.files.base import ContentFile
from rest_framework import fields, serializers
from . import api_errors
class SlugCreatedRelatedField(serializers.SlugRelatedField):
def to_internal_value(self, data):
return self.get_queryset().get_or_create(**{self.slug_field: data})[0]
class Base64FileField(fields.Field):
def __init__(self, *args, **kwargs):
super(Base64FileField, self).__init__(*args, **kwargs)
def to_internal_value(self, data):
if isinstance(data, six.string_types):
# base64 encoded image - decode
name = uuid.uuid4()
try:
content = base64.b64decode(data)
except ValueError:
raise api_errors.APIError('NOT_BASE64')
data = ContentFile(content, name=name)
else:
raise api_errors.APIError('NOT_STRING')
return data

View File

@ -1,10 +1,3 @@
# fargo - document box
# Copyright (C) 2015-2016 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
@ -14,21 +7,22 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import base64
import datetime
from django.conf import settings
from django.core.files.base import ContentFile
from django.core.urlresolvers import reverse
from django.contrib.auth.models import User
from django.utils.text import slugify
from django.utils.timezone import now
from rest_framework import serializers
from rest_framework.generics import GenericAPIView, ListAPIView
from rest_framework.permissions import IsAdminUser
from rest_framework.response import Response
from rest_framework import status
from rest_framework import status, filters, mixins, viewsets, routers, exceptions
from .models import Origin, Document, UserDocument
from .models import Origin, Document, UserDocument, Validation
from . import utils, api_errors, api_fields
try:
from mellon.models import UserSAMLIdentifier
@ -36,74 +30,89 @@ except ImportError:
UserSAMLIdentifier = None
class PushDocumentSerializer(serializers.Serializer):
origin = serializers.CharField(required=True)
file_b64_content = serializers.CharField(required=True)
file_name = serializers.CharField(required=False)
document_type = serializers.ChoiceField(
required=True, choices=[x.get('name') for x in settings.FARGO_DOCUMENT_TYPES])
user_email = serializers.CharField(required=False)
user_nameid = serializers.CharField(required=False)
class UserSerializerMixin(serializers.Serializer):
user_email = serializers.CharField(write_only=True, required=False)
user_nameid = serializers.CharField(write_only=True, required=False)
def validate(self, data):
if not (data.get('user_email') or data.get('user_nameid')):
raise serializers.ValidationError('user must be specified')
raise api_errors.APIError('MISSING_USER')
if data.get('user_email'):
user = User.objects.get(email=data.get('user_email'))
elif data.get('user_nameid'):
user = UserSAMLIdentifier.objects.get(name_id=data.get('user_nameid')).user
data.pop('user_email', None)
data.pop('user_nameid', None)
data['user'] = user
return data
def validate_user_email(self, value):
try:
User.objects.get(email=value)
except User.DoesNotExist:
raise serializers.ValidationError('unknown user')
raise api_errors.APIError('UNKNOWN_USER')
return value
def validate_user_nameid(self, value):
if value:
if UserSAMLIdentifier is None:
raise serializers.ValidationError('nameid lookups require django-mellon')
assert UserSAMLIdentifier, 'nameid lookups require django-mellon'
try:
UserSAMLIdentifier.objects.get(name_id=value)
except UserSAMLIdentifier.DoesNotExist:
raise serializers.ValidationError('unknown user')
raise api_errors.APIError('UNKNOWN_USER')
return value
class PushDocument(GenericAPIView):
class CommonAPIMixin(object):
def handle_exception(self, exc):
if isinstance(exc, exceptions.APIException):
exc.detail = {'result': 0, 'errors': exc.detail}
return super(CommonAPIMixin, self).handle_exception(exc)
def finalize_response(self, request, response, *args, **kwargs):
if not isinstance(response.data, dict) or not 'result' in response.data:
response.data = {'result': 1, 'data': response.data}
return super(CommonAPIMixin, self).finalize_response(request, response, *args, **kwargs)
class PushDocumentSerializer(UserSerializerMixin):
origin = serializers.CharField(required=True)
file_b64_content = api_fields.Base64FileField(required=True)
file_name = serializers.CharField(required=False)
document_type = serializers.ChoiceField(
required=True, choices=[x.get('name') for x in settings.FARGO_DOCUMENT_TYPES])
class PushDocument(CommonAPIMixin, GenericAPIView):
serializer_class = PushDocumentSerializer
permission_classes = (IsAdminUser,)
def post(self, request, format=None):
serializer = self.get_serializer(data=request.data)
if not serializer.is_valid():
response = {'result': 0, 'errors': serializer.errors}
return Response(response, status.HTTP_400_BAD_REQUEST)
raise serializers.ValidationError(serializer.errors)
data = serializer.validated_data
if data.get('user_email'):
user = User.objects.get(email=data.get('user_email'))
elif data.get('user_nameid'):
user = UserSAMLIdentifier.objects.get(name_id=data.get('user_nameid')).user
origin, created = Origin.objects.get_or_create(
slug=slugify(data.get('origin')),
defaults={'label': data.get('origin')})
document_file = ContentFile(base64.decodestring(data.get('file_b64_content')))
document_file = data['file_b64_content']
if data.get('file_name'):
document_file.name = data.get('file_name')
document = Document(content=document_file)
document.save()
user_document = UserDocument(
user=user,
user=data.get('user'),
filename=data.get('file_name'),
document=document,
origin=origin)
user_document.save()
response = {'result': 1}
response_status = status.HTTP_200_OK
return Response(response, response_status)
return Response(None, response_status)
push_document = PushDocument.as_view()
@ -127,3 +136,80 @@ class RecentDocuments(ListAPIView):
).order_by('-created')[:10]
recent_documents = RecentDocuments.as_view()
class ValidationSerializer(UserSerializerMixin, serializers.ModelSerializer):
origin = api_fields.SlugCreatedRelatedField(slug_field='label', queryset=Origin.objects.all())
url = serializers.SerializerMethodField()
def __init__(self, *args, **kwargs):
schema = kwargs.pop('schema')
super(ValidationSerializer, self).__init__(*args, **kwargs)
self.document_type = schema['name']
self.document_type_schema = schema
for field in schema['metadata']:
name = field['name']
required = field.get('required', True)
self.fields[name] = serializers.CharField(
source='data.%s' % name, required=required, allow_blank=True)
def get_url(self, instance):
url = reverse('fargo-api-validation-detail',
kwargs={'document_type': instance.document_type, 'pk': instance.pk})
if 'request' in self.context:
url = self.context['request'].build_absolute_uri(url)
return url
def validate(self, data):
data = super(ValidationSerializer, self).validate(data)
data['document_type'] = self.document_type
data['created'] = now()
data['start'] = data['created'].date()
data['end'] = data['start'] + datetime.timedelta(seconds=settings.FARGO_VALIDATION_LIFETIME)
data['creator'] = data['creator'][:256]
return data
class Meta:
model = Validation
exclude = ('data', 'user', 'document_type')
read_only_fields = ('created', 'start', 'end')
class FilterByUser(filters.BaseFilterBackend):
def filter_queryset(self, request, queryset, view):
if 'user_email' in request.GET:
return queryset.filter(user__email=request.GET['user_email'])
elif 'user_nameid' in request.GET:
return queryset.filter(user__usersamlidentifier__name_id=request.GET['user_nameid'])
return queryset
class ValidationAPI(CommonAPIMixin,
mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
viewsets.GenericViewSet):
serializer_class = ValidationSerializer
permission_classes = (IsAdminUser,)
filter_backends = [FilterByUser]
queryset = Validation.objects.all()
def get_queryset(self):
return super(ValidationAPI, self).get_queryset().filter(document_type=self.document_type)
def initial(self, request, document_type, *args, **kwargs):
self.schema = utils.get_document_type_schema(settings, document_type)
if not self.schema:
error = serializers.ValidationError('unknown document type')
error.status_code = status.HTTP_404_NOT_FOUND
raise error
self.document_type = document_type
super(ValidationAPI, self).initial(request, document_type, *args, **kwargs)
def get_serializer(self, *args, **kwargs):
# pass schema to serializer class
return super(ValidationAPI, self).get_serializer(schema=self.schema, *args, **kwargs)
router = routers.SimpleRouter()
router.register(r'validation/(?P<document_type>[^/]*)', ValidationAPI,
base_name='fargo-api-validation')

View File

@ -2,6 +2,8 @@ from django.conf import settings
from django.core.urlresolvers import reverse
from django.db import models
from django.utils.translation import ugettext_lazy as _
from django.utils.text import slugify
from jsonfield import JSONField
@ -12,6 +14,11 @@ class Origin(models.Model):
label = models.CharField(_('Label'), max_length=80)
slug = models.SlugField(_('Slug'))
def save(self, *args, **kwargs):
if not self.slug:
self.slug = slugify(self.label)
return super(Origin, self).save(*args, **kwargs)
class UserDocument(models.Model):
'''Document uploaded by an user or an agent'''

View File

@ -134,6 +134,10 @@ MELLON_USERNAME_TEMPLATE = '{attributes[name_id_content]}'
MELLON_IDENTITY_PROVIDERS = []
# Fargo settings
FARGO_VALIDATION_LIFETIME = 3600 * 24 * 31 * 6 # nearly 6 months
FARGO_DOCUMENT_TYPES = [
{
'name': 'justificatif-de-domicile',

View File

@ -5,7 +5,7 @@ from django.contrib import admin
from .fargo.views import (home, jsonp, json, document, download, pick, delete,
upload, remote_download, login, logout, pick_list,
metadata, validation, document_types)
from .fargo.api_views import (push_document, recent_documents)
from .fargo.api_views import (push_document, recent_documents, router)
urlpatterns = patterns(
'',
@ -30,8 +30,9 @@ urlpatterns = patterns(
url(r'^logout/$', logout, name='auth_logout'),
url(r'^document-types/$', document_types, name='document_types'),
url(r'^api/documents/push$', push_document, name='api-push-document'),
url(r'^api/documents/push$', push_document, name='fargo-api-push-document'),
url(r'^api/documents/recently-added$', recent_documents),
url(r'^api/', include(router.urls)),
)

82
tests/conftest.py Normal file
View File

@ -0,0 +1,82 @@
import logging
import pytest
import django_webtest
from django.contrib.auth.models import User
class WebTestMixin(django_webtest.WebTestMixin):
csrf_checks = False
@pytest.fixture
def app(request):
wtm = WebTestMixin()
wtm._patch_settings()
request.addfinalizer(wtm._unpatch_settings)
return django_webtest.DjangoTestApp()
@pytest.fixture
def concurrency(settings):
'''Select a level of concurrency based on the db, sqlite3 is less robust
thant postgres due to its transaction lock timeout of 5 seconds.
'''
if 'sqlite' in settings.DATABASES['default']['ENGINE']:
return 20
else:
return 100
@pytest.fixture
def private_settings(request):
import django.conf
from django.conf import UserSettingsHolder
old = django.conf.settings._wrapped
django.conf.settings._wrapped = UserSettingsHolder(old)
def finalizer():
django.conf.settings._wrapped = old
request.addfinalizer(finalizer)
return django.conf.settings
@pytest.fixture
def caplog(caplog):
import py.io
caplog.setLevel(logging.INFO)
caplog.handler.stream = py.io.TextIO()
caplog.handler.records = []
return caplog
@pytest.fixture
def john_doe(db):
user = User(
username='john.doe',
first_name='John',
last_name='Doe',
email='john.doe@example.com')
user.set_password('john.doe')
user.save()
return user
@pytest.fixture
def jane_doe(db):
user = User(
username='jane.doe',
first_name='Jane',
last_name='Doe',
email='jane.doe@example.com')
user.set_password('jane.doe')
user.save()
return user
@pytest.fixture
def admin_user(db):
try:
user = User.objects.get(username='admin')
except User.DoesNotExist:
user = User.objects.create_superuser('admin', email=None, password='admin')
return user

View File

@ -1,4 +1,18 @@
import os
DEBUG = False
LANGUAGE_CODE = 'en-US'
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.' + os.environ.get('DB_ENGINE', 'sqlite3'),
'NAME': 'mellon.sqlite3',
'TEST': {
'NAME': 'mellon-test-' + os.environ.get('DB_ENGINE', 'sqlite3'),
},
}
}
import tempfile
MEDIA_ROOT = tempfile.mkdtemp('fargo-test')

87
tests/test_api.py Normal file
View File

@ -0,0 +1,87 @@
import base64
import pytest
from django.core.urlresolvers import reverse
from django.utils.http import urlencode
from fargo.fargo import utils, models
from test_manager import login
pytestmark = pytest.mark.django_db
def test_create_validation(settings, app, admin_user, john_doe, jane_doe):
login(app)
data = {
'user_email': john_doe.email,
}
schema = utils.get_document_type_schema(settings, 'justificatif-de-domicile')
url = reverse('fargo-api-validation-list', kwargs={'document_type': 'justificatif-de-domicile'})
assert models.Validation.objects.count() == 0
response = app.post_json(url, data, status=400)
assert response.json['result'] == 0
assert set(response.json['errors'].keys()) == set([field['name'] for field in
schema['metadata']] + ['creator', 'origin'])
assert models.Validation.objects.count() == 0
data.update({
'nom': 'Doe',
'prenoms': 'John',
'numero': '189',
'rue': 'rue du bois',
'code-postal': '75014',
'ville': 'Paris',
'creator': 'FooBar',
'origin': 'wcs.example.com',
})
response1 = app.post_json(url, data, status=201)
assert set(response1.json.keys()) == set(['result', 'data'])
assert response1.json['result'] == 1
assert set(data.keys()) - set(['user_email']) < set(response1.json['data'].keys())
assert models.Validation.objects.count() == 1
data['user_email'] = jane_doe.email
response2 = app.post_json(url, data, status=201)
assert models.Validation.objects.count() == 2
response3 = app.get(url)
assert response3.json['data']['count'] == 2
response4 = app.get(url + '?%s' % urlencode({'user_email': john_doe.email}))
assert response4.json['data']['count'] == 1
response5 = app.get(response2.json['data']['url'], status=200)
assert response5.json['result'] == 1
assert response5.json['data'] == response2.json['data']
def test_push_document(app, admin_user, john_doe):
login(app)
data = {
'user_email': john_doe.email,
}
url = reverse('fargo-api-push-document')
# error checking test
assert models.Origin.objects.count() == 0
assert models.UserDocument.objects.count() == 0
assert models.Document.objects.count() == 0
response = app.post_json(url, data, status=400)
assert models.Origin.objects.count() == 0
assert models.UserDocument.objects.count() == 0
assert models.Document.objects.count() == 0
assert response.json['result'] == 0
assert (set(response.json['errors'].keys())
== set(['origin', 'file_b64_content', 'document_type']))
data.update({
'origin': 'wcs',
'file_b64_content': base64.b64encode('coin'),
'file_name': 'monfichier.pdf',
'document_type': 'justificatif-de-domicile',
})
response = app.post_json(url, data, status=200)
assert response.json['result'] == 1
assert models.Origin.objects.count() == 1
assert models.Origin.objects.get().label == 'wcs'
assert models.Origin.objects.get().slug == 'wcs'
assert models.UserDocument.objects.count() == 1
assert models.UserDocument.objects.get().user == john_doe
assert models.Document.objects.count() == 1
assert models.Document.objects.get().content.read() == 'coin'
assert (models.UserDocument.objects.get().document
== models.Document.objects.get())
assert (models.UserDocument.objects.get().origin
== models.Origin.objects.get())

View File

@ -1,28 +1,12 @@
import base64
import os
import StringIO
from django.core.files.storage import default_storage
from django.conf import settings
from django.contrib.auth.models import User
import pytest
from webtest import TestApp
pytestmark = pytest.mark.django_db
@pytest.fixture
def admin_user():
try:
user = User.objects.get(username='admin')
except User.DoesNotExist:
user = User.objects.create_superuser('admin', email=None, password='admin')
return user
def login(app, username='admin', password='admin'):
def login(app, username='admin', password='admin', user=None):
login_page = app.get('/login/')
login_form = login_page.forms[0]
login_form['username'] = username
login_form['password'] = password
if user:
login_form['username'] = user.username
login_form['password'] = user.username
else:
login_form['username'] = username
login_form['password'] = password
resp = login_form.submit()
assert resp.status_int == 302
return app

View File

@ -1,4 +1,4 @@
from webtest import TestApp
from webtest import TestApp, Upload
import pytest
import urlparse
@ -6,8 +6,18 @@ from fargo.wsgi import application
pytestmark = pytest.mark.django_db
from test_manager import admin_user, login
from test_manager import login
def test_unlogged():
app = TestApp(application)
assert urlparse.urlparse(app.get('/', status=302).location).path == '/login/'
def test_upload(app, john_doe):
login(app, user=john_doe)
response1 = app.get('/')
form = response1.form
form['content'] = Upload('monfichier.pdf', 'coin', 'application/pdf')
response2 = form.submit().follow()
assert 'monfichier.pdf' in response2.content