From f853278f5a7f7f1c110c5cbb1a17e2b734bc6590 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sybren=20A=2E=20St=C3=BCvel?= Date: Thu, 8 Sep 2016 12:41:49 +0200 Subject: [PATCH] Moved task management code from flask views to task manager. --- attract/__init__.py | 21 +++++++++++++--- attract/task_manager.py | 53 +++++++++++++++++++++++++++++++++++++++++ attract/tasks.py | 27 ++++----------------- tests/test_tasks.py | 35 +++++++++++++++++++++++++++ 4 files changed, 111 insertions(+), 25 deletions(-) create mode 100644 tests/test_tasks.py diff --git a/attract/__init__.py b/attract/__init__.py index e8d3e83..b89582c 100644 --- a/attract/__init__.py +++ b/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.""" diff --git a/attract/task_manager.py b/attract/task_manager.py index ad98840..2cdc8c5 100644 --- a/attract/task_manager.py +++ b/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 diff --git a/attract/tasks.py b/attract/tasks.py index 95c0582..9ffeccc 100644 --- a/attract/tasks.py +++ b/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('//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', diff --git a/tests/test_tasks.py b/tests/test_tasks.py new file mode 100644 index 0000000..0cb34d0 --- /dev/null +++ b/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)