x/python-ganttclient
Code Issues Proposed changes

Refactor sqlalchemy api to perform contextual authorization.

Browse Source 
All database calls now examine the context object for information about what kind of user is accessing the data.  If an administrator is accessing, full privileges are granted.  If a normal user is accessing, then checks are made to ensure that the user does indeed have the rights to the data.

Also refactored NovaBase and removed several methods since they would have to be changed when we move away from sqlalchemy models and begin using sqlalchemy table definitions.
This commit is contained in:
Devin Carlen
2010-10-01 15:08:19 +00:00
committed by Tarmac
parent 452728aa13 d83dbc4b00
commit 27473e075b
2 changed files with 33 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
nova/tests/compute_unittest.py
View File

@@ -30,7 +30,7 @@ from nova import flags
from nova import test
from nova import utils
from nova.auth import manager
from nova.api import context
FLAGS = flags.FLAGS
@@ -96,7 +96,9 @@ class ComputeTestCase(test.TrialTestCase):
self.assertEqual(instance_ref['deleted_at'], None)
terminate = datetime.datetime.utcnow()
yield self.compute.terminate_instance(self.context, instance_id)
instance_ref = db.instance_get({'deleted': True}, instance_id)
self.context = context.get_admin_context(user=self.user,
read_deleted=True)
instance_ref = db.instance_get(self.context, instance_id)
self.assert_(instance_ref['launched_at'] < terminate)
self.assert_(instance_ref['deleted_at'] > terminate)

50
nova/tests/network_unittest.py
View File

@@ -56,12 +56,12 @@ class NetworkTestCase(test.TrialTestCase):
'netuser',
name))
# create the necessary network data for the project
self.network.set_network_host(self.context, self.projects[i].id)
instance_ref = db.instance_create(None,
{'mac_address': utils.generate_mac()})
user_context = context.APIRequestContext(project=self.projects[i],
user=self.user)
self.network.set_network_host(user_context, self.projects[i].id)
instance_ref = self._create_instance(0)
self.instance_id = instance_ref['id']
instance_ref = db.instance_create(None,
{'mac_address': utils.generate_mac()})
instance_ref = self._create_instance(1)
self.instance2_id = instance_ref['id']
def tearDown(self): # pylint: disable-msg=C0103
@@ -74,6 +74,15 @@ class NetworkTestCase(test.TrialTestCase):
self.manager.delete_project(project)
self.manager.delete_user(self.user)
def _create_instance(self, project_num, mac=None):
if not mac:
mac = utils.generate_mac()
project = self.projects[project_num]
self.context.project = project
return db.instance_create(self.context,
{'project_id': project.id,
'mac_address': mac})
def _create_address(self, project_num, instance_id=None):
"""Create an address in given project num"""
if instance_id is None:
@@ -81,9 +90,15 @@ class NetworkTestCase(test.TrialTestCase):
self.context.project = self.projects[project_num]
return self.network.allocate_fixed_ip(self.context, instance_id)
def _deallocate_address(self, project_num, address):
self.context.project = self.projects[project_num]
self.network.deallocate_fixed_ip(self.context, address)
def test_public_network_association(self):
"""Makes sure that we can allocaate a public ip"""
# TODO(vish): better way of adding floating ips
self.context.project = self.projects[0]
pubnet = IPy.IP(flags.FLAGS.public_range)
address = str(pubnet[0])
try:
@@ -109,7 +124,7 @@ class NetworkTestCase(test.TrialTestCase):
address = self._create_address(0)
self.assertTrue(is_allocated_in_project(address, self.projects[0].id))
lease_ip(address)
self.network.deallocate_fixed_ip(self.context, address)
self._deallocate_address(0, address)
# Doesn't go away until it's dhcp released
self.assertTrue(is_allocated_in_project(address, self.projects[0].id))
@@ -130,14 +145,14 @@ class NetworkTestCase(test.TrialTestCase):
lease_ip(address)
lease_ip(address2)
self.network.deallocate_fixed_ip(self.context, address)
self._deallocate_address(0, address)
release_ip(address)
self.assertFalse(is_allocated_in_project(address, self.projects[0].id))
# First address release shouldn't affect the second
self.assertTrue(is_allocated_in_project(address2, self.projects[1].id))
self.network.deallocate_fixed_ip(self.context, address2)
self._deallocate_address(1, address2)
release_ip(address2)
self.assertFalse(is_allocated_in_project(address2,
self.projects[1].id))
@@ -148,24 +163,19 @@ class NetworkTestCase(test.TrialTestCase):
lease_ip(first)
instance_ids = []
for i in range(1, 5):
mac = utils.generate_mac()
instance_ref = db.instance_create(None,
{'mac_address': mac})
instance_ref = self._create_instance(i, mac=utils.generate_mac())
instance_ids.append(instance_ref['id'])
address = self._create_address(i, instance_ref['id'])
mac = utils.generate_mac()
instance_ref = db.instance_create(None,
{'mac_address': mac})
instance_ref = self._create_instance(i, mac=utils.generate_mac())
instance_ids.append(instance_ref['id'])
address2 = self._create_address(i, instance_ref['id'])
mac = utils.generate_mac()
instance_ref = db.instance_create(None,
{'mac_address': mac})
instance_ref = self._create_instance(i, mac=utils.generate_mac())
instance_ids.append(instance_ref['id'])
address3 = self._create_address(i, instance_ref['id'])
lease_ip(address)
lease_ip(address2)
lease_ip(address3)
self.context.project = self.projects[i]
self.assertFalse(is_allocated_in_project(address,
self.projects[0].id))
self.assertFalse(is_allocated_in_project(address2,
@@ -181,7 +191,7 @@ class NetworkTestCase(test.TrialTestCase):
for instance_id in instance_ids:
db.instance_destroy(None, instance_id)
release_ip(first)
self.network.deallocate_fixed_ip(self.context, first)
self._deallocate_address(0, first)
def test_vpn_ip_and_port_looks_valid(self):
"""Ensure the vpn ip and port are reasonable"""
@@ -242,9 +252,7 @@ class NetworkTestCase(test.TrialTestCase):
addresses = []
instance_ids = []
for i in range(num_available_ips):
mac = utils.generate_mac()
instance_ref = db.instance_create(None,
{'mac_address': mac})
instance_ref = self._create_instance(0)
instance_ids.append(instance_ref['id'])
address = self._create_address(0, instance_ref['id'])
addresses.append(address)