Moved task management code from flask views to task manager.

attract/__init__.py

@@ -1,18 +1,22 @@
 import logging
 
+import flask
+from werkzeug.local import LocalProxy
 from pillar.extension import PillarExtension
 
-from . import task_manager
+import attract.task_manager
+
+EXTENSION_NAME = 'attract'
 
 
 class AttractExtension(PillarExtension):
     def __init__(self):
         self._log = logging.getLogger('%s.AttractExtension' % __name__)
-        self.task_manager = task_manager.TaskManager()
+        self.task_manager = attract.task_manager.TaskManager()
 
     @property
     def name(self):
-        return 'attract'
+        return EXTENSION_NAME
 
     def flask_config(self):
         """Returns extension-specific defaults for the Flask configuration.
@@ -68,3 +72,14 @@ class AttractExtension(PillarExtension):
         from . import subversion
 
         subversion.task_logged.connect(self.task_manager.task_logged_in_svn)
+
+
+def _get_task_manager():
+    """Returns the Attract task manager of the current application."""
+
+    current_attract = flask.current_app.pillar_extensions[EXTENSION_NAME]
+    return current_attract.task_manager
+
+
+current_task_manager = LocalProxy(_get_task_manager)
+"""Attract Task manager of the current app."""

attract/task_manager.py

@@ -1,6 +1,10 @@
 """Task management."""
 
 import attr
+import flask_login
+
+import pillarsdk
+from pillar.web.system_util import pillar_api
 
 from . import attrs_extra
 
@@ -19,3 +23,52 @@ class TaskManager(object):
         """
 
         self._log.info("Task '%s' logged in SVN: %s", task_id, log_entry)
+
+    def create_task(self, project):
+        """Creates a new task, owned by the current user.
+
+        :rtype: pillarsdk.Node
+        """
+
+        api = pillar_api()
+        node_type = project.get_node_type('attract.task')
+        if not node_type:
+            raise ValueError('Project %s not set up for Attract' % project._id)
+
+        node_props = dict(
+            name='New task',
+            project=project['_id'],
+            user=flask_login.current_user.objectid,
+            node_type=node_type['name'],
+            properties={
+                'status': node_type['dyn_schema']['status']['default'],
+            },
+        )
+
+        task = pillarsdk.Node(node_props)
+        task.create(api=api)
+        return task
+
+    def edit_task(self, task_id, **fields):
+        """Edits a task.
+
+        :type task_id: str
+        :type fields: dict
+        :rtype: pillarsdk.Node
+        """
+
+        api = pillar_api()
+        task = pillarsdk.Node.find(task_id, api=api)
+
+        task.name = fields.pop('name')
+        task.description = fields.pop('description')
+        task.properties.status = fields.pop('status')
+
+        self._log.info('Saving task %s', task.to_dict())
+
+        if fields:
+            self._log.warning('edit_task(%r, ...) called with unknown fields %r; ignoring them.',
+                              task_id, fields)
+
+        task.update(api=api)
+        return task

attract/tasks.py

@@ -1,6 +1,6 @@
 import logging
 
-from flask import Blueprint, render_template, request
+from flask import Blueprint, render_template, request, current_app
 import flask
 import flask_login
 
@@ -8,11 +8,13 @@ import pillarsdk
 from pillar.web.system_util import pillar_api
 
 from .modules import attract_project_view
+from . import current_task_manager
 
 blueprint = Blueprint('attract.tasks', __name__, url_prefix='/tasks')
 log = logging.getLogger(__name__)
 
 
+
 @blueprint.route('/')
 def index():
     return render_template('attract/tasks/index.html')
@@ -52,13 +54,8 @@ def view_embed_task(project, task_id):
 def save(project, task_id):
     log.info('Saving task %s', task_id)
     log.debug('Form data: %s', request.form)
-    api = pillar_api()
-    task = pillarsdk.Node.find(task_id, api=api)
 
-    task.name = request.form['name']
-    task.description = request.form['description']
-    task.properties.status = request.form['status']
-    task.update(api=api)
+    task = current_task_manager.edit_task(task_id, **request.form.to_dict())
 
     return flask.jsonify({'task_id': task_id, 'etag': task._etag})
 
@@ -66,21 +63,7 @@ def save(project, task_id):
 @blueprint.route('/<project_url>/create')
 @attract_project_view()
 def create_task(project):
-    api = pillar_api()
-    node_type = project.get_node_type('attract.task')
-
-    node_props = dict(
-        name='New task',
-        project=project['_id'],
-        user=flask_login.current_user.objectid,
-        node_type=node_type['name'],
-        properties={
-            'status': node_type['dyn_schema']['status']['default'],
-        },
-    )
-
-    task = pillarsdk.Node(node_props)
-    task.create(api=api)
+    task = current_task_manager.create_task(project)
 
     resp = flask.make_response()
     resp.headers['Location'] = flask.url_for('attract.tasks.view_embed_task',

tests/test_tasks.py

@@ -0,0 +1,35 @@
+import responses
+from bson import ObjectId
+
+import pillarsdk
+import pillar.tests
+import pillar.auth
+import pillar.tests.common_test_data as ctd
+
+from abstract_attract_test import AbstractAttractTest
+
+
+class TaskWorkflowTest(AbstractAttractTest):
+    def setUp(self, **kwargs):
+        AbstractAttractTest.setUp(self, **kwargs)
+
+        self.mngr = self.app.pillar_extensions['attract'].task_manager
+        self.proj_id, self.project = self.ensure_project_exists()
+
+        self.sdk_project = pillarsdk.Project(pillar.tests.mongo_to_sdk(self.project))
+
+    @responses.activate
+    def test_create_task(self):
+        with self.app.test_request_context():
+            # Log in as project admin user
+            pillar.auth.login_user(ctd.EXAMPLE_PROJECT_OWNER_ID)
+
+            self.mock_blenderid_validate_happy()
+            task = self.mngr.create_task(self.sdk_project)
+            self.assertIsNotNone(task)
+
+        # Test directly with MongoDB
+        with self.app.test_request_context():
+            nodes_coll = self.app.data.driver.db['nodes']
+            found = nodes_coll.find_one(ObjectId(task['_id']))
+            self.assertIsNotNone(found)