308 lines
13 KiB
Python
308 lines
13 KiB
Python
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')
|