Page MenuHome

test_tasks.py
No OneTemporary

File Metadata

Created
Fri, Sep 20, 2:09 PM

test_tasks.py

# -*- encoding: utf-8 -*-
import responses
from bson import ObjectId
import pillarsdk
import pillarsdk.exceptions as sdk_exceptions
import pillar.tests
import pillar.tests.common_test_data as ctd
from abstract_attract_test import AbstractAttractTest
from pillarsdk.utils import remove_private_keys
class TaskWorkflowTest(AbstractAttractTest):
def setUp(self, **kwargs):
AbstractAttractTest.setUp(self, **kwargs)
self.mngr = self.attract.task_manager
self.proj_id, self.project = self.ensure_project_exists()
self.sdk_project = pillarsdk.Project(pillar.tests.mongo_to_sdk(self.project))
def create_task(self, task_type=None, parent=None):
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, task_type=task_type, parent=parent)
self.assertIsInstance(task, pillarsdk.Node)
return task
@responses.activate
def test_create_task(self):
task = self.create_task(task_type='Just düüüh it')
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)
self.assertEqual('Just düüüh it', found['properties']['task_type'])
# Test it through the API
resp = self.get('/api/nodes/%s' % task['_id'])
found = resp.json()
self.assertEqual('Just düüüh it', found['properties']['task_type'])
@responses.activate
def test_edit_task(self):
task = self.create_task()
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()
self.assertRaises(sdk_exceptions.PreconditionFailed,
self.mngr.edit_task,
task._id,
task_type='je møder',
name='nööw name',
description='€ ≠ ¥',
status='todo',
_etag='jemoeder')
self.mngr.edit_task(task._id,
task_type='je møder',
name='nööw name',
description='€ ≠ ¥',
status='todo',
_etag=task._etag)
# 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.assertEqual('je møder', found['properties']['task_type'])
self.assertEqual('todo', found['properties']['status'])
self.assertEqual('nööw name', found['name'])
self.assertEqual('€ ≠ ¥', found['description'])
@responses.activate
def test_edit_activity(self):
"""Perform the edit, then check the resulting activity on the shot."""
task = self.create_task()
# Only change the name -- the activity should contain both the old and the new name.
old_name = task['name']
new_name = 'nööw name'
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()
self.mngr.edit_task(task._id,
task_type=task['properties'].task_type,
name=new_name,
description=task.description,
status='todo',
_etag=task._etag)
with self.app.test_request_context():
acts = self.attract.activities_for_node(task['_id'])
self.assertEqual(2, acts['_meta']['total']) # Creation + edit
edit_act = acts['_items'][1]
self.assertIn(old_name, edit_act['verb'])
self.assertIn(new_name, edit_act['verb'])
@responses.activate
def test_load_save_task(self):
"""Test for the Eve hooks -- we should be able to PUT what we GET."""
task_parent = self.create_task(task_type='Just düüüh it')
task_child = self.create_task(task_type='mamaaaah',
parent=task_parent['_id'])
self.create_valid_auth_token(ctd.EXAMPLE_PROJECT_OWNER_ID, 'token')
url = '/api/nodes/%s' % task_child['_id']
resp = self.get(url)
json_task = resp.json()
self.put(url,
json=remove_private_keys(json_task),
auth_token='token',
headers={'If-Match': json_task['_etag']})
@responses.activate
def test_delete_task(self):
task = self.create_task()
task_id = task['_id']
self.create_valid_auth_token(ctd.EXAMPLE_PROJECT_OWNER_ID, 'token')
node_url = '/api/nodes/%s' % task_id
self.get(node_url, auth_token='token')
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()
self.assertRaises(sdk_exceptions.PreconditionFailed,
self.mngr.delete_task,
task._id,
'jemoeder')
self.mngr.delete_task(task._id, task._etag)
# 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.assertTrue(found['_deleted'])
# Test with Eve
self.get(node_url, auth_token='token', expected_status=404)
@responses.activate
def test_delete_task_nonadmin(self):
from pillar.api.projects.utils import get_admin_group_id
from attract.tasks import routes
self.enter_app_context()
task = self.create_task()
task_id = task['_id']
# Create a project member who is not admin.
admin_gid = get_admin_group_id(self.proj_id)
self.create_user(6 * 'dafe',
roles=('subscriber', 'attract-user'),
groups=[admin_gid],
token='mortal-token')
task = self.get(f'/api/nodes/{task_id}', auth_token='mortal-token').get_json()
with self.app.test_request_context(method='DELETE', data={'etag': task['_etag']}):
pillar.auth.login_user('mortal-token', load_from_db=True)
resp, status_code = routes.delete(str(task_id))
self.assertEqual(status_code, 204)
self.assertEqual(resp, '')
# Test directly with MongoDB
nodes_coll = self.app.data.driver.db['nodes']
found = nodes_coll.find_one(ObjectId(task_id))
self.assertTrue(found['_deleted'])
# Test with Eve
self.get(f'/api/nodes/{task_id}', auth_token='mortal-token', expected_status=404)
@responses.activate
def test_delete_task_nonmember(self):
from attract.tasks import routes
self.enter_app_context()
task = self.create_task()
task_id = task['_id']
# Create a user who is not admin and not a project member
self.create_user(6 * 'dafe',
roles=('subscriber', 'attract-user'),
groups=[],
token='mortal-token')
with self.app.test_request_context(method='DELETE', data={'etag': task['_etag']}):
pillar.auth.login_user('mortal-token', load_from_db=True)
with self.assertRaises(sdk_exceptions.ForbiddenAccess):
routes.delete(str(task_id))
# Test directly with MongoDB
nodes_coll = self.app.data.driver.db['nodes']
found = nodes_coll.find_one(ObjectId(task_id))
self.assertFalse(found.get('_deleted', False))

Event Timeline