permissions: change permission class

invenio_files_rest/permissions.py

@@ -8,7 +8,7 @@
 
 """Permissions for files using Invenio-Access."""
 
-from invenio_access import DynamicPermission, action_factory
+from invenio_access import Permission, action_factory
 
 from .models import Bucket, MultipartObject, ObjectVersion
 
@@ -118,12 +118,12 @@ def permission_factory(obj, action):
         the action is global.
     :param action: The required action.
     :raises RuntimeError: If the object is unknown.
-    :returns: A :class:`invenio_access.permissions.DynamicPermission` instance.
+    :returns: A :class:`invenio_access.permissions.Permission` instance.
     """
     need_class = _action2need_map[action]
 
     if obj is None:
-        return DynamicPermission(need_class(None))
+        return Permission(need_class(None))
 
     arg = None
     if isinstance(obj, Bucket):
@@ -135,4 +135,4 @@ def permission_factory(obj, action):
     else:
         raise RuntimeError('Unknown object')
 
-    return DynamicPermission(need_class(arg))
+    return Permission(need_class(arg))

tests/conftest.py

@@ -22,7 +22,7 @@
 from flask_celeryext import FlaskCeleryExt
 from flask_menu import Menu
 from invenio_access import InvenioAccess
-from invenio_access.models import ActionUsers
+from invenio_access.models import ActionRoles, ActionUsers, Role
 from invenio_accounts import InvenioAccounts
 from invenio_accounts.testutils import create_test_user
 from invenio_accounts.views import blueprint as accounts_blueprint
@@ -346,6 +346,44 @@ def permissions(db, bucket):
     yield users
 
 
+@pytest.yield_fixture()
+def admin_user(db):
+    """Permission for admin users."""
+    perms = [
+        location_update_all,
+        bucket_read_all,
+        bucket_read_versions_all,
+        bucket_update_all,
+        bucket_listmultiparts_all,
+        object_read_all,
+        object_read_version_all,
+        object_delete_all,
+        object_delete_version_all,
+        multipart_read_all,
+        multipart_delete_all,
+        bucket_read_all,
+        object_read_all,
+        bucket_update_all,
+        object_delete_all,
+        multipart_read_all,
+        object_read_all,
+    ]
+
+    admin = Role(name='admin')
+
+    for perm in perms:
+        db.session.add(ActionRoles.allow(perm, role=admin))
+
+    admin_user = create_test_user(email='admin@invenio-software.org',
+                                  password='pass1',
+                                  active=True)
+    admin.users.append(admin_user)
+
+    db.session.commit()
+
+    yield admin_user
+
+
 @pytest.fixture()
 def get_md5():
     """Get MD5 of data."""

tests/test_views_multipart.py

@@ -66,8 +66,10 @@ def test_post_init(client, headers, permissions, bucket, get_json):
                 assert k in data
 
 
-def test_post_init_querystring(client, bucket, get_json):
+def test_post_init_querystring(client, bucket, get_json, admin_user):
     """Test init multipart upload."""
+    login_user(client, admin_user)
+
     res = client.post(
         obj_url(bucket),
         query_string='uploads&size=10&partSize=4',
@@ -84,8 +86,10 @@ def test_get_init_not_allowed(client, bucket, get_json):
     assert res.status_code == 405
 
 
-def test_post_invalid_partsizes(client, headers, bucket, get_json):
+def test_post_invalid_partsizes(client, headers, bucket, get_json, admin_user):
     """Test invalid multipart init."""
+    login_user(client, admin_user)
+
     # Part size too large
     res = client.post(
         obj_url(bucket), query_string='uploads', headers=headers,
@@ -105,8 +109,10 @@ def test_post_invalid_partsizes(client, headers, bucket, get_json):
     assert res.status_code == 400
 
 
-def test_post_size_limits(client, db, headers, bucket):
+def test_post_size_limits(client, db, headers, bucket, admin_user):
     """Test invalid multipart init."""
+    login_user(client, admin_user)
+
     bucket.quota_size = 100
     db.session.commit()
 
@@ -126,8 +132,10 @@ def test_post_size_limits(client, db, headers, bucket):
     assert res.status_code == 400
 
 
-def test_post_locked_bucket(client, db, headers, bucket, get_json):
+def test_post_locked_bucket(client, db, headers, bucket, get_json, admin_user):
     """Test invalid multipart init."""
+    login_user(client, admin_user)
+
     bucket.locked = True
     db.session.commit()
 
@@ -145,8 +153,10 @@ def test_post_locked_bucket(client, db, headers, bucket, get_json):
     assert res.status_code == 404
 
 
-def test_post_invalidkey(client, db, headers, bucket):
+def test_post_invalidkey(client, db, headers, bucket, admin_user):
     """Test invalid multipart init."""
+    login_user(client, admin_user)
+
     object_url = url_for(
         'invenio_files_rest.object_api',
         bucket_id=str(bucket.id),
@@ -206,8 +216,11 @@ def test_put_not_found(client, db, bucket, permissions, multipart,
     assert res.status_code == 404
 
 
-def test_put_wrong_sizes(client, db, bucket, multipart, multipart_url):
+def test_put_wrong_sizes(client, db, bucket, multipart, multipart_url,
+                         admin_user):
     """Test invalid part sizes."""
+    login_user(client, admin_user)
+
     cases = [
         b'a' * (multipart.chunk_size + 1),
         b'a' * (multipart.chunk_size - 1),
@@ -221,8 +234,11 @@ def test_put_wrong_sizes(client, db, bucket, multipart, multipart_url):
         assert res.status_code == 400
 
 
-def test_put_ngfileupload(client, db, bucket, multipart, multipart_url):
+def test_put_ngfileupload(client, db, bucket, multipart, multipart_url,
+                          admin_user):
     """Test invalid part sizes."""
+    login_user(client, admin_user)
+
     res = client.put(
         multipart_url,
         data={
@@ -239,8 +255,11 @@ def test_put_ngfileupload(client, db, bucket, multipart, multipart_url):
     assert res.status_code == 200
 
 
-def test_put_invalid_part_number(client, db, bucket, multipart, multipart_url):
+def test_put_invalid_part_number(client, db, bucket, multipart, multipart_url,
+                                 admin_user):
     """Test invalid part number."""
+    login_user(client, admin_user)
+
     data = b'a' * multipart.chunk_size
     for c in [400, 2000, 'a']:
         res = client.put(
@@ -250,8 +269,11 @@ def test_put_invalid_part_number(client, db, bucket, multipart, multipart_url):
         assert res.status_code == 400
 
 
-def test_put_completed_multipart(client, db, bucket, multipart, multipart_url):
+def test_put_completed_multipart(client, db, bucket, multipart, multipart_url,
+                                 admin_user):
     """Test uploading to a completed multipart upload."""
+    login_user(client, admin_user)
+
     multipart.completed = True
     db.session.commit()
 
@@ -262,8 +284,11 @@ def test_put_completed_multipart(client, db, bucket, multipart, multipart_url):
     assert res.status_code == 403
 
 
-def test_put_badstream(client, db, bucket, multipart, multipart_url, get_json):
+def test_put_badstream(client, db, bucket, multipart, multipart_url, get_json,
+                       admin_user):
     """Test uploading to a completed multipart upload."""
+    login_user(client, admin_user)
+
     client.put(
         multipart_url + '&partNumber={0}'.format(1),
         input_stream=BytesIO(b'a' * multipart.chunk_size),
@@ -313,15 +338,20 @@ def test_get(client, db, bucket, permissions, multipart, multipart_url,
             assert len(data['parts']) == 3
 
 
-def test_get_empty(client, multipart, multipart_url, get_json):
+def test_get_empty(client, multipart, multipart_url, get_json, admin_user):
     """Test get parts when empty."""
+    login_user(client, admin_user)
+
     data = get_json(client.get(multipart_url), code=200)
     assert len(data['parts']) == 0
     assert data['id'] == str(multipart.upload_id)
 
 
-def test_get_serialization(client, multipart, multipart_url, get_json):
+def test_get_serialization(client, multipart, multipart_url, get_json,
+                           admin_user):
     """Test get parts when empty."""
+    login_user(client, admin_user)
+
     client.put(
         multipart_url + '&partNumber={0}'.format(1),
         input_stream=BytesIO(b'a' * multipart.chunk_size),
@@ -393,8 +423,10 @@ def _mock_celery_result():
 
 
 def test_post_complete_fail(client, headers, bucket, multipart,
-                            multipart_url, parts, get_json):
+                            multipart_url, parts, get_json, admin_user):
     """Test completing multipart when merge fails."""
+    login_user(client, admin_user)
+
     # Mock celery task to emulate real usage.
     task_result = MagicMock()
     task_result.ready = MagicMock(side_effect=[False, False, True])
@@ -419,8 +451,10 @@ def test_post_complete_fail(client, headers, bucket, multipart,
 
 
 def test_post_complete_timeout(app, client, headers, bucket, multipart,
-                               multipart_url, parts, get_json):
+                               multipart_url, parts, get_json, admin_user):
     """Test completing multipart when merge fails."""
+    login_user(client, admin_user)
+
     max_rounds = int(
         app.config['FILES_REST_TASK_WAIT_MAX_SECONDS'] //
         app.config['FILES_REST_TASK_WAIT_INTERVAL'])

tests/test_views_objectversion.py

@@ -127,8 +127,11 @@ def test_last_modified_utc_conversion(client, headers, bucket, permissions):
     assert abs(get_resp.last_modified - updated) < timedelta(seconds=1)
 
 
-def test_get_unreadable_file(client, headers, bucket, objects, db):
+def test_get_unreadable_file(client, headers, bucket, objects, db,
+                             admin_user):
     """Test getting an object with an unreadable file."""
+    login_user(client, admin_user)
+
     obj = objects[0]
     assert obj.is_head
     obj.file.readable = False
@@ -290,8 +293,10 @@ def test_put_versioning(client, bucket, permissions, get_md5, get_json):
     (None, None, 200, None),
 ])
 def test_put_file_size_errors(client, db, bucket, quota_size, max_file_size,
-                              expected, err):
+                              expected, err, admin_user):
     """Test that file size errors are properly raised."""
+    login_user(client, admin_user)
+
     filedata = b'a' * 75
     object_url = url_for(
         'invenio_files_rest.object_api', bucket_id=bucket.id, key='test.txt')
@@ -315,7 +320,9 @@ def test_put_file_size_errors(client, db, bucket, quota_size, max_file_size,
         assert resp.status_code == 400
 
 
-def test_put_invalid_key(client, db, bucket):
+def test_put_invalid_key(client, db, bucket, admin_user):
+    login_user(client, admin_user)
+
     """Test invalid key name."""
     key = 'a' * 2000
     object_url = url_for(
@@ -326,8 +333,10 @@ def test_put_invalid_key(client, db, bucket):
     assert resp.status_code == 400
 
 
-def test_put_zero_size(client, bucket):
+def test_put_zero_size(client, bucket, admin_user):
     """Test zero size file."""
+    login_user(client, admin_user)
+
     object_url = url_for(
         'invenio_files_rest.object_api', bucket_id=bucket.id, key='test.txt')
 
@@ -336,8 +345,10 @@ def test_put_zero_size(client, bucket):
     assert resp.status_code == 400
 
 
-def test_put_deleted_locked(client, db, bucket):
+def test_put_deleted_locked(client, db, bucket, admin_user):
     """Test that file size errors are properly raised."""
+    login_user(client, admin_user)
+
     object_url = url_for(
         'invenio_files_rest.object_api', bucket_id=bucket.id, key='test.txt')
 
@@ -358,8 +369,10 @@ def test_put_deleted_locked(client, db, bucket):
     assert resp.status_code == 404
 
 
-def test_put_error(client, bucket):
+def test_put_error(client, bucket, admin_user):
     """Test upload - cancelled by user."""
+    login_user(client, admin_user)
+
     object_url = url_for(
         'invenio_files_rest.object_api', bucket_id=bucket.id, key='test.txt')
 
@@ -375,8 +388,10 @@ def test_put_error(client, bucket):
     assert len(list(opener.opendir(bucket.location.uri).walk('.'))) == 3
 
 
-def test_put_multipartform(client, bucket):
+def test_put_multipartform(client, bucket, admin_user):
     """Test upload via multipart/form-data."""
+    login_user(client, admin_user)
+
     object_url = url_for(
         'invenio_files_rest.object_api', bucket_id=bucket.id, key='test.txt')
 
@@ -457,7 +472,8 @@ def test_delete_versions(client, db, bucket, versions, permissions, user,
         )).status_code == 404
 
 
-def test_delete_locked_deleted(client, db, bucket, versions):
+def test_delete_locked_deleted(client, db, bucket, versions,
+                               admin_user):
     """Test a deleted/locked bucket."""
     obj = versions[0]
     object_url = url_for(
@@ -467,6 +483,8 @@ def test_delete_locked_deleted(client, db, bucket, versions):
     bucket.locked = True
     db.session.commit()
 
+    login_user(client, admin_user)
+
     # Latest version
     resp = client.delete(object_url)
     assert resp.status_code == 403
@@ -487,14 +505,16 @@ def test_delete_locked_deleted(client, db, bucket, versions):
     assert resp.status_code == 404
 
 
-def test_delete_unwritable(client, db, bucket, versions):
+def test_delete_unwritable(client, db, bucket, versions, admin_user):
     """Test deleting a file which is not writable."""
     obj = versions[0]
 
     # Unwritable file.
     obj.file.writable = False
     db.session.commit()
 
+    login_user(client, admin_user)
+
     # Delete specific version
     with patch('invenio_files_rest.views.remove_file_data') as task:
         resp = client.delete(url_for(