# -*- encoding: utf-8 -*- import responses from bson import ObjectId import pillarsdk import pillarsdk.exceptions as sdk_exceptions from pillarsdk.utils import remove_private_keys 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) import pillar.tests 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 self.login_as_admin() 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 def login_as_admin(self): self.login_api_as(ctd.EXAMPLE_PROJECT_OWNER_ID) @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 self.login_as_admin() 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 self.login_as_admin() 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 self.login_as_admin() 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 import pillar.auth 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 import pillar.auth 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))