test_project_management.py
No OneTemporary
Actions

Size

26 KB

Subscribers

None

test_project_management.py
View Options

	# -- encoding: utf-8 --

	"""Unit tests for creating and editing projects_blueprint."""

	import functools
	import json
	import logging
	import urllib

	from bson import ObjectId

	from common_test_class import AbstractPillarTest

	log = logging.getLogger(__name__)


	class AbstractProjectTest(AbstractPillarTest):
	def _create_user_with_token(self, roles, token, user_id='cafef00df00df00df00df00d'):
	user_id = self.create_user(roles=roles, user_id=user_id)
	self.create_valid_auth_token(user_id, token)
	return user_id

	def _create_project(self, project_name, token):
	resp = self.client.post('/p/create',
	headers={'Authorization': self.make_header(token)},
	data={'project_name': project_name})
	return resp

	def _create_user_and_project(self, roles, user_id='cafef00df00df00df00df00d', token='token',
	project_name=u'Prøject El Niño'):
	self._create_user_with_token(roles, token, user_id=user_id)
	resp = self._create_project(project_name, token)

	self.assertEqual(201, resp.status_code, resp.data)
	project = json.loads(resp.data)

	return project


	class ProjectCreationTest(AbstractProjectTest):
	def test_project_creation_wrong_role(self):
	self._create_user_with_token([u'whatever'], 'token')
	resp = self._create_project(u'Prøject El Niño', 'token')

	self.assertEqual(403, resp.status_code)

	# Test that the project wasn't created.
	with self.app.test_request_context():
	projects = self.app.data.driver.db['projects']
	self.assertEqual(0, len(list(projects.find())))

	def test_project_creation_good_role(self):
	user_id = self._create_user_with_token([u'subscriber'], 'token')
	resp = self._create_project(u'Prøject El Niño', 'token')
	self.assertEqual(201, resp.status_code)

	# The response of a POST is the entire project, but we'll test a GET on
	# the returned Location nevertheless.
	project_info = json.loads(resp.data.decode('utf-8'))
	project_id = project_info['_id']

	# Test that the Location header contains the location of the project document.
	self.assertEqual('http://localhost/projects/%s' % project_id,
	resp.headers['Location'])

	# GET the project from the URL in the Location header to see if that works too.
	auth_header = {'Authorization': self.make_header('token')}
	resp = self.client.get(resp.headers['Location'], headers=auth_header)
	project = json.loads(resp.data.decode('utf-8'))
	project_id = project['_id']

	# Check some of the more complex/interesting fields.
	self.assertEqual(u'Prøject El Niño', project['name'])
	self.assertEqual(str(user_id), project['user'])
	self.assertEqual('p-%s' % project_id, project['url'])
	self.assertEqual(1, len(project['permissions']['groups']))

	# Check the etag
	resp = self.client.get('/projects/%s' % project_id, headers=auth_header)
	from_db = json.loads(resp.data)
	self.assertEqual(from_db['_etag'], project['_etag'])

	group_id = ObjectId(project['permissions']['groups'][0]['group'])

	# Check that there is a group for the project, and that the user is member of it.
	with self.app.test_request_context():
	groups = self.app.data.driver.db['groups']
	users = self.app.data.driver.db['users']

	group = groups.find_one(group_id)
	db_user = users.find_one(user_id)

	self.assertEqual(str(project_id), group['name'])
	self.assertIn(group_id, db_user['groups'])

	def test_project_creation_access_admin(self):
	"""Admin-created projects should be public"""

	proj = self._create_user_and_project(roles={u'admin'})
	self.assertEqual(['GET'], proj['permissions']['world'])

	def test_project_creation_access_subscriber(self):
	"""Subscriber-created projects should be private"""

	proj = self._create_user_and_project(roles={u'subscriber'})
	self.assertEqual([], proj['permissions']['world'])
	self.assertTrue(proj['is_private'])

	# Also check the database contents
	with self.app.test_request_context():
	project_id = ObjectId(proj['_id'])
	db_proj = self.app.data.driver.db['projects'].find_one(project_id)
	self.assertEqual([], db_proj['permissions']['world'])
	self.assertTrue(db_proj['is_private'])

	def test_project_list(self):
	"""Test that we get an empty list when querying for non-existing projects, instead of 403"""

	proj_a = self._create_user_and_project(user_id=24 * 'a',
	roles={u'subscriber'},
	project_name=u'Prøject A',
	token='token-a')
	proj_b = self._create_user_and_project(user_id=24 * 'b',
	roles={u'subscriber'},
	project_name=u'Prøject B',
	token='token-b')

	# Assertion: each user must have access to their own project.
	resp = self.client.get('/projects/%s' % proj_a['_id'],
	headers={'Authorization': self.make_header('token-a')})
	self.assertEqual(200, resp.status_code, resp.data)
	resp = self.client.get('/projects/%s' % proj_b['_id'],
	headers={'Authorization': self.make_header('token-b')})
	self.assertEqual(200, resp.status_code, resp.data)

	# Getting a project list should return projects you have access to.
	resp = self.client.get('/projects',
	headers={'Authorization': self.make_header('token-a')})
	self.assertEqual(200, resp.status_code)
	proj_list = json.loads(resp.data)
	self.assertEqual({u'Prøject A'}, {p['name'] for p in proj_list['_items']})

	resp = self.client.get('/projects',
	headers={'Authorization': self.make_header('token-b')})
	self.assertEqual(200, resp.status_code)
	proj_list = json.loads(resp.data)
	self.assertEqual({u'Prøject B'}, {p['name'] for p in proj_list['_items']})

	# No access to anything for user C, should result in empty list.
	self._create_user_with_token(roles={u'subscriber'}, token='token-c', user_id=12 * 'c')
	resp = self.client.get('/projects',
	headers={'Authorization': self.make_header('token-c')})
	self.assertEqual(200, resp.status_code)
	proj_list = json.loads(resp.data)
	self.assertEqual([], proj_list['_items'])


	class ProjectEditTest(AbstractProjectTest):
	def test_editing_as_subscriber(self):
	"""Test that we can set certain fields, but not all."""

	from application.utils import remove_private_keys, PillarJSONEncoder
	dumps = functools.partial(json.dumps, cls=PillarJSONEncoder)

	project_info = self._create_user_and_project([u'subscriber'])
	project_url = '/projects/%(_id)s' % project_info

	resp = self.client.get(project_url,
	headers={'Authorization': self.make_header('token')})
	project = json.loads(resp.data.decode('utf-8'))

	# Create another user we can try and assign the project to.
	other_user_id = 'f00dd00df00dd00df00dd00d'
	self._create_user_with_token(['subscriber'], 'other-token', user_id=other_user_id)

	# Unauthenticated should be forbidden
	resp = self.client.put('/projects/%s' % project['_id'],
	data=dumps(remove_private_keys(project)),
	headers={'Content-Type': 'application/json'})
	self.assertEqual(403, resp.status_code)

	# Regular user should be able to PUT, but only be able to edit certain fields.
	put_project = remove_private_keys(project)
	put_project['url'] = u'very-offensive-url'
	put_project['description'] = u'Blender je besplatan set alata za izradu interaktivnog 3D ' \
	u'sadržaja pod različitim operativnim sustavima.'
	put_project['name'] = u'โครงการปั่นเมฆ'
	put_project['summary'] = u'Это переведена на Google'
	put_project['status'] = 'pending'
	put_project['category'] = 'software'
	put_project['user'] = other_user_id

	# Try making the project public. This should update is_private as well.
	put_project['permissions']['world'] = ['GET']

	resp = self.client.put(project_url,
	data=dumps(put_project),
	headers={'Authorization': self.make_header('token'),
	'Content-Type': 'application/json',
	'If-Match': project['_etag']})
	self.assertEqual(200, resp.status_code, resp.data)

	# Re-fetch from database to see which fields actually made it there.
	# equal to put_project -> changed in DB
	# equal to project -> not changed in DB
	resp = self.client.get(project_url,
	headers={'Authorization': self.make_header('token')})
	db_proj = json.loads(resp.data)
	self.assertEqual(project['url'], db_proj['url'])
	self.assertEqual(put_project['description'], db_proj['description'])
	self.assertEqual(put_project['name'], db_proj['name'])
	self.assertEqual(put_project['summary'], db_proj['summary'])
	self.assertEqual(project['status'], db_proj['status'])
	self.assertEqual(project['category'], db_proj['category'])

	# Project should be consistent.
	self.assertEqual(False, db_proj['is_private'])
	self.assertEqual(['GET'], db_proj['permissions']['world'])

	def test_editing_as_admin(self):
	"""Test that we can set all fields as admin."""

	from application.utils import remove_private_keys, PillarJSONEncoder
	dumps = functools.partial(json.dumps, cls=PillarJSONEncoder)

	project_info = self._create_user_and_project([u'subscriber', u'admin'])
	project_url = '/projects/%(_id)s' % project_info

	resp = self.client.get(project_url)
	project = json.loads(resp.data.decode('utf-8'))

	# Create another user we can try and assign the project to.
	other_user_id = 'f00dd00df00dd00df00dd00d'
	self._create_user_with_token(['subscriber'], 'other-token', user_id=other_user_id)

	# Admin user should be able to PUT everything.
	put_project = remove_private_keys(project)
	put_project['url'] = u'very-offensive-url'
	put_project['description'] = u'Blender je besplatan set alata za izradu interaktivnog 3D ' \
	u'sadržaja pod različitim operativnim sustavima.'
	put_project['name'] = u'โครงการปั่นเมฆ'
	put_project['summary'] = u'Это переведена на Google'
	put_project['is_private'] = False
	put_project['status'] = 'pending'
	put_project['category'] = 'software'
	put_project['user'] = other_user_id

	resp = self.client.put(project_url,
	data=dumps(put_project),
	headers={'Authorization': self.make_header('token'),
	'Content-Type': 'application/json',
	'If-Match': project['_etag']})
	self.assertEqual(200, resp.status_code, resp.data)

	# Re-fetch from database to see which fields actually made it there.
	# equal to put_project -> changed in DB
	# equal to project -> not changed in DB
	resp = self.client.get('/projects/%s' % project['_id'])
	db_proj = json.loads(resp.data)
	self.assertEqual(put_project['url'], db_proj['url'])
	self.assertEqual(put_project['description'], db_proj['description'])
	self.assertEqual(put_project['name'], db_proj['name'])
	self.assertEqual(put_project['summary'], db_proj['summary'])
	self.assertEqual(put_project['is_private'], db_proj['is_private'])
	self.assertEqual(put_project['status'], db_proj['status'])
	self.assertEqual(put_project['category'], db_proj['category'])
	self.assertEqual(put_project['user'], db_proj['user'])

	def test_edits_by_nonowner_admin(self):
	"""Any admin should be able to edit any project."""

	from application.utils import remove_private_keys, PillarJSONEncoder
	dumps = functools.partial(json.dumps, cls=PillarJSONEncoder)

	# Create test project.
	project = self._create_user_and_project([u'subscriber'])
	project_id = project['_id']
	project_url = '/projects/%s' % project_id

	# Create test user.
	self._create_user_with_token(['admin'], 'admin-token', user_id='cafef00dbeef')

	# Admin user should be able to PUT.
	put_project = remove_private_keys(project)
	put_project['name'] = u'โครงการปั่นเมฆ'

	resp = self.client.put(project_url,
	data=dumps(put_project),
	headers={'Authorization': self.make_header('admin-token'),
	'Content-Type': 'application/json',
	'If-Match': project['_etag']})
	self.assertEqual(200, resp.status_code, resp.data)

	def test_edits_by_nonowner_subscriber(self):
	"""A subscriber should only be able to edit their own projects."""

	from application.utils import remove_private_keys, PillarJSONEncoder
	dumps = functools.partial(json.dumps, cls=PillarJSONEncoder)

	# Create test project.
	project = self._create_user_and_project([u'subscriber'])
	project_id = project['_id']
	project_url = '/projects/%s' % project_id

	# Create test user.
	my_user_id = 'cafef00dbeefcafef00dbeef'
	self._create_user_with_token(['subscriber'], 'mortal-token', user_id=my_user_id)

	# Regular subscriber should not be able to do this.
	put_project = remove_private_keys(project)
	put_project['name'] = u'Болту́н -- нахо́дка для шпио́на.'
	put_project['user'] = my_user_id
	resp = self.client.put(project_url,
	data=dumps(put_project),
	headers={'Authorization': self.make_header('mortal-token'),
	'Content-Type': 'application/json',
	'If-Match': project['_etag']})
	self.assertEqual(403, resp.status_code, resp.data)

	def test_delete_by_admin(self):
	# Create public test project.
	project_info = self._create_user_and_project([u'admin'])
	project_id = project_info['_id']
	project_url = '/projects/%s' % project_id

	# Create admin user that doesn't own the project, to check that
	# non-owner admins can delete projects too.
	self._create_user_with_token(['admin'], 'admin-token', user_id='cafef00dbeef')

	# Admin user should be able to DELETE.
	resp = self.client.delete(project_url,
	headers={'Authorization': self.make_header('admin-token'),
	'If-Match': project_info['_etag']})
	self.assertEqual(204, resp.status_code, resp.data)

	# Check that the project is gone.
	resp = self.client.get(project_url)
	self.assertEqual(404, resp.status_code, resp.data)

	# ... but we should still get it in the body.
	db_proj = json.loads(resp.data)
	self.assertEqual(u'Prøject El Niño', db_proj['name'])
	self.assertTrue(db_proj['_deleted'])

	# Querying for deleted projects should include it.
	# TODO: limit this to admin users only.
	# Also see http://python-eve.org/features.html#soft-delete
	projection = json.dumps({'name': 1, 'permissions': 1})
	where = json.dumps({'_deleted': True}) # MUST be True, 1 does not work.
	resp = self.client.get('/projects?where=%s&projection=%s' %
	(urllib.quote(where), urllib.quote(projection)))
	self.assertEqual(200, resp.status_code, resp.data)

	projlist = json.loads(resp.data)
	self.assertEqual(1, projlist['_meta']['total'])
	self.assertEqual(u'Prøject El Niño', projlist['_items'][0]['name'])

	def test_delete_by_subscriber(self):
	# Create test project.
	project_info = self._create_user_and_project([u'subscriber'])
	project_id = project_info['_id']
	project_url = '/projects/%s' % project_id

	# Create test user.
	self._create_user_with_token(['subscriber'], 'mortal-token', user_id='cafef00dbeef')

	# Other user should NOT be able to DELETE.
	resp = self.client.delete(project_url,
	headers={'Authorization': self.make_header('mortal-token'),
	'If-Match': project_info['_etag']})
	self.assertEqual(403, resp.status_code, resp.data)

	# Owner should be able to DELETE
	resp = self.client.delete(project_url,
	headers={'Authorization': self.make_header('token'),
	'If-Match': project_info['_etag']})
	self.assertEqual(204, resp.status_code, resp.data)


	class ProjectNodeAccess(AbstractProjectTest):
	def setUp(self, **kwargs):
	super(ProjectNodeAccess, self).setUp(**kwargs)

	from application.utils import PillarJSONEncoder

	# Project is created by regular subscriber, so should be private.
	self.user_id = self._create_user_with_token([u'subscriber'], 'token')
	resp = self._create_project(u'Prøject El Niño', 'token')
	self.assertEqual(201, resp.status_code)
	self.assertEqual('application/json', resp.mimetype)
	self.project = json.loads(resp.data)
	self.project_id = ObjectId(self.project['_id'])

	self.other_user_id = self._create_user_with_token([u'subscriber'], 'other-token',
	user_id='deadbeefdeadbeefcafef00d')

	self.test_node = {
	'description': '',
	'node_type': 'asset',
	'user': self.user_id,
	'properties': {
	'status': 'published',
	'content_type': 'image',
	},
	'name': 'Micak is a cool cat',
	'project': self.project_id,
	}

	# Add a node to the project
	resp = self.client.post('/nodes',
	headers={'Authorization': self.make_header('token'),
	'Content-Type': 'application/json'},
	data=json.dumps(self.test_node, cls=PillarJSONEncoder),
	)
	self.assertEqual(201, resp.status_code, (resp.status_code, resp.data))
	self.node_info = json.loads(resp.data)
	self.node_id = self.node_info['_id']
	self.node_url = '/nodes/%s' % self.node_id

	def test_node_access(self):
	"""Getting nodes should adhere to project access rules."""

	# Getting the node as the project owner should work.
	resp = self.client.get(self.node_url,
	headers={'Authorization': self.make_header('token')})
	self.assertEqual(200, resp.status_code, (resp.status_code, resp.data))

	# Getting the node as an outsider should not work.
	resp = self.client.get(self.node_url,
	headers={'Authorization': self.make_header('other-token')})
	self.assertEqual(403, resp.status_code, (resp.status_code, resp.data))

	def test_node_resource_access(self):
	# The owner of the project should get the node.
	resp = self.client.get('/nodes',
	headers={'Authorization': self.make_header('token')})
	self.assertEqual(200, resp.status_code, (resp.status_code, resp.data))
	listed_nodes = json.loads(resp.data)['_items']
	self.assertEquals(self.node_id, listed_nodes[0]['_id'])

	# Listing all nodes should not include nodes from private projects.
	resp = self.client.get('/nodes',
	headers={'Authorization': self.make_header('other-token')})
	self.assertEqual(403, resp.status_code, (resp.status_code, resp.data))

	def test_is_private_updated_by_world_permissions(self):
	"""For backward compatibility, is_private should reflect absence of world-GET"""

	from application.utils import remove_private_keys, dumps

	project_url = '/projects/%s' % self.project_id
	put_project = remove_private_keys(self.project)

	# Create admin user.
	self._create_user_with_token(['admin'], 'admin-token', user_id='cafef00dbeef')

	# Make the project public
	put_project['permissions']['world'] = ['GET'] # make public
	put_project['is_private'] = True # This should be overridden.

	resp = self.client.put(project_url,
	data=dumps(put_project),
	headers={'Authorization': self.make_header('admin-token'),
	'Content-Type': 'application/json',
	'If-Match': self.project['_etag']})
	self.assertEqual(200, resp.status_code, resp.data)

	with self.app.test_request_context():
	projects = self.app.data.driver.db['projects']
	db_proj = projects.find_one(self.project_id)
	self.assertEqual(['GET'], db_proj['permissions']['world'])
	self.assertFalse(db_proj['is_private'])

	# Make the project private
	put_project['permissions']['world'] = []

	resp = self.client.put(project_url,
	data=dumps(put_project),
	headers={'Authorization': self.make_header('admin-token'),
	'Content-Type': 'application/json',
	'If-Match': db_proj['_etag']})
	self.assertEqual(200, resp.status_code, resp.data)

	with self.app.test_request_context():
	projects = self.app.data.driver.db['projects']
	db_proj = projects.find_one(self.project_id)
	self.assertEqual([], db_proj['permissions']['world'])
	self.assertTrue(db_proj['is_private'])

	def test_add_remove_user(self):
	from application.modules import projects
	from application.utils import dumps

	project_mng_user_url = '/p/users'

	# Use our API to add user to group
	payload = {
	'project_id': self.project_id,
	'user_id': self.other_user_id,
	'action': 'add'}

	resp = self.client.post(project_mng_user_url,
	data=dumps(payload),
	content_type='application/json',
	headers={
	'Authorization': self.make_header('token'),
	'If-Match': self.project['_etag']})
	self.assertEqual(200, resp.status_code, resp.data)

	# Check if the user is now actually member of the group.
	with self.app.test_request_context():
	users = self.app.data.driver.db['users']

	db_user = users.find_one(self.other_user_id)
	admin_group = projects.get_admin_group(self.project)

	self.assertIn(admin_group['_id'], db_user['groups'])

	# Update payload to remove the user we just added
	payload['action'] = 'remove'

	resp = self.client.post(project_mng_user_url,
	data=dumps(payload),
	content_type='application/json',
	headers={
	'Authorization': self.make_header('token'),
	'If-Match': self.project['_etag']})
	self.assertEqual(200, resp.status_code, resp.data)

	# Check if the user is now actually removed from the group.
	with self.app.test_request_context():
	users = self.app.data.driver.db['users']

	db_user = users.find_one(self.other_user_id)
	self.assertNotIn(admin_group['_id'], db_user['groups'])

	def test_remove_self(self):
	"""Every user should be able to remove themselves from a project,
	regardless of permissions.
	"""

	from application.modules import projects
	from application.utils import dumps

	project_mng_user_url = '/p/users'

	# Use our API to add user to group
	payload = {
	'project_id': self.project_id,
	'user_id': self.other_user_id,
	'action': 'add'}

	resp = self.client.post(project_mng_user_url,
	data=dumps(payload),
	content_type='application/json',
	headers={'Authorization': self.make_header('token')})
	self.assertEqual(200, resp.status_code, resp.data)

	# Update payload to remove the user we just added, and call it as that user.
	payload['action'] = 'remove'

	resp = self.client.post(project_mng_user_url,
	data=dumps(payload),
	content_type='application/json',
	headers={'Authorization': self.make_header('other-token')})
	self.assertEqual(200, resp.status_code, resp.data)

	# Check if the user is now actually removed from the group.
	with self.app.test_request_context():
	users = self.app.data.driver.db['users']

	db_user = users.find_one(self.other_user_id)
	admin_group = projects.get_admin_group(self.project)
	self.assertNotIn(admin_group['_id'], db_user['groups'])

File Metadata

Mime Type: text/x-python
Expires: Thu, May 19, 12:35 AM (2 d)
Storage Engine: local-disk
Storage Format: Raw Data
Storage Handle: b3/59/b289a5b2bbd7ee37172c9ba006f4

test_project_management.pyNo OneTemporaryActions

test_project_management.pyView Options

File Metadata

Event Timeline

test_project_management.py
No OneTemporary
Actions

test_project_management.py
View Options