Produits Entr'ouvert » Passerelle

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

from __future__ import unicode_literals

from django.db import models, migrations

class Migration(migrations.Migration):

    dependencies = [

        ('base', '0002_auto_20151009_0326'),

    ]

    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'NOTSET', 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(help_text='Username on DMS platform', max_length=128, verbose_name='Service username')),

                ('password', models.CharField(help_text='Password on DMS platform', max_length=128, verbose_name='Password')),

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

            ],

            options={

                'abstract': False,

            },

        ),

    ]

# 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, CmisException as CmisLibException

import json

import base64

import logging

from django.db import models

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from django.utils.translation import ugettext_lazy as _

from django.http import HttpResponse

class CmisException(Exception):

    http_status = 200

    log_error = False

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'), help_text=_('Username on DMS platform'))

    password = models.CharField(max_length=128,

                                verbose_name=_('Password'), help_text=_('Password on DMS platform'))

    category = _('Business Process Connectors')

    @classmethod

    def get_icon_class(cls):

        return 'ressources'

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

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

        data = json.loads(request.body)

        title = data['filename']

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

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

        content_type = data['contentType']

        repo = self.cmis_connection()

        try:

            folder = repo.getObjectByPath(path)

        except ObjectNotFoundException:

            raise CmisException('Path not found on platform.')

        except Exception as e:

            raise CmisException(str(e))

        try:

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

        except UpdateConflictException:

            raise CmisException('The document already exists on platform.')

        except Exception as e:

            raise CmisException(str(e))

        return doc.properties

    def cmis_connection(self):

        try:

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

        except PermissionDeniedException:

            raise CmisException('Wrong username or password to connect to platform.')

        except ObjectNotFoundException:

            raise CmisException('Platform endpoint not found.')

        return cmis_client.defaultRepository

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

{% load i18n passerelle %}

{% 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

# 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 django.conf.urls import patterns, url

from .views import *

urlpatterns = patterns('',

    url(r'^(?P<slug>[\w-]+)/$', CmisConnectorDetailView.as_view(),

        name='cmisconnector-view'))

# 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 django.views.generic import DetailView

from .models import CmisConnector

class CmisConnectorDetailView(DetailView):

    model = CmisConnector

    template_name = 'cmis/cmisconnector_detail.html'

    'cmis',

import pytest

import mock

import base64

import ast

from cmis.models import CmisConnector

from django.contrib.contenttypes.models import ContentType

from passerelle.base.models import ApiUser, AccessRight

from django.core.urlresolvers import reverse

@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='cmis')

    obj_type = ContentType.objects.get_for_model(conn)

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

                               resource_type=obj_type, resource_pk=conn.pk)

    return conn

@pytest.fixture()

def url():

    return reverse('generic-endpoint', kwargs={

                  'connector': 'cmis', 'slug': 'cmis', 'endpoint': 'upload_file'})

fake_file = {

    'filename': 'test.txt',

    'path': '/a/path',

    'content': base64.b64encode('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('cmis.models.CmisClient')

def test_cmis_upload(mock_connection, app, setup, url):

    mock_connection.return_value = MockedCmisClient()

    response = app.post_json(url, fake_file, status=200)

    body = ast.literal_eval(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']