Complete move of baremetal machine tests

Migrating tests to wrap up machine actions.

Ports will be migrated separately as there
will also need to be portgroups, which could
be wrapped into the same file/class depending
on how that ends up being implemented.

Change-Id: I239bbc134f42a47e8df75531dbd50916a97a6068
This commit is contained in:
Julia Kreger 2017-09-10 16:23:00 +00:00
parent b5e905f9b8
commit 395d927081
4 changed files with 564 additions and 237 deletions

View File

@ -207,7 +207,7 @@ class OperatorCloud(openstackcloud.OpenStackCloud):
nic = self.manager.submit_task(
_tasks.MachinePortCreate(address=row['mac'],
node_uuid=machine['uuid']))
created_nics.append(nic.uuid)
created_nics.append(nic['uuid'])
except Exception as e:
self.log.debug("ironic NIC registration failed", exc_info=True)
@ -273,7 +273,10 @@ class OperatorCloud(openstackcloud.OpenStackCloud):
machine = self.get_machine(machine['uuid'])
if (machine['reservation'] is None and
machine['provision_state'] is not 'enroll'):
# NOTE(TheJulia): In this case, the node has
# has moved on from the previous state and is
# likely not being verified, as no lock is
# present on the node.
self.node_set_provision_state(
machine['uuid'], 'provide')
machine = self.get_machine(machine['uuid'])
@ -288,8 +291,10 @@ class OperatorCloud(openstackcloud.OpenStackCloud):
raise OpenStackCloudException(
"Machine encountered a failure: %s"
% machine['last_error'])
return machine
if not isinstance(machine, str):
return self._normalize_machine(machine)
else:
return machine
def unregister_machine(self, nics, uuid, wait=False, timeout=600):
"""Unregister Baremetal from Ironic

View File

@ -248,6 +248,16 @@ def make_fake_machine(machine_name, machine_id=None):
id=machine_id,
name=machine_name))
def make_fake_port(address, node_id=None, port_id=None):
if not node_id:
node_id = uuid.uuid4().hex
if not port_id:
port_id = uuid.uuid4().hex
return meta.obj_to_munch(FakeMachinePort(
id=port_id,
address=address,
node_id=node_id))
class FakeFloatingIP(object):
def __init__(self, id, pool, ip, fixed_ip, instance_id):
@ -346,7 +356,7 @@ class FakeVolumeSnapshot(object):
class FakeMachine(object):
def __init__(self, id, name=None, driver=None, driver_info=None,
chassis_uuid=None, instance_info=None, instance_uuid=None,
properties=None):
properties=None, reservation=None, last_error=None):
self.uuid = id
self.name = name
self.driver = driver
@ -355,6 +365,8 @@ class FakeMachine(object):
self.instance_info = instance_info
self.instance_uuid = instance_uuid
self.properties = properties
self.reservation = reservation
self.last_error = last_error
class FakeMachinePort(object):

View File

@ -32,6 +32,12 @@ class TestBaremetalNode(base.IronicTestCase):
super(TestBaremetalNode, self).setUp()
self.fake_baremetal_node = fakes.make_fake_machine(
self.name, self.uuid)
# TODO(TheJulia): Some tests below have fake ports,
# since they are required in some processes. Lets refactor
# them at some point to use self.fake_baremetal_port.
self.fake_baremetal_port = fakes.make_fake_port(
'00:01:02:03:04:05',
node_id=self.uuid)
def test_list_machines(self):
fake_baremetal_two = fakes.make_fake_machine('two', str(uuid.uuid4()))
@ -820,6 +826,542 @@ class TestBaremetalNode(base.IronicTestCase):
self.assertIsNone(return_value)
self.assert_calls()
def test_register_machine(self):
mac_address = '00:01:02:03:04:05'
nics = [{'mac': mac_address}]
node_uuid = self.fake_baremetal_node['uuid']
# TODO(TheJulia): There is a lot of duplication
# in testing creation. Surely this hsould be a helper
# or something. We should fix this, after we have
# ironicclient removed, as in the mean time visibility
# will be helpful.
node_to_post = {
'chassis_uuid': None,
'driver': None,
'driver_info': None,
'name': self.fake_baremetal_node['name'],
'properties': None,
'uuid': node_uuid}
self.fake_baremetal_node['provision_state'] = 'available'
if 'provision_state' in node_to_post:
node_to_post.pop('provision_state')
self.register_uris([
dict(
method='POST',
uri=self.get_mock_url(
resource='nodes'),
json=self.fake_baremetal_node,
validate=dict(json=node_to_post)),
dict(
method='POST',
uri=self.get_mock_url(
resource='ports'),
validate=dict(json={'address': mac_address,
'node_uuid': node_uuid}),
json=self.fake_baremetal_port),
])
return_value = self.op_cloud.register_machine(nics, **node_to_post)
self.assertDictEqual(self.fake_baremetal_node, return_value)
self.assert_calls()
# TODO(TheJulia): After we remove ironicclient,
# we need to de-duplicate these tests. Possibly
# a dedicated class, although we should do it then
# as we may find differences that need to be accounted
# for newer microversions.
def test_register_machine_enroll(self):
mac_address = '00:01:02:03:04:05'
nics = [{'mac': mac_address}]
node_uuid = self.fake_baremetal_node['uuid']
node_to_post = {
'chassis_uuid': None,
'driver': None,
'driver_info': None,
'name': self.fake_baremetal_node['name'],
'properties': None,
'uuid': node_uuid}
self.fake_baremetal_node['provision_state'] = 'enroll'
manageable_node = self.fake_baremetal_node.copy()
manageable_node['provision_state'] = 'manageable'
available_node = self.fake_baremetal_node.copy()
available_node['provision_state'] = 'available'
self.register_uris([
dict(
method='POST',
uri=self.get_mock_url(
resource='nodes'),
validate=dict(json=node_to_post),
json=self.fake_baremetal_node),
dict(
method='POST',
uri=self.get_mock_url(
resource='ports'),
validate=dict(json={'address': mac_address,
'node_uuid': node_uuid}),
json=self.fake_baremetal_port),
dict(
method='PUT',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid'],
'states', 'provision']),
validate=dict(json={'target': 'manage'})),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=manageable_node),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=manageable_node),
dict(
method='PUT',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid'],
'states', 'provision']),
validate=dict(json={'target': 'provide'})),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=available_node),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=available_node),
])
# NOTE(When we migrate to a newer microversion, this test
# may require revision. It was written for microversion
# ?1.13?, which accidently got reverted to 1.6 at one
# point during code being refactored soon after the
# change landed. Presently, with the lock at 1.6,
# this code is never used in the current code path.
return_value = self.op_cloud.register_machine(nics, **node_to_post)
self.assertDictEqual(available_node, return_value)
self.assert_calls()
def test_register_machine_enroll_wait(self):
mac_address = self.fake_baremetal_port
nics = [{'mac': mac_address}]
node_uuid = self.fake_baremetal_node['uuid']
node_to_post = {
'chassis_uuid': None,
'driver': None,
'driver_info': None,
'name': self.fake_baremetal_node['name'],
'properties': None,
'uuid': node_uuid}
self.fake_baremetal_node['provision_state'] = 'enroll'
manageable_node = self.fake_baremetal_node.copy()
manageable_node['provision_state'] = 'manageable'
available_node = self.fake_baremetal_node.copy()
available_node['provision_state'] = 'available'
self.register_uris([
dict(
method='POST',
uri=self.get_mock_url(
resource='nodes'),
validate=dict(json=node_to_post),
json=self.fake_baremetal_node),
dict(
method='POST',
uri=self.get_mock_url(
resource='ports'),
validate=dict(json={'address': mac_address,
'node_uuid': node_uuid}),
json=self.fake_baremetal_port),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=self.fake_baremetal_node),
dict(
method='PUT',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid'],
'states', 'provision']),
validate=dict(json={'target': 'manage'})),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=self.fake_baremetal_node),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=manageable_node),
dict(
method='PUT',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid'],
'states', 'provision']),
validate=dict(json={'target': 'provide'})),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=available_node),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=available_node),
])
return_value = self.op_cloud.register_machine(nics, wait=True,
**node_to_post)
self.assertDictEqual(available_node, return_value)
self.assert_calls()
def test_register_machine_enroll_failure(self):
mac_address = '00:01:02:03:04:05'
nics = [{'mac': mac_address}]
node_uuid = self.fake_baremetal_node['uuid']
node_to_post = {
'chassis_uuid': None,
'driver': None,
'driver_info': None,
'name': self.fake_baremetal_node['name'],
'properties': None,
'uuid': node_uuid}
self.fake_baremetal_node['provision_state'] = 'enroll'
failed_node = self.fake_baremetal_node.copy()
failed_node['reservation'] = 'conductor0'
failed_node['provision_state'] = 'verifying'
failed_node['last_error'] = 'kaboom!'
self.register_uris([
dict(
method='POST',
uri=self.get_mock_url(
resource='nodes'),
json=self.fake_baremetal_node,
validate=dict(json=node_to_post)),
dict(
method='POST',
uri=self.get_mock_url(
resource='ports'),
validate=dict(json={'address': mac_address,
'node_uuid': node_uuid}),
json=self.fake_baremetal_port),
dict(
method='PUT',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid'],
'states', 'provision']),
validate=dict(json={'target': 'manage'})),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=failed_node),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=failed_node),
])
self.assertRaises(
exc.OpenStackCloudException,
self.op_cloud.register_machine,
nics,
**node_to_post)
self.assert_calls()
def test_register_machine_enroll_timeout(self):
mac_address = '00:01:02:03:04:05'
nics = [{'mac': mac_address}]
node_uuid = self.fake_baremetal_node['uuid']
node_to_post = {
'chassis_uuid': None,
'driver': None,
'driver_info': None,
'name': self.fake_baremetal_node['name'],
'properties': None,
'uuid': node_uuid}
self.fake_baremetal_node['provision_state'] = 'enroll'
busy_node = self.fake_baremetal_node.copy()
busy_node['reservation'] = 'conductor0'
busy_node['provision_state'] = 'verifying'
self.register_uris([
dict(
method='POST',
uri=self.get_mock_url(
resource='nodes'),
json=self.fake_baremetal_node,
validate=dict(json=node_to_post)),
dict(
method='POST',
uri=self.get_mock_url(
resource='ports'),
validate=dict(json={'address': mac_address,
'node_uuid': node_uuid}),
json=self.fake_baremetal_port),
dict(
method='PUT',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid'],
'states', 'provision']),
validate=dict(json={'target': 'manage'})),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=self.fake_baremetal_node),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=busy_node),
])
# NOTE(TheJulia): This test shortcircuits the timeout loop
# such that it executes only once. The very last returned
# state to the API is essentially a busy state that we
# want to block on until it has cleared.
self.assertRaises(
exc.OpenStackCloudException,
self.op_cloud.register_machine,
nics,
timeout=0.001,
lock_timeout=0.001,
**node_to_post)
self.assert_calls()
def test_register_machine_enroll_timeout_wait(self):
mac_address = '00:01:02:03:04:05'
nics = [{'mac': mac_address}]
node_uuid = self.fake_baremetal_node['uuid']
node_to_post = {
'chassis_uuid': None,
'driver': None,
'driver_info': None,
'name': self.fake_baremetal_node['name'],
'properties': None,
'uuid': node_uuid}
self.fake_baremetal_node['provision_state'] = 'enroll'
self.register_uris([
dict(
method='POST',
uri=self.get_mock_url(
resource='nodes'),
json=self.fake_baremetal_node,
validate=dict(json=node_to_post)),
dict(
method='POST',
uri=self.get_mock_url(
resource='ports'),
validate=dict(json={'address': mac_address,
'node_uuid': node_uuid}),
json=self.fake_baremetal_port),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=self.fake_baremetal_node),
dict(
method='PUT',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid'],
'states', 'provision']),
validate=dict(json={'target': 'manage'})),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=self.fake_baremetal_node),
])
self.assertRaises(
exc.OpenStackCloudException,
self.op_cloud.register_machine,
nics,
wait=True,
timeout=0.001,
**node_to_post)
self.assert_calls()
def test_register_machine_port_create_failed(self):
mac_address = '00:01:02:03:04:05'
nics = [{'mac': mac_address}]
node_uuid = self.fake_baremetal_node['uuid']
node_to_post = {
'chassis_uuid': None,
'driver': None,
'driver_info': None,
'name': self.fake_baremetal_node['name'],
'properties': None,
'uuid': node_uuid}
self.fake_baremetal_node['provision_state'] = 'available'
self.register_uris([
dict(
method='POST',
uri=self.get_mock_url(
resource='nodes'),
json=self.fake_baremetal_node,
validate=dict(json=node_to_post)),
dict(
method='POST',
uri=self.get_mock_url(
resource='ports'),
status_code=400,
json={'error': 'invalid'},
validate=dict(json={'address': mac_address,
'node_uuid': node_uuid})),
dict(
method='DELETE',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']])),
])
self.assertRaises(exc.OpenStackCloudException,
self.op_cloud.register_machine,
nics, **node_to_post)
self.assert_calls()
def test_unregister_machine(self):
mac_address = self.fake_baremetal_port['address']
nics = [{'mac': mac_address}]
port_uuid = self.fake_baremetal_port['uuid']
# NOTE(TheJulia): The two values below should be the same.
port_node_uuid = self.fake_baremetal_port['node_uuid']
port_url_address = 'detail?address=%s' % mac_address
self.fake_baremetal_node['provision_state'] = 'available'
self.register_uris([
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=self.fake_baremetal_node),
dict(
method='GET',
uri=self.get_mock_url(
resource='ports',
append=[port_url_address]),
json={'ports': [{'address': mac_address,
'node_uuid': port_node_uuid,
'uuid': port_uuid}]}),
dict(
method='DELETE',
uri=self.get_mock_url(
resource='ports',
append=[self.fake_baremetal_port['uuid']])),
dict(
method='DELETE',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']])),
])
self.op_cloud.unregister_machine(nics,
self.fake_baremetal_node['uuid'])
self.assert_calls()
def test_unregister_machine_timeout(self):
mac_address = self.fake_baremetal_port['address']
nics = [{'mac': mac_address}]
port_uuid = self.fake_baremetal_port['uuid']
port_node_uuid = self.fake_baremetal_port['node_uuid']
port_url_address = 'detail?address=%s' % mac_address
self.fake_baremetal_node['provision_state'] = 'available'
self.register_uris([
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=self.fake_baremetal_node),
dict(
method='GET',
uri=self.get_mock_url(
resource='ports',
append=[port_url_address]),
json={'ports': [{'address': mac_address,
'node_uuid': port_node_uuid,
'uuid': port_uuid}]}),
dict(
method='DELETE',
uri=self.get_mock_url(
resource='ports',
append=[self.fake_baremetal_port['uuid']])),
dict(
method='DELETE',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']])),
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=self.fake_baremetal_node),
])
self.assertRaises(
exc.OpenStackCloudException,
self.op_cloud.unregister_machine,
nics,
self.fake_baremetal_node['uuid'],
wait=True,
timeout=0.001)
self.assert_calls()
def test_unregister_machine_unavailable(self):
# This is a list of invalid states that the method
# should fail on.
invalid_states = ['active', 'cleaning', 'clean wait', 'clean failed']
mac_address = self.fake_baremetal_port['address']
nics = [{'mac': mac_address}]
url_list = []
for state in invalid_states:
self.fake_baremetal_node['provision_state'] = state
url_list.append(
dict(
method='GET',
uri=self.get_mock_url(
resource='nodes',
append=[self.fake_baremetal_node['uuid']]),
json=self.fake_baremetal_node))
self.register_uris(url_list)
for state in invalid_states:
self.assertRaises(
exc.OpenStackCloudException,
self.op_cloud.unregister_machine,
nics,
self.fake_baremetal_node['uuid'])
self.assert_calls()
def test_update_machine_patch_no_action(self):
self.register_uris([dict(
method='GET',

View File

@ -16,7 +16,6 @@ from distutils import version as du_version
import mock
import munch
import testtools
import uuid
import os_client_config as occ
from os_client_config import cloud_config
@ -31,11 +30,6 @@ class TestShadeOperator(base.RequestsMockTestCase):
def setUp(self):
super(TestShadeOperator, self).setUp()
self.machine_id = uuid.uuid4().hex
self.machine_name = self.getUniqueString('machine')
self.node = fakes.make_fake_machine(
machine_id=self.machine_id,
machine_name=self.machine_name)
def test_operator_cloud(self):
self.assertIsInstance(self.op_cloud, shade.OperatorCloud)
@ -71,232 +65,6 @@ class TestShadeOperator(base.RequestsMockTestCase):
self.assertRaises(exc.OpenStackCloudException,
self.op_cloud.list_nics_for_machine, None)
@mock.patch.object(shade.OperatorCloud, 'ironic_client')
def test_register_machine(self, mock_client):
class fake_node(object):
uuid = "00000000-0000-0000-0000-000000000000"
provision_state = "available"
reservation = None
last_error = None
expected_return_value = dict(
uuid="00000000-0000-0000-0000-000000000000",
provision_state="available",
reservation=None,
last_error=None
)
mock_client.node.create.return_value = fake_node
mock_client.node.get.return_value = fake_node
nics = [{'mac': '00:00:00:00:00:00'}]
return_value = self.op_cloud.register_machine(nics)
self.assertDictEqual(expected_return_value, return_value)
self.assertTrue(mock_client.node.create.called)
self.assertTrue(mock_client.port.create.called)
self.assertFalse(mock_client.node.get.called)
@mock.patch.object(shade.OperatorCloud, 'ironic_client')
@mock.patch.object(shade.OperatorCloud, 'node_set_provision_state')
def test_register_machine_enroll(
self,
mock_set_state,
mock_client):
machine_uuid = "00000000-0000-0000-0000-000000000000"
class fake_node_init_state(object):
uuid = machine_uuid
provision_state = "enroll"
reservation = None
last_error = None
class fake_node_post_manage(object):
uuid = machine_uuid
provision_state = "enroll"
reservation = "do you have a flag?"
last_error = None
class fake_node_post_manage_done(object):
uuid = machine_uuid
provision_state = "manage"
reservation = None
last_error = None
class fake_node_post_provide(object):
uuid = machine_uuid
provision_state = "available"
reservation = None
last_error = None
class fake_node_post_enroll_failure(object):
uuid = machine_uuid
provision_state = "enroll"
reservation = None
last_error = "insufficent lolcats"
expected_return_value = dict(
uuid=machine_uuid,
provision_state="available",
reservation=None,
last_error=None
)
mock_client.node.get.side_effect = iter([
fake_node_init_state,
fake_node_post_manage,
fake_node_post_manage_done,
fake_node_post_provide])
mock_client.node.create.return_value = fake_node_init_state
nics = [{'mac': '00:00:00:00:00:00'}]
return_value = self.op_cloud.register_machine(nics)
self.assertDictEqual(expected_return_value, return_value)
self.assertTrue(mock_client.node.create.called)
self.assertTrue(mock_client.port.create.called)
self.assertTrue(mock_client.node.get.called)
mock_client.reset_mock()
mock_client.node.get.side_effect = iter([
fake_node_init_state,
fake_node_post_manage,
fake_node_post_manage_done,
fake_node_post_provide])
return_value = self.op_cloud.register_machine(nics, wait=True)
self.assertDictEqual(expected_return_value, return_value)
self.assertTrue(mock_client.node.create.called)
self.assertTrue(mock_client.port.create.called)
self.assertTrue(mock_client.node.get.called)
mock_client.reset_mock()
mock_client.node.get.side_effect = iter([
fake_node_init_state,
fake_node_post_manage,
fake_node_post_enroll_failure])
self.assertRaises(
shade.OpenStackCloudException,
self.op_cloud.register_machine,
nics)
self.assertRaises(
shade.OpenStackCloudException,
self.op_cloud.register_machine,
nics,
wait=True)
@mock.patch.object(shade.OperatorCloud, 'ironic_client')
@mock.patch.object(shade.OperatorCloud, 'node_set_provision_state')
def test_register_machine_enroll_timeout(
self,
mock_set_state,
mock_client):
machine_uuid = "00000000-0000-0000-0000-000000000000"
class fake_node_init_state(object):
uuid = machine_uuid
provision_state = "enroll"
reservation = "do you have a flag?"
last_error = None
mock_client.node.get.return_value = fake_node_init_state
mock_client.node.create.return_value = fake_node_init_state
nics = [{'mac': '00:00:00:00:00:00'}]
self.assertRaises(
shade.OpenStackCloudException,
self.op_cloud.register_machine,
nics,
lock_timeout=0.001)
self.assertTrue(mock_client.node.create.called)
self.assertTrue(mock_client.port.create.called)
self.assertTrue(mock_client.node.get.called)
mock_client.node.get.reset_mock()
mock_client.node.create.reset_mock()
self.assertRaises(
shade.OpenStackCloudException,
self.op_cloud.register_machine,
nics,
wait=True,
timeout=0.001)
self.assertTrue(mock_client.node.create.called)
self.assertTrue(mock_client.port.create.called)
self.assertTrue(mock_client.node.get.called)
@mock.patch.object(shade.OperatorCloud, 'ironic_client')
def test_register_machine_port_create_failed(self, mock_client):
class fake_node(object):
uuid = "00000000-0000-0000-0000-000000000000"
provision_state = "available"
resevation = None
last_error = None
nics = [{'mac': '00:00:00:00:00:00'}]
mock_client.node.create.return_value = fake_node
mock_client.port.create.side_effect = (
exc.OpenStackCloudException("Error"))
self.assertRaises(exc.OpenStackCloudException,
self.op_cloud.register_machine,
nics)
self.assertTrue(mock_client.node.create.called)
self.assertTrue(mock_client.port.create.called)
self.assertTrue(mock_client.node.delete.called)
@mock.patch.object(shade.OperatorCloud, 'ironic_client')
def test_unregister_machine(self, mock_client):
class fake_node(object):
provision_state = 'available'
class fake_port(object):
uuid = '00000000-0000-0000-0000-000000000001'
mock_client.port.get_by_address.return_value = fake_port
mock_client.node.get.return_value = fake_node
nics = [{'mac': '00:00:00:00:00:00'}]
uuid = "00000000-0000-0000-0000-000000000000"
self.op_cloud.unregister_machine(nics, uuid)
self.assertTrue(mock_client.node.delete.called)
self.assertTrue(mock_client.port.get_by_address.called)
self.assertTrue(mock_client.port.delete.called)
self.assertTrue(mock_client.port.get_by_address.called)
mock_client.port.get_by_address.assert_called_with(
address='00:00:00:00:00:00')
mock_client.port.delete.assert_called_with(
port_id='00000000-0000-0000-0000-000000000001')
@mock.patch.object(shade.OperatorCloud, 'ironic_client')
def test_unregister_machine_unavailable(self, mock_client):
invalid_states = ['active', 'cleaning', 'clean wait', 'clean failed']
nics = [{'mac': '00:00:00:00:00:00'}]
uuid = "00000000-0000-0000-0000-000000000000"
for state in invalid_states:
class fake_node(object):
provision_state = state
mock_client.node.get.return_value = fake_node
self.assertRaises(
exc.OpenStackCloudException,
self.op_cloud.unregister_machine,
nics,
uuid)
self.assertFalse(mock_client.node.delete.called)
self.assertFalse(mock_client.port.delete.called)
self.assertFalse(mock_client.port.get_by_address.called)
self.assertTrue(mock_client.node.get.called)
mock_client.node.reset_mock()
mock_client.node.reset_mock()
@mock.patch.object(shade.OperatorCloud, 'ironic_client')
def test_unregister_machine_timeout(self, mock_client):
class fake_node(object):
provision_state = 'available'
mock_client.node.get.return_value = fake_node
nics = [{'mac': '00:00:00:00:00:00'}]
uuid = "00000000-0000-0000-0000-000000000000"
self.assertRaises(
exc.OpenStackCloudException,
self.op_cloud.unregister_machine,
nics,
uuid,
wait=True,
timeout=0.001)
self.assertTrue(mock_client.node.delete.called)
self.assertTrue(mock_client.port.delete.called)
self.assertTrue(mock_client.port.get_by_address.called)
self.assertTrue(mock_client.node.get.called)
@mock.patch.object(shade.OpenStackCloud, '_image_client')
def test_get_image_name(self, mock_client):