Produits Entr'ouvert » Passerelle

    python-cmislib,

# -*- coding: utf-8 -*-

from __future__ import unicode_literals

from django.db import models, migrations

class Migration(migrations.Migration):

    dependencies = [

        ('base', '0005_resourcelog'),

    ]

    operations = [

        migrations.CreateModel(

            name='CmisConnector',

            fields=[

                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),

                ('title', models.CharField(max_length=50)),

                ('slug', models.SlugField()),

                ('description', models.TextField()),

                ('log_level', models.CharField(default=b'INFO', max_length=10, verbose_name='Log Level', choices=[(b'NOTSET', b'NOTSET'), (b'DEBUG', b'DEBUG'), (b'INFO', b'INFO'), (b'WARNING', b'WARNING'), (b'ERROR', b'ERROR'), (b'CRITICAL', b'CRITICAL'), (b'FATAL', b'FATAL')])),

                ('cmis_endpoint', models.CharField(help_text='URL of the CMIS endpoint', max_length=250, verbose_name='CMIS endpoint')),

                ('username', models.CharField(max_length=128, verbose_name='Service username')),

                ('password', models.CharField(max_length=128, verbose_name='Password')),

                ('users', models.ManyToManyField(to='base.ApiUser', blank=True)),

            ],

            options={

                'verbose_name': 'CMIS connector',

            },

        ),

    ]

# passerelle.apps.cmis

# Copyright (C) 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

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# 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/>.

from cmislib import CmisClient

from cmislib.model import (PermissionDeniedException, ObjectNotFoundException,

                           UpdateConflictException)

import json

import base64

import logging

from django.db import models

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

class CmisConnector(BaseResource):

    cmis_endpoint = models.CharField(max_length=250, verbose_name=_('CMIS endpoint'),

                                     help_text=_('URL of the CMIS endpoint'))

    username = models.CharField(max_length=128, verbose_name=_('Service username'))

    password = models.CharField(max_length=128, verbose_name=_('Password'))

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('CMIS connector')

    @endpoint(serializer_type='json-api', methods=['post'], perm='can_upload_file')

    def upload_file(self, request, **kwargs):

        try:

            data = json.loads(request.body)

        except ValueError:

            raise APIError('CmisError : Invalid JSON string sent.')

        if not set(data.keys()) >= set(['filename', 'path', 'content', 'contentType']):

            raise APIError('CmisError : missing keys %s' % (set(['filename', 'path', 'content', 'contentType']) -

                                                     set(data.keys())))

        title = data['filename']

        path = data['path'].encode('utf-8')

        if not path:

            path = '/'

        try:

            content = base64.b64decode(data['content'])

        except (ValueError, TypeError) as e:

            raise APIError('CmisError : content must be a base64 string')

        content_type = data['contentType']

        repo_name = kwargs.get('repo', False)

        repo = self.cmis_connection_repository(repo_name)

        try:

            folder = repo.getObjectByPath(path)

        except ObjectNotFoundException:

            raise APIError('CmisError : Path not found on platform.')

        try:

            doc = folder.createDocumentFromString(title, contentString=content, contentType=content_type)

        except UpdateConflictException:

            raise APIError('CmisError : The document already exists on platform.')

        return doc.properties

    def cmis_connection_repository(self, repo_name):

        try:

            cmis_client = CmisClient(self.cmis_endpoint, self.username, self.password)

        except PermissionDeniedException:

            raise APIError('CmisError : Wrong username or password to connect to platform.')

        except ObjectNotFoundException:

            raise APIError('CmisError : Platform endpoint not found.')

        if repo_name:

            for repo in cmis_client.getRepositories():

                if repo_name == repo['repositoryName']:

                    return cmis_client.getRepository(repo['repositoryId'])

        return cmis_client.defaultRepository

{% extends "passerelle/manage/service_view.html" %}

{% load i18n passerelle %}

{% block description %}

{% endblock %}

{% block endpoints %}

<p>

{% blocktrans %}

The API currently doesn't support all parameters and is limited to the JSON

format.

{% endblocktrans %}

</p>

{% endblock %}

{% block security %}

<p>

{% trans 'Upload is limited to the following API users:' %}

</p>

{% access_rights_table resource=object permission='can_upload_file' %}

{% endblock %}

    'passerelle.apps.cmis',

import pytest

import mock

import base64

import json

from django.contrib.contenttypes.models import ContentType

from django.core.urlresolvers import reverse

from passerelle.base.models import ApiUser, AccessRight

from passerelle.apps.cmis.models import CmisConnector

@pytest.fixture()

def setup(db):

    api = ApiUser.objects.create(username='all', keytype='', key='')

    conn = CmisConnector.objects.create(cmis_endpoint='http://cmis_endpoint.com/cmis', username='admin',

                                        password='admin', slug='slug-cmis')

    obj_type = ContentType.objects.get_for_model(conn)

    AccessRight.objects.create(codename='can_upload_file', apiuser=api,

                               resource_type=obj_type, resource_pk=conn.pk)

    return conn

fake_file = {

    'filename': 'test.txt',

    'path': '/a/path',

    'content': base64.b64encode('fake file content.'),

    'contentType': 'text/plain'

}

fake_file_error_one = {

    'path': '/a/path',

    'content': base64.b64encode('fake file content.'),

    'contentType': 'text/plain'

}

fake_file_error_two = {

    'filename': 'test.txt',

    'path': '/a/path',

    'content': 'fake file content.',

    'contentType': 'text/plain'

}

class MockedCmisDocument(mock.Mock):

    @property

    def properties(self):

        return {

            'cmis:baseTypeId': 'cmis:document',

            'cmis:contentStreamFileName': self.title,

            'cmis:contentStreamMimeType': self.contentType,

        }

class MockedCmisFolder(mock.Mock):

    def createDocumentFromString(self, title, contentString, contentType):

        return MockedCmisDocument(title=title, contentString=contentString, contentType=contentType)

class MockedCmisRepository(mock.Mock):

    def getObjectByPath(self, path):

        return MockedCmisFolder()

class MockedCmisClient(mock.Mock):

    @property

    def defaultRepository(self):

        return MockedCmisRepository()

@mock.patch('passerelle.apps.cmis.models.CmisClient')

def test_cmis_upload(mock_connection, app, setup):

    mock_connection.return_value = MockedCmisClient()

    response = app.post_json('/cmis/slug-cmis/upload_file', fake_file)

    body = json.loads(response.body)

    assert 'data' in body

    assert body['err'] == 0

    data = body['data']

    assert 'cmis:baseTypeId' in data

    assert 'cmis:contentStreamFileName' in data

    assert 'cmis:contentStreamMimeType' in data

    assert data['cmis:baseTypeId'] == 'cmis:document'

    assert data['cmis:contentStreamFileName'] == fake_file['filename']

    assert data['cmis:contentStreamMimeType'] == fake_file['contentType']

@mock.patch('passerelle.apps.cmis.models.CmisClient')

def test_cmis_upload_error_key(mock_connection, app, setup):

    mock_connection.return_value = MockedCmisClient()

    resp = app.post_json('/cmis/slug-cmis/upload_file', fake_file_error_one)

    assert 'err' in resp.json

    assert 'err_desc' in resp.json

    assert resp.json['err'] == 1

    assert 'CmisError : missing keys' in resp.json['err_desc']

@mock.patch('passerelle.apps.cmis.models.CmisClient')

def test_cmis_upload_error_base64(mock_connection, app, setup):

    mock_connection.return_value = MockedCmisClient()

    resp = app.post_json('/cmis/slug-cmis/upload_file', fake_file_error_two)

    assert 'err' in resp.json

    assert 'err_desc' in resp.json

    assert resp.json['err'] == 1

    assert 'CmisError : content must be a base64 string' in resp.json['err_desc']