passerelle/tests/test_cmis.py

597 lines
22 KiB
Python

import base64
import os
import re
import xml.etree.ElementTree as ET
import httplib2
import mock
import py
import pytest
from cmislib import CmisClient
from cmislib.exceptions import (
CmisException,
InvalidArgumentException,
ObjectNotFoundException,
PermissionDeniedException,
UpdateConflictException,
)
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from django.utils.encoding import force_bytes, force_text
from django.utils.six.moves.urllib import error as urllib2
from mock import Mock, call
from test_manager import login
from passerelle.apps.cmis.models import CmisConnector
from passerelle.base.models import AccessRight, ApiUser
def b64encode(content):
return force_text(base64.b64encode(force_bytes(content)))
@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,
content_type=None,
object_type=None,
properties=None,
):
assert content_type == "image/jpeg"
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": 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
assert json_result['err'] == 0
assert json_result['data']['properties'] == {"toto": "tata"}
file_name_overwrite = "testfile.whatever.overwrite"
response = app.post_json(
'/cmis/slug-cmis/uploadfile',
params={
"path": "/some/folder/structure",
"file": {"filename": file_name, "content": b64encode(file_content), "content_type": "image/jpeg"},
"filename": file_name_overwrite,
},
)
result_file = py.path.local(file_name_overwrite)
assert result_file.exists()
with result_file.open('rb'):
assert result_file.read() == file_content
json_result = response.json
assert json_result['err'] == 0
assert json_result['data']['properties'] == {"toto": "tata"}
def test_upload_file_metadata(app, setup, monkeypatch):
class FakeFolder:
def createDocument(self, filename, contentFile, properties, contentType=None):
return Mock(properties=properties)
from passerelle.apps.cmis.models import CMISGateway
monkeypatch.setattr(CMISGateway, '_get_or_create_folder', lambda x, y: FakeFolder())
response = app.post_json(
'/cmis/slug-cmis/uploadfile',
params={
"path": "/some/folder/structure",
"file": {"filename": "bla", "content": b64encode('bla')},
"object_type": "D:dui:type",
"properties": {
"cmis:description": "Coucou",
"dui:tnumDossier": "42",
},
"properties/dui:ttypeStructure": "Accueil de loisirs",
},
)
assert response.json['data']['properties'] == {
"cmis:objectTypeId": "D:dui:type",
"cmis:description": "Coucou",
"dui:tnumDossier": "42",
"dui:ttypeStructure": "Accueil de loisirs",
}
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": b64encode('aaaa'), "content_type": "image/jpeg"},
},
expect_errors=True,
)
assert response.status_code == 400
assert response.json['err'] == 1
assert response.json['err_desc'].startswith('"filename" or "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": b64encode('aaaa'), "content_type": "image/jpeg"},
},
expect_errors=True,
)
assert response.status_code == 400
assert response.json['err'] == 1
assert response.json['err_desc'] == "file/filename: 1 is not of type 'string'"
response = app.post_json(
'/cmis/slug-cmis/uploadfile',
params={
"path": "/some/folder/structure",
"file": {"content": b64encode('aaaa'), "content_type": "image/jpeg"},
"filename": 1,
},
expect_errors=True,
)
assert response.status_code == 400
assert response.json['err'] == 1
assert response.json['err_desc'] == "filename: 1 is not of type '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": b64encode('aaaa'), "content_type": "image/jpeg"},
},
expect_errors=True,
)
assert response.status_code == 400
assert response.json['err'] == 1
assert "',.,' does not match " in response.json['err_desc']
response = app.post_json(
'/cmis/slug-cmis/uploadfile',
params={
"path": "/some/folder/structure",
"file": {"content": b64encode('aaaa'), "content_type": "image/jpeg"},
"filename": ",.,",
},
expect_errors=True,
)
assert response.status_code == 400
assert response.json['err'] == 1
assert "',.,' does not match " in response.json['err_desc']
def test_uploadfile_error_if_no_path(app, setup):
response = app.post_json(
'/cmis/slug-cmis/uploadfile',
params={
"file": {"filename": 'somefile.txt', "content": b64encode('aaaa'), "content_type": "image/jpeg"}
},
expect_errors=True,
)
assert response.status_code == 400
assert response.json['err'] == 1
assert response.json['err_desc'] == "'path' is a required property"
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": b64encode('aaaa'), "content_type": "image/jpeg"},
},
expect_errors=True,
)
assert response.status_code == 400
assert response.json['err'] == 1
assert response.json['err_desc'] == "path: 1 is not of type '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": b64encode('aaaa'), "content_type": "image/jpeg"},
},
expect_errors=True,
)
assert response.status_code == 400
assert response.json['err'] == 1
assert "'no/leading/slash' does not match " in response.json['err_desc']
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['err'] == 1
assert response.json['err_desc'] == "file: 'content' is a required property"
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['err'] == 1
assert response.json['err_desc'] == "file/content: 1 is not of type '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['err'] == 1
assert response.json['err_desc'].startswith('"file[\'content\']" must be a valid base64 string')
def test_upload_file_error_metadata(app, setup):
response = app.post_json(
'/cmis/slug-cmis/uploadfile',
params={
"path": "/some/folder/structure",
"file": {"filename": "bla", "content": b64encode('bla')},
"properties": {"dui:tnumDossier": "42"},
},
expect_errors=True,
)
assert response.status_code == 400
assert response.json['err'] == 1
assert 'object_type' in response.json['err_desc']
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": b64encode('aaaa'), "content_type": "image/jpeg"},
},
)
assert response.json['err'] == 1
assert response.json['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', b'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() == b'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"),
(InvalidArgumentException, "invalid property"),
(CmisException, "cmis binding error"),
],
)
def test_wrap_cmis_error(app, setup, monkeypatch, cmis_exc, err_msg):
from passerelle.apps.cmis.models import wrap_cmis_error
from passerelle.utils.jsonresponse import APIError
@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 FILE_PATH_PATTERN
RE_FILE_PATH = re.compile(FILE_PATH_PATTERN)
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 FILE_NAME_PATTERN
RE_FILE_NAME = re.compile(FILE_NAME_PATTERN)
assert RE_FILE_NAME.match('toto.tata')
assert RE_FILE_NAME.match('TOTO.TATA')
def test_cmis_types_view(setup, app, admin_user, monkeypatch):
class FakeCmisType:
class FakeCmisProperty:
def __init__(self, id):
self.id = id
self.description = 'cmis:prop'
self.propertyType = 'string'
self.required = True
def __init__(self, id, children=None):
self.id = id
self.description = 'hop'
prop = self.FakeCmisProperty('cmis:prop')
prop2 = self.FakeCmisProperty('cmis:prop2')
self.properties = {prop.id: prop, prop2.id: prop2}
self.children = children or []
class FakeCmisRepo:
def __init__(self, root_types):
self.root_types = root_types
def getTypeDefinition(self, id):
if id in self.root_types:
return self.root_types[id]
for type in self.root_types.values():
for child in type.children:
if child.id == id:
return child
raise ObjectNotFoundException
def getTypeDefinitions(self):
return self.root_types.values()
def getTypeChildren(self, id):
return self.getTypeDefinition(id).children
children = [FakeCmisType('cmis:child1'), FakeCmisType('cmis:child2')]
root_type1 = FakeCmisType('cmis:root1', children=children)
root_type2 = FakeCmisType('cmis:root2')
root_types = {root_type1.id: root_type1, root_type2.id: root_type2}
repo = FakeCmisRepo(root_types)
from cmislib import CmisClient
monkeypatch.setattr(CmisClient, 'getDefaultRepository', lambda self: repo)
app = login(app)
resp = app.get('/cmis/slug-cmis/')
resp = resp.click('Explore available object types')
assert all(id in resp.text for id in root_types)
assert not 'Back to' in resp.text
assert not 'Children' in resp.text
assert not 'Properties' in resp.text
resp = resp.click(root_type1.id)
assert all(id in resp.text for id in root_type1.properties)
assert all(child.id in resp.text for child in root_type1.children)
resp = resp.click(children[0].id)
assert "No more children." in resp.text
resp = resp.click("Back to base types list")
resp = resp.click(root_type2.id)
assert "No more children." in resp.text
resp = app.get('/manage/cmis/slug-cmis/type?id=wrong', status=404)
@mock.patch('httplib2.Http.request')
def test_raw_uploadfile(mocked_request, app, setup):
""" Simulate the bellow bash query :
$ http https://passerelle.dev.publik.love/cmis/ged/uploadfile \
file:='{"filename": "test2", "content": "c2FsdXQK"}' path=/test-eo
"""
file_name = "test2"
file_content = 'salut\n'
path = "/test-eo"
url = reverse(
'generic-endpoint', kwargs={'connector': 'cmis', 'endpoint': 'uploadfile', 'slug': setup.slug}
)
def cmis_mocked_request(uri, method="GET", body=None, **kwargs):
"""simulate the 3 (ordered) HTTP queries involved"""
response = {'status': '200'}
if method == 'GET' and uri == 'http://example.com/cmisatom':
content = open('%s/tests/data/cmis/cmis1.out.xml' % os.getcwd(), 'rb').read()
elif (
method == 'GET'
and uri
== 'http://example.com/cmisatom/test/path?path=/test-eo&filter=&includeAllowableActions=false&includeACL=false&includePolicyIds=false&includeRelationships=&renditionFilter='
):
content = open('%s/tests/data/cmis/cmis2.out.xml' % os.getcwd(), 'rb').read()
elif method == 'POST' and uri == 'http://example.com/cmisatom/test/children?id=L3Rlc3QtZW8%3D':
expected_input = open('%s/tests/data/cmis/cmis3.in.xml' % os.getcwd(), 'r').read()
expected_input = expected_input.replace('\n', '')
expected_input = re.sub('> *<', '><', expected_input)
input1 = ET.tostring(ET.XML(expected_input))
# reorder properties
input2 = ET.XML(body)
objects = input2.find('{http://docs.oasis-open.org/ns/cmis/restatom/200908/}object')
properties = objects.find('{http://docs.oasis-open.org/ns/cmis/core/200908/}properties')
data = []
for elem in properties:
key = elem.tag
data.append((key, elem))
data.sort()
properties[:] = [item[-1] for item in data]
input2 = ET.tostring(input2)
if input1 != input2:
raise Exception('expect [[%s]] but get [[%s]]' % (body, expected_input))
content = open('%s/tests/data/cmis/cmis3.out.xml' % os.getcwd(), 'rb').read()
else:
raise Exception('my fault error, url is not yet mocked: %s' % uri)
return (response, content)
mocked_request.side_effect = cmis_mocked_request
params = {
"path": path,
"file": {"filename": file_name, "content": b64encode(file_content), "content_type": "image/jpeg"},
}
response = app.post_json(url, params=params)
json_result = response.json
assert json_result['err'] == 0
assert json_result['data']['properties']['cmis:objectTypeId'] == "cmis:document"
assert json_result['data']['properties']['cmis:name'] == file_name
def test_cmis_check_status(app, setup, monkeypatch):
cmis_gateway = Mock()
type(cmis_gateway).repo = mock.PropertyMock(side_effect=CmisException)
cmis_gateway_cls = Mock(return_value=cmis_gateway)
import passerelle.apps.cmis.models
monkeypatch.setattr(passerelle.apps.cmis.models, 'CMISGateway', cmis_gateway_cls)
with pytest.raises(CmisException):
setup.check_status()