# -*- encoding: utf-8 -*- import unittest import responses from bson import ObjectId import pillarsdk import pillarsdk.exceptions as sdk_exceptions import pillar.tests import pillar.auth import pillar.tests.common_test_data as ctd from abstract_attract_test import AbstractAttractTest class AbstractShotTest(AbstractAttractTest): def setUp(self, **kwargs): AbstractAttractTest.setUp(self, **kwargs) self.tmngr = self.app.pillar_extensions['attract'].task_manager self.smngr = self.app.pillar_extensions['attract'].shot_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, shot_id, task_type): 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.tmngr.create_task(self.sdk_project, parent=shot_id, task_type=task_type) self.assertIsInstance(task, pillarsdk.Node) return task def create_shot(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() shot = self.smngr.create_shot(self.sdk_project) self.assertIsInstance(shot, pillarsdk.Node) return shot class ShotManagerTest(AbstractShotTest): @responses.activate def test_tasks_for_shot(self): shot1 = self.create_shot() shot2 = self.create_shot() shot1_id = shot1['_id'] shot2_id = shot2['_id'] task1 = self.create_task(shot1_id, u'fx') task2 = self.create_task(shot1_id, u'fx') task3 = self.create_task(shot1_id, u'høken') task4 = self.create_task(shot2_id, u'effects') task5 = self.create_task(shot2_id, u'effects') task6 = self.create_task(shot2_id, u'ïnžane') 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() shot_id_to_task = self.smngr.tasks_for_shots([shot1, shot2], [u'fx', u'høken', u'effects']) # Just test based on task IDs, as strings are turned into datetimes etc. by the API, # so we can't test equality. for all_tasks in shot_id_to_task.values(): for task_type, tasks in all_tasks.items(): all_tasks[task_type] = {task['_id'] for task in tasks} self.assertEqual({ u'fx': {task1['_id'], task2['_id']}, u'høken': {task3['_id']}, }, shot_id_to_task[shot1_id]) self.assertEqual({ u'effects': {task4['_id'], task5['_id']}, None: {task6['_id']}, }, shot_id_to_task[shot2_id]) @responses.activate def test_edit_shot(self): shot = self.create_shot() 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.smngr.edit_shot, shot_id=shot['_id'], name=u'ผัดไทย', description=u'Shoot the Pad Thai', status='todo', _etag='jemoeder') self.smngr.edit_shot(shot_id=shot['_id'], name=u'ผัดไทย', description=u'Shoot the Pad Thai', status='todo', _etag=shot._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(shot['_id'])) self.assertEqual(u'ผัดไทย', found['name']) self.assertEqual(u'todo', found['properties']['status']) self.assertEqual(u'Shoot the Pad Thai', found['description']) self.assertNotIn(u'notes', found['properties']) class NodeSetattrTest(unittest.TestCase): def test_simple(self): from attract.shots import node_setattr node = {} node_setattr(node, 'a', 5) self.assertEqual({'a': 5}, node) node_setattr(node, 'b', {'complexer': 'value'}) self.assertEqual({'a': 5, 'b': {'complexer': 'value'}}, node) def test_dotted(self): from attract.shots import node_setattr node = {} self.assertRaises(KeyError, node_setattr, node, 'a.b', 5) node = {'b': {}} node_setattr(node, 'b.simple', 'value') self.assertEqual({'b': {'simple': 'value'}}, node) node_setattr(node, 'b.complex', {'yes': 'value'}) self.assertEqual({'b': {'simple': 'value', 'complex': {'yes': 'value'}}}, node) node_setattr(node, 'b.complex', {'yes': 5}) self.assertEqual({'b': {'simple': 'value', 'complex': {'yes': 5}}}, node) def test_none_simple(self): from attract.shots import node_setattr node = {} node_setattr(node, 'a', None) node_setattr(node, None, 'b') self.assertEqual({None: 'b'}, node) def test_none_dotted(self): from attract.shots import node_setattr node = {} self.assertRaises(KeyError, node_setattr, node, 'a.b', None) node = {'b': {}} node_setattr(node, 'b.simple', None) self.assertEqual({'b': {}}, node) node_setattr(node, 'b.complex', {'yes': None}) self.assertEqual({'b': {'complex': {'yes': None}}}, node) node_setattr(node, 'b.complex.yes', None) self.assertEqual({'b': {'complex': {}}}, node) node_setattr(node, 'b.complex', {None: 5}) self.assertEqual({'b': {'complex': {None: 5}}}, node) class PatchShotTest(AbstractShotTest): @responses.activate def test_patch_from_blender_happy(self): shot = self.create_shot() self.create_valid_auth_token(ctd.EXAMPLE_PROJECT_OWNER_ID, 'token') url = '/api/nodes/%s' % shot._id patch = { 'op': 'from-blender', '$set': { 'properties.trim_start_in_frames': 123, 'properties.duration_in_edit_in_frames': 4215, 'properties.cut_in_timeline_in_frames': 1245, 'properties.status': 'todo', } } self.patch(url, json=patch, auth_token='token') @responses.activate def test_patch_bad_op(self): shot = self.create_shot() self.create_valid_auth_token(ctd.EXAMPLE_PROJECT_OWNER_ID, 'token') url = '/api/nodes/%s' % shot._id patch = {'properties.status': 'todo'} self.patch(url, json=patch, auth_token='token', expected_status=400) @responses.activate def test_patch_from_blender_bad_fields(self): shot = self.create_shot() self.create_valid_auth_token(ctd.EXAMPLE_PROJECT_OWNER_ID, 'token') url = '/api/nodes/%s' % shot._id patch = { 'op': 'from-blender', '$set': { 'invalid.property': 'JE MOEDER', } } self.patch(url, json=patch, auth_token='token', expected_status=400) @responses.activate def test_patch_from_blender_bad_status(self): shot = self.create_shot() self.create_valid_auth_token(ctd.EXAMPLE_PROJECT_OWNER_ID, 'token') url = '/api/nodes/%s' % shot._id patch = { 'op': 'from-blender', '$set': { 'properties.status': 'JE MOEDER', } } self.patch(url, json=patch, auth_token='token', expected_status=422) @responses.activate def test_patch_unauthenticated(self): shot = self.create_shot() url = '/api/nodes/%s' % shot._id patch = { 'op': 'from-blender', '$set': { 'properties.status': 'in_progress', } } self.patch(url, json=patch, expected_status=403) @responses.activate def test_patch_bad_user(self): shot = self.create_shot() self.create_user(24 * 'a') self.create_valid_auth_token(24 * 'a', 'other') url = '/api/nodes/%s' % shot._id patch = { 'op': 'from-blender', '$set': { 'properties.status': 'in_progress', } } self.patch(url, json=patch, auth_token='other', expected_status=403) class RequiredAfterCreationTest(AbstractShotTest): """ This tests Pillar stuff, but requires attract_shot since that's what the required_after_creation=False was created for. Placing the test here was easier than creating a node type in Pillar specifically for this test case. Once we use that validator in Pillar itself, we can move this test there too. """ def test_create_shot(self): from attract.node_types import node_type_shot self.user_id = self.create_project_admin(self.project) self.create_valid_auth_token(self.user_id, 'token') node_type_name = node_type_shot['name'] shot = {'name': u'test shot', 'description': u'', 'properties': {u'trim_start_in_frames': 0, u'duration_in_edit_in_frames': 1, u'cut_in_timeline_in_frames': 0}, 'node_type': node_type_name, 'project': unicode(self.proj_id), 'user': unicode(self.user_id)} resp = self.post('/api/nodes', json=shot, auth_token='token', expected_status=201) info = resp.json() resp = self.get('/api/nodes/%(_id)s' % info, auth_token='token') json_shot = resp.json() self.assertEqual(node_type_shot['dyn_schema']['status']['default'], json_shot['properties']['status']) return json_shot # TODO: should test editing a shot as well, but I had issues with the PillarSDK # not handling deleting of properties.