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,

            },

        ),

    ]

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'])

    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

    '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']