Deducting asset node content type from file content type.

pillar/application/modules/nodes.py

@@ -2,6 +2,7 @@ import logging
 
 from bson import ObjectId
 from flask import current_app
+from werkzeug.exceptions import UnprocessableEntity
 
 from application.modules import file_storage
 from application.utils.authorization import check_permissions
@@ -73,6 +74,7 @@ def after_replacing_node(item, original):
     """Push an update to the Algolia index when a node item is updated. If the
     project is private, prevent public indexing.
     """
+
     projects_collection = current_app.data.driver.db['projects']
     project = projects_collection.find_one({'_id': item['project']},
                                            {'is_private': 1})
@@ -153,6 +155,40 @@ def after_inserting_nodes(items):
         )
 
 
+def deduct_content_type(node_doc, original):
+    """Deduct the content type from the attached file, if any."""
+
+    if node_doc['node_type'] != 'asset':
+        log.debug('deduct_content_type: called on node type %r, ignoring', node_doc['node_type'])
+        return
+
+    node_id = node_doc['_id']
+    try:
+        file_id = ObjectId(node_doc['properties']['file'])
+    except KeyError:
+        log.warning('deduct_content_type: Asset without properties.file, rejecting.')
+        raise UnprocessableEntity('Missing file property for asset node')
+
+    files = current_app.data.driver.db['files']
+    file_doc = files.find_one({'_id': file_id},
+                              {'content_type': 1})
+    if not file_doc:
+        log.warning('deduct_content_type: Node %s refers to non-existing file %s, rejecting.',
+                    node_id, file_id)
+        raise UnprocessableEntity('File property refers to non-existing file')
+
+    # Guess the node content type from the file content type
+    file_type = file_doc['content_type']
+    if file_type.startswith('video/'):
+        content_type = 'video'
+    elif file_type.startswith('image/'):
+        content_type = 'image'
+    else:
+        content_type = 'file'
+
+    node_doc['properties']['content_type'] = content_type
+
+
 def setup_app(app):
     from application import before_returning_item_permissions, before_returning_resource_permissions
 
@@ -166,3 +202,5 @@ def setup_app(app):
     app.on_replaced_nodes += after_replacing_node
     app.on_insert_nodes += before_inserting_nodes
     app.on_inserted_nodes += after_inserting_nodes
+
+    app.on_replace_nodes += deduct_content_type

tests/test_nodes.py

@@ -0,0 +1,70 @@
+from bson import ObjectId
+from eve.methods.post import post_internal
+from eve.methods.put import put_internal
+from flask import g
+from werkzeug.exceptions import UnprocessableEntity
+
+from common_test_class import AbstractPillarTest
+
+
+class NodeContentTypeTest(AbstractPillarTest):
+    def test_node_types(self):
+        """Tests that the node's content_type properties is updated correctly from its file."""
+
+        def mkfile(file_id, content_type):
+            file_id, _ = self.ensure_file_exists(file_overrides={
+                '_id': ObjectId(file_id),
+                'content_type': content_type})
+            return file_id
+
+        file_id_image = mkfile('cafef00dcafef00dcafef00d', 'image/jpeg')
+        file_id_video = mkfile('cafef00dcafef00dcafecafe', 'video/matroska')
+        file_id_blend = mkfile('cafef00dcafef00ddeadbeef', 'application/x-blender')
+
+        user_id = self.create_user()
+        project_id, _ = self.ensure_project_exists()
+
+        def perform_test(file_id, expected_type):
+            node_doc = {'picture': file_id_image,
+                        'description': '',
+                        'project': project_id,
+                        'node_type': 'asset',
+                        'user': user_id,
+                        'properties': {'status': 'published',
+                                       'tags': [],
+                                       'order': 0,
+                                       'categories': ''},
+                        'name': 'My first test node'}
+
+            with self.app.test_request_context():
+                g.current_user = {'user_id': user_id,
+                                  # This group is hardcoded in the EXAMPLE_PROJECT.
+                                  'groups': [ObjectId('5596e975ea893b269af85c0e')],
+                                  'roles': {u'subscriber', u'admin'}}
+                nodes = self.app.data.driver.db['nodes']
+
+                # Create the node.
+                r, _, _, status = post_internal('nodes', node_doc)
+                self.assertEqual(status, 201, r)
+                node_id = r['_id']
+
+                # Get from database to check its default content type.
+                db_node = nodes.find_one(node_id)
+                self.assertNotIn('content_type', db_node['properties'])
+
+                # PUT it again, without a file -- should be blocked.
+                self.assertRaises(UnprocessableEntity, put_internal, 'nodes', node_doc,
+                                  _id=node_id)
+
+                # PUT it with a file.
+                node_doc['properties']['file'] = str(file_id)
+                r, _, _, status = put_internal('nodes', node_doc, _id=node_id)
+                self.assertEqual(status, 200, r)
+
+                # Get from database to test the final node.
+                db_node = nodes.find_one(node_id)
+                self.assertEqual(expected_type, db_node['properties']['content_type'])
+
+        perform_test(file_id_image, 'image')
+        perform_test(file_id_video, 'video')
+        perform_test(file_id_blend, 'file')