243 lines
8.2 KiB
Python
243 lines
8.2 KiB
Python
# Copyright 2013 IBM Corp.
|
|
|
|
import unittest
|
|
import mox
|
|
from sqlalchemy.orm.session import Session as session
|
|
from powervc.neutron.db.powervc_db_v2 import PowerVCAgentDB
|
|
from powervc.neutron.db import powervc_models_v2 as model
|
|
from sqlalchemy.engine.base import Transaction as transaction
|
|
from test.fake_powervc_network import FakePowerVCNetwork
|
|
from test.fake_os_network import FakeOSNetwork
|
|
from powervc.neutron.common import utils
|
|
from powervc.neutron.db.powervc_models_v2 import PowerVCMapping
|
|
from sqlalchemy.orm import Query as query
|
|
|
|
|
|
"""
|
|
The class TestPowerVCNeutronDB is used to implement
|
|
the UT test of the methods in the class PowerVCAgentDB
|
|
"""
|
|
|
|
|
|
class TestPowerVCNeutronDB(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
"""
|
|
This method is used to initialize the UT environment
|
|
"""
|
|
|
|
# Initialize the FakePowerVCNetwork instance
|
|
self.fakePowerVCNetwork = FakePowerVCNetwork()
|
|
|
|
# Initialize the FakeOSNetwork instance
|
|
self.fakeOSNetwork = FakeOSNetwork()
|
|
|
|
# Initialize the PowerVCMapping instance
|
|
async_key = utils.gen_network_sync_key(
|
|
self.fakePowerVCNetwork.powerNetInstance)
|
|
self.powerVCMapping = PowerVCMapping(
|
|
obj_type="Network", sync_key=async_key)
|
|
self.powerVCMapping.local_id = self.fakeOSNetwork.\
|
|
fakeOSNetworkInstance['id']
|
|
self.powerVCMapping.pvc_id = self.fakePowerVCNetwork.\
|
|
powerNetInstance['id']
|
|
self.powerVCMapping.status = "Active"
|
|
self.powerVCMapping.id = None
|
|
# Initialize the PowerVCAgentDB instance
|
|
|
|
def __init__(self, session):
|
|
self.session = session
|
|
|
|
PowerVCAgentDB.__init__ = __init__
|
|
self.powervcagentdb = PowerVCAgentDB(session)
|
|
|
|
# Initialize the MOX instance
|
|
self.aMox = mox.Mox()
|
|
|
|
def tearDown(self):
|
|
pass
|
|
|
|
def test_create_object(self):
|
|
"""
|
|
Test the method def _create_object(self, obj_type, sync_key,
|
|
local_id=None, pvc_id=None)
|
|
"""
|
|
|
|
obj_type = "Network"
|
|
sync_key = utils.gen_network_sync_key(
|
|
self.fakePowerVCNetwork.powerNetInstance)
|
|
local_id = self.fakeOSNetwork.fakeOSNetworkInstance['id']
|
|
pvc_id = self.fakePowerVCNetwork.powerNetInstance['id']
|
|
|
|
inputPowerVCMObj = model.PowerVCMapping(obj_type, sync_key)
|
|
|
|
self.aMox.StubOutWithMock(session, 'begin')
|
|
session.begin(subtransactions=True).AndReturn(transaction(None, None))
|
|
|
|
self.aMox.StubOutWithMock(model, 'PowerVCMapping')
|
|
model.PowerVCMapping(obj_type, sync_key).AndReturn(inputPowerVCMObj)
|
|
|
|
self.aMox.StubOutWithMock(session, 'add')
|
|
session.add(inputPowerVCMObj).AndReturn("")
|
|
|
|
self.aMox.ReplayAll()
|
|
|
|
self.powervcagentdb._create_object(
|
|
obj_type, sync_key, update_data=None,
|
|
local_id=local_id, pvc_id=pvc_id)
|
|
|
|
self.aMox.VerifyAll()
|
|
|
|
self.assertEqual(
|
|
self.powerVCMapping.local_id, inputPowerVCMObj.local_id)
|
|
self.assertEqual(self.powerVCMapping.pvc_id, inputPowerVCMObj.pvc_id)
|
|
self.assertEqual(self.powerVCMapping.status, inputPowerVCMObj.status)
|
|
self.aMox.UnsetStubs()
|
|
|
|
def test_delete_existing_object(self):
|
|
"""
|
|
Test the method _delete_object(self, obj) when the object exists
|
|
Test scenario:
|
|
When the data is in the database, the delete operation should
|
|
complete successfully
|
|
"""
|
|
|
|
self.aMox.StubOutWithMock(session, 'query')
|
|
session.query(model.PowerVCMapping).AndReturn(query)
|
|
|
|
self.aMox.StubOutWithMock(query, 'filter_by')
|
|
query.filter_by(id=self.powerVCMapping['id']).AndReturn(query)
|
|
|
|
self.aMox.StubOutWithMock(query, 'one')
|
|
query.one().AndReturn(self.powerVCMapping)
|
|
|
|
self.aMox.StubOutWithMock(session, 'begin')
|
|
session.begin(subtransactions=True).AndReturn(transaction(None, None))
|
|
|
|
self.aMox.StubOutWithMock(session, 'delete')
|
|
returnValue = session.delete(self.powerVCMapping).AndReturn(True)
|
|
|
|
self.aMox.ReplayAll()
|
|
|
|
self.powervcagentdb._delete_object(self.powerVCMapping)
|
|
|
|
self.aMox.VerifyAll()
|
|
|
|
self.assertEqual(returnValue, True)
|
|
|
|
self.aMox.UnsetStubs()
|
|
|
|
def test_get_objects_with_status(self):
|
|
"""Test the method def _get_objects(self, obj_type, status)
|
|
Test scenario:
|
|
Get the object when the status is not None
|
|
"""
|
|
|
|
self.aMox.StubOutWithMock(session, 'query')
|
|
session.query(model.PowerVCMapping).AndReturn(query)
|
|
|
|
self.aMox.StubOutWithMock(query, 'filter_by')
|
|
query.filter_by(obj_type=self.powerVCMapping.obj_type,
|
|
status=self.powerVCMapping.status).AndReturn(query)
|
|
|
|
self.aMox.StubOutWithMock(query, 'all')
|
|
query.all().AndReturn(self.powerVCMapping)
|
|
|
|
self.aMox.ReplayAll()
|
|
returnValue = self.powervcagentdb._get_objects(
|
|
obj_type=self.powerVCMapping.obj_type,
|
|
status=self.powerVCMapping.status)
|
|
self.aMox.VerifyAll()
|
|
self.assertEqual(returnValue, self.powerVCMapping)
|
|
|
|
self.aMox.UnsetStubs()
|
|
|
|
def test_get_object(self):
|
|
"""
|
|
Test the method _get_object() using a sync key
|
|
Test scenario:
|
|
Get the object with sync_key
|
|
"""
|
|
|
|
obj_type = self.powerVCMapping.obj_type
|
|
sync_key = self.powerVCMapping.sync_key
|
|
|
|
self.aMox.StubOutWithMock(session, 'query')
|
|
session.query(model.PowerVCMapping).AndReturn(query)
|
|
|
|
self.aMox.StubOutWithMock(query, 'filter_by')
|
|
query.filter_by(
|
|
obj_type=obj_type, sync_key=sync_key).AndReturn(query)
|
|
|
|
self.aMox.StubOutWithMock(query, 'one')
|
|
query.one().AndReturn(self.powerVCMapping)
|
|
|
|
self.aMox.ReplayAll()
|
|
returnValue = self.powervcagentdb._get_object(
|
|
obj_type=obj_type, sync_key=sync_key)
|
|
self.aMox.VerifyAll()
|
|
self.assertEqual(returnValue, self.powerVCMapping)
|
|
self.aMox.UnsetStubs()
|
|
|
|
def test_set_object_pvc_id(self):
|
|
"""
|
|
Test the method _set_object_pvc_id(self, obj, pvc_id)
|
|
Test scenario:
|
|
Set the pvc_id of the specified object when local_id is none
|
|
"""
|
|
|
|
obj_id = self.powerVCMapping.id
|
|
self.powerVCMapping.pvc_id = None
|
|
self.powerVCMapping.local_id = None
|
|
self.powerVCMapping.status = None
|
|
|
|
self.aMox.StubOutWithMock(session, 'query')
|
|
session.query(model.PowerVCMapping).AndReturn(query)
|
|
|
|
self.aMox.StubOutWithMock(query, 'filter_by')
|
|
query.filter_by(id=obj_id).AndReturn(query)
|
|
|
|
self.aMox.StubOutWithMock(query, 'one')
|
|
query.one().AndReturn(self.powerVCMapping)
|
|
|
|
self.aMox.StubOutWithMock(session, 'merge')
|
|
session.merge(self.powerVCMapping).AndReturn("")
|
|
|
|
self.aMox.ReplayAll()
|
|
self.powervcagentdb._set_object_pvc_id(self.powerVCMapping, 'test')
|
|
self.aMox.VerifyAll()
|
|
self.assertEqual(self.powerVCMapping.status, 'Creating')
|
|
self.assertEqual(self.powerVCMapping.pvc_id, 'test')
|
|
self.aMox.UnsetStubs()
|
|
|
|
def test_set_object_local_id(self):
|
|
"""
|
|
Test the method _set_object_local_id(self, obj, local_id)
|
|
Test scenario:
|
|
Set the local_id of the specified object when the pvc_id is none
|
|
"""
|
|
|
|
obj_id = self.powerVCMapping.id
|
|
self.powerVCMapping.pvc_id = None
|
|
self.powerVCMapping.local_id = None
|
|
self.powerVCMapping.status = None
|
|
|
|
self.aMox.StubOutWithMock(session, 'query')
|
|
session.query(model.PowerVCMapping).AndReturn(query)
|
|
|
|
self.aMox.StubOutWithMock(query, 'filter_by')
|
|
query.filter_by(id=obj_id).AndReturn(query)
|
|
|
|
self.aMox.StubOutWithMock(query, 'one')
|
|
query.one().AndReturn(self.powerVCMapping)
|
|
|
|
self.aMox.StubOutWithMock(session, 'merge')
|
|
session.merge(self.powerVCMapping).AndReturn("")
|
|
|
|
self.aMox.ReplayAll()
|
|
self.powervcagentdb._set_object_local_id(self.powerVCMapping, 'test')
|
|
self.aMox.VerifyAll()
|
|
self.assertEqual(self.powerVCMapping.status, 'Creating')
|
|
self.assertEqual(self.powerVCMapping.local_id, 'test')
|
|
self.aMox.UnsetStubs()
|