passerelle/tests/test_cmis.py

import base64
import httplib2
import urllib2

from cmislib import CmisClient
from cmislib.exceptions import CmisException
from cmislib.exceptions import ObjectNotFoundException
from cmislib.exceptions import PermissionDeniedException
from cmislib.exceptions import UpdateConflictException
from django.contrib.contenttypes.models import ContentType
from mock import call, Mock
import py
import pytest

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://example.com/cmisatom', username='admin', password='admin',
        slug='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


def test_uploadfile(app, setup, tmpdir, monkeypatch):

    class FakeCMISGateway(object):

        def __init__(self, *args, **kwargs):
            pass

        def create_doc(self, file_name, file_path, file_byte_content):
            with open(file_name, 'wb') as f:
                f.write(file_byte_content)
                return Mock(properties={"toto": "tata"})

    file_name = "testfile.whatever"
    file_content = 'aaaa'
    monkeypatch.chdir(tmpdir)
    import passerelle.apps.cmis.models
    monkeypatch.setattr(passerelle.apps.cmis.models, 'CMISGateway', FakeCMISGateway)
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"path": "/some/folder/structure",
                "file": {"filename": file_name,
                         "content": base64.b64encode(file_content),
                         "content_type": "image/jpeg"}})
    result_file = py.path.local(file_name)
    assert result_file.exists()
    with result_file.open('rb'):
        assert result_file.read() == file_content
    json_result = response.json_body
    assert json_result['err'] == 0
    assert json_result['data']['properties'] == {"toto": "tata"}


def test_uploadfile_error_if_no_file_name(app, setup):
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"path": "/some/folder/structure",
                "file": {"content": base64.b64encode('aaaa'), "content_type": "image/jpeg"}},
        expect_errors=True)
    assert response.status_code == 400
    assert response.json_body['err'] == 1
    assert response.json_body['err_desc'].startswith('"file[\'filename\']" is required')


def test_uploadfile_error_if_non_string_file_name(app, setup):
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"path": "/some/folder/structure",
                "file": {"filename": 1, "content": base64.b64encode('aaaa'),
                         "content_type": "image/jpeg"}},
        expect_errors=True)
    assert response.status_code == 400
    assert response.json_body['err'] == 1
    assert response.json_body['err_desc'].startswith('"file[\'filename\']" must be string')


def test_uploadfile_error_if_non_valid_file_name(app, setup):
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"path": "/some/folder/structure",
                "file": {"filename": ",.,", "content": base64.b64encode('aaaa'),
                         "content_type": "image/jpeg"}},
        expect_errors=True)
    assert response.status_code == 400
    assert response.json_body['err'] == 1
    assert response.json_body['err_desc'].startswith('"file[\'filename\']" must be valid file name')


def test_uploadfile_error_if_no_path(app, setup):
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"file": {"filename": 'somefile.txt', "content": base64.b64encode('aaaa'),
                         "content_type": "image/jpeg"}},
        expect_errors=True)
    assert response.status_code == 400
    assert response.json_body['err'] == 1
    assert response.json_body['err_desc'].startswith('"path" is required')


def test_uploadfile_error_if_non_string_path(app, setup):
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"path": 1,
                "file": {"filename": 'somefile.txt', "content": base64.b64encode('aaaa'),
                         "content_type": "image/jpeg"}},
        expect_errors=True)
    assert response.status_code == 400
    assert response.json_body['err'] == 1
    assert response.json_body['err_desc'].startswith('"path" must be string')


def test_uploadfile_error_if_no_regular_path(app, setup):
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"path": "no/leading/slash",
                "file": {"filename": 'somefile.txt', "content": base64.b64encode('aaaa'),
                         "content_type": "image/jpeg"}},
        expect_errors=True)
    assert response.status_code == 400
    assert response.json_body['err'] == 1
    assert response.json_body['err_desc'].startswith('"path" must be valid path')


def test_uploadfile_error_if_no_file_content(app, setup):
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"path": "/some/folder/structure",
                "file": {"filename": 'somefile.txt', "content_type": "image/jpeg"}},
        expect_errors=True)
    assert response.status_code == 400
    assert response.json_body['err'] == 1
    assert response.json_body['err_desc'].startswith('"file[\'content\']" is required')


def test_uploadfile_error_if_non_string_file_content(app, setup):
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"path": "/some/folder/structure",
                "file": {"filename": 'somefile.txt', "content": 1, "content_type": "image/jpeg"}},
        expect_errors=True)
    assert response.status_code == 400
    assert response.json_body['err'] == 1
    assert response.json_body['err_desc'].startswith('"file[\'content\']" must be string')


def test_uploadfile_error_if_no_proper_base64_encoding(app, setup):
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"path": "/some/folder/structure",
                "file": {"filename": 'somefile.txt', "content": "1", "content_type": "image/jpeg"}},
        expect_errors=True)
    assert response.status_code == 400
    assert response.json_body['err'] == 1
    assert response.json_body['err_desc'].startswith(
        '"file[\'content\']" must be a valid base64 string')


def test_uploadfile_cmis_gateway_error(app, setup, monkeypatch):
    from passerelle.utils.jsonresponse import APIError
    cmis_gateway = Mock()
    cmis_gateway.create_doc.side_effect = APIError("some error")
    cmis_gateway_cls = Mock(return_value=cmis_gateway)
    import passerelle.apps.cmis.models
    monkeypatch.setattr(passerelle.apps.cmis.models, 'CMISGateway', cmis_gateway_cls)
    response = app.post_json(
        '/cmis/slug-cmis/uploadfile',
        params={"path": "/some/folder/structure",
                "file": {"filename": "file_name", "content": base64.b64encode('aaaa'),
                         "content_type": "image/jpeg"}})
    assert response.json_body['err'] == 1
    assert response.json_body['err_desc'].startswith("some error")


def test_get_or_create_folder_already_existing(monkeypatch):
    default_repository = Mock()
    default_repository.getObjectByPath.return_value = 'folder'
    cmis_client_cls = Mock(
        return_value=Mock(spec=CmisClient, defaultRepository=default_repository))
    import passerelle.apps.cmis.models
    monkeypatch.setattr(passerelle.apps.cmis.models, 'CmisClient', cmis_client_cls)
    gateway = passerelle.apps.cmis.models.CMISGateway('cmis_endpoint', 'user', 'pass', Mock())
    assert gateway._get_or_create_folder('/whatever') == 'folder'
    default_repository.getObjectByPath.assert_has_calls([call('/whatever')])


def test_get_or_create_folder_one_level_creation(monkeypatch):
    root_folder = Mock()
    root_folder.createFolder.return_value = 'folder'
    default_repository = Mock(
        rootFolder=root_folder, **{'getObjectByPath.side_effect': ObjectNotFoundException()})
    cmis_client_cls = Mock(
        return_value=Mock(spec=CmisClient, defaultRepository=default_repository))
    import passerelle.apps.cmis.models
    monkeypatch.setattr(passerelle.apps.cmis.models, 'CmisClient', cmis_client_cls)
    gateway = passerelle.apps.cmis.models.CMISGateway('cmis-url', 'user', 'password', Mock())
    assert gateway._get_or_create_folder('/whatever') == 'folder'
    default_repository.getObjectByPath.assert_has_calls([call('/whatever'), call('/whatever')])
    root_folder.createFolder.assert_called_once_with('whatever')


def test_get_or_create_folder_two_level_creation(monkeypatch):
    whatever_folder = Mock()
    whatever_folder.createFolder.return_value = 'folder'
    root_folder = Mock()
    root_folder.createFolder.return_value = whatever_folder
    default_repository = Mock(rootFolder=root_folder)
    default_repository.getObjectByPath.side_effect = ObjectNotFoundException()
    cmis_client_cls = Mock(
        return_value=Mock(spec=CmisClient, defaultRepository=default_repository))
    import passerelle.apps.cmis.models
    monkeypatch.setattr(passerelle.apps.cmis.models, 'CmisClient', cmis_client_cls)
    gateway = passerelle.apps.cmis.models.CMISGateway('cmis_url', 'user', 'password', Mock())
    assert gateway._get_or_create_folder('/whatever/man') == 'folder'
    default_repository.getObjectByPath.assert_has_calls(
        [call('/whatever/man'), call('/whatever'), call('/whatever/man')])
    root_folder.createFolder.assert_called_once_with('whatever')
    whatever_folder.createFolder.assert_called_once_with('man')


def test_get_or_create_folder_with_some_existing_and_some_not(monkeypatch):
    whatever_folder = Mock()
    whatever_folder.createFolder.return_value = 'folder'

    def getObjectByPath(path):
        if path == '/whatever':
            return whatever_folder
        elif path == '/whatever/man':
            raise ObjectNotFoundException()
        else:
            raise Exception("I should not be called with: %s" % path)

    root_folder = Mock()
    default_repository = Mock(rootFolder=root_folder)
    default_repository.getObjectByPath.side_effect = getObjectByPath
    cmis_client_cls = Mock(
        return_value=Mock(spec=CmisClient, defaultRepository=default_repository))
    import passerelle.apps.cmis.models
    monkeypatch.setattr(passerelle.apps.cmis.models, 'CmisClient', cmis_client_cls)
    gateway = passerelle.apps.cmis.models.CMISGateway('cmis_url', 'user', 'password', Mock())
    assert gateway._get_or_create_folder('/whatever/man') == 'folder'
    root_folder.createFolder.assert_not_called()
    whatever_folder.createFolder.assert_called_once_with('man')


def test_create_doc():
    from passerelle.apps.cmis.models import CMISGateway
    gateway = CMISGateway('cmis_url', 'user', 'password', Mock())
    folder = Mock()
    folder.createDocument.return_value = 'doc'
    gateway._get_or_create_folder = Mock(return_value=folder)
    assert gateway.create_doc('filename', '/some/path', 'file_content') == 'doc'
    gateway._get_or_create_folder.assert_called_once_with('/some/path')
    args, kwargs = folder.createDocument.call_args
    assert args[0] == 'filename'
    content_file = kwargs['contentFile']
    assert content_file.read() == 'file_content'


@pytest.mark.parametrize("cmis_exc,err_msg", [
    (httplib2.HttpLib2Error, "connection error"),
    # FIXME used for cmslib 0.5 compat
    (urllib2.URLError, "connection error"),
    (PermissionDeniedException, "permission denied"),
    (UpdateConflictException, "update conflict"),
    (CmisException, "cmis binding error")
])
def test_wrap_cmis_error(app, setup, monkeypatch, cmis_exc, err_msg):
    from passerelle.utils.jsonresponse import APIError
    from passerelle.apps.cmis.models import wrap_cmis_error

    @wrap_cmis_error
    def dummy_func():
        raise cmis_exc("some error")

    with pytest.raises(APIError) as excinfo:
        dummy_func()
    assert str(excinfo.value).startswith(err_msg)


def test_re_file_path():
    from passerelle.apps.cmis.models import RE_FILE_PATH
    assert RE_FILE_PATH.match('/')
    assert RE_FILE_PATH.match('/some')
    assert RE_FILE_PATH.match('/some/path')
    assert RE_FILE_PATH.match('/SOME/PATH')
    assert RE_FILE_PATH.match('/some/long/path')
    assert RE_FILE_PATH.match('/some/digits/12/and/CAPITALS')
    assert RE_FILE_PATH.match('/some/!#$%&+-^_`~;[]{}+=~')
    assert not RE_FILE_PATH.match('/trailing/slash/')
    assert not RE_FILE_PATH.match('no/leading/slash')
    assert not RE_FILE_PATH.match('/multiple//slash')
    assert not RE_FILE_PATH.match('')


def test_re_file_name():
    from passerelle.apps.cmis.models import RE_FILE_NAME
    assert RE_FILE_NAME.match('toto.tata')
    assert RE_FILE_NAME.match('TOTO.TATA')