A service for managing and provisioning Bare Metal servers.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

7145 lines
350 KiB

# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Tests for the API /nodes/ methods.
"""
import datetime
from http import client as http_client
import json
import os
from unittest import mock
from urllib import parse as urlparse
import fixtures
from oslo_config import cfg
from oslo_utils import timeutils
from oslo_utils import uuidutils
from testtools import matchers
from ironic.api.controllers import base as api_base
from ironic.api.controllers import v1 as api_v1
from ironic.api.controllers.v1 import node as api_node
from ironic.api.controllers.v1 import notification_utils
from ironic.api.controllers.v1 import utils as api_utils
from ironic.api.controllers.v1 import versions
from ironic.common import boot_devices
from ironic.common import components
from ironic.common import driver_factory
from ironic.common import exception
from ironic.common import indicator_states
from ironic.common import policy
from ironic.common import states
from ironic.conductor import rpcapi
from ironic import objects
from ironic.objects import fields as obj_fields
from ironic import tests as tests_root
from ironic.tests import base
from ironic.tests.unit.api import base as test_api_base
from ironic.tests.unit.api import utils as test_api_utils
from ironic.tests.unit.objects import utils as obj_utils
with open(
os.path.join(
os.path.dirname(tests_root.__file__),
'json_samples', 'network_data.json')) as fl:
NETWORK_DATA = json.load(fl)
class TestListNodes(test_api_base.BaseApiTest):
def setUp(self):
super(TestListNodes, self).setUp()
self.chassis = obj_utils.create_test_chassis(self.context)
p = mock.patch.object(rpcapi.ConductorAPI, 'get_topic_for',
autospec=True)
self.mock_gtf = p.start()
self.mock_gtf.return_value = 'test-topic'
self.addCleanup(p.stop)
self.mock_get_conductor_for = self.useFixture(
fixtures.MockPatchObject(rpcapi.ConductorAPI, 'get_conductor_for',
autospec=True)).mock
self.mock_get_conductor_for.return_value = 'fake.conductor'
def _create_association_test_nodes(self):
# create some unassociated nodes
unassociated_nodes = []
for id in range(3):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid())
unassociated_nodes.append(node.uuid)
# created some associated nodes
associated_nodes = []
for id in range(4):
node = obj_utils.create_test_node(
self.context, uuid=uuidutils.generate_uuid(),
instance_uuid=uuidutils.generate_uuid())
associated_nodes.append(node.uuid)
return {'associated': associated_nodes,
'unassociated': unassociated_nodes}
def test_empty(self):
data = self.get_json('/nodes')
self.assertEqual([], data['nodes'])
def test_one(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
data = self.get_json(
'/nodes',
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertIn('instance_uuid', data['nodes'][0])
self.assertIn('maintenance', data['nodes'][0])
self.assertIn('power_state', data['nodes'][0])
self.assertIn('provision_state', data['nodes'][0])
self.assertIn('uuid', data['nodes'][0])
self.assertEqual(node.uuid, data['nodes'][0]["uuid"])
self.assertNotIn('driver', data['nodes'][0])
self.assertNotIn('driver_info', data['nodes'][0])
self.assertNotIn('driver_internal_info', data['nodes'][0])
self.assertNotIn('extra', data['nodes'][0])
self.assertNotIn('properties', data['nodes'][0])
self.assertNotIn('chassis_uuid', data['nodes'][0])
self.assertNotIn('reservation', data['nodes'][0])
self.assertNotIn('console_enabled', data['nodes'][0])
self.assertNotIn('target_power_state', data['nodes'][0])
self.assertNotIn('target_provision_state', data['nodes'][0])
self.assertNotIn('provision_updated_at', data['nodes'][0])
self.assertNotIn('maintenance_reason', data['nodes'][0])
self.assertNotIn('clean_step', data['nodes'][0])
self.assertNotIn('raid_config', data['nodes'][0])
self.assertNotIn('target_raid_config', data['nodes'][0])
self.assertNotIn('network_interface', data['nodes'][0])
self.assertNotIn('resource_class', data['nodes'][0])
for field in api_utils.V31_FIELDS:
self.assertNotIn(field, data['nodes'][0])
self.assertNotIn('storage_interface', data['nodes'][0])
self.assertNotIn('traits', data['nodes'][0])
# never expose the chassis_id
self.assertNotIn('chassis_id', data['nodes'][0])
self.assertNotIn('bios_interface', data['nodes'][0])
self.assertNotIn('deploy_step', data['nodes'][0])
self.assertNotIn('conductor_group', data['nodes'][0])
self.assertNotIn('automated_clean', data['nodes'][0])
self.assertNotIn('protected', data['nodes'][0])
self.assertNotIn('protected_reason', data['nodes'][0])
self.assertNotIn('owner', data['nodes'][0])
self.assertNotIn('retired', data['nodes'][0])
self.assertNotIn('retired_reason', data['nodes'][0])
self.assertNotIn('lessee', data['nodes'][0])
self.assertNotIn('network_data', data['nodes'][0])
def test_get_one(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(node.uuid, data['uuid'])
self.assertIn('driver', data)
self.assertIn('driver_info', data)
self.assertEqual('******', data['driver_info']['fake_password'])
self.assertEqual('bar', data['driver_info']['foo'])
self.assertIn('driver_internal_info', data)
self.assertIn('instance_info', data)
self.assertEqual('******', data['instance_info']['configdrive'])
self.assertEqual('******', data['instance_info']['image_url'])
self.assertEqual('bar', data['instance_info']['foo'])
self.assertIn('extra', data)
self.assertIn('properties', data)
self.assertIn('chassis_uuid', data)
self.assertIn('reservation', data)
self.assertIn('maintenance_reason', data)
self.assertIn('name', data)
self.assertIn('inspection_finished_at', data)
self.assertIn('inspection_started_at', data)
self.assertIn('clean_step', data)
self.assertIn('states', data)
self.assertIn('network_interface', data)
self.assertIn('resource_class', data)
for field in api_utils.V31_FIELDS:
self.assertIn(field, data)
self.assertIn('storage_interface', data)
self.assertIn('traits', data)
# never expose the chassis_id
self.assertNotIn('chassis_id', data)
self.assertIn('bios_interface', data)
self.assertIn('deploy_step', data)
self.assertIn('conductor_group', data)
self.assertIn('automated_clean', data)
self.assertIn('protected', data)
self.assertIn('protected_reason', data)
self.assertIn('owner', data)
self.assertIn('lessee', data)
self.assertNotIn('allocation_id', data)
self.assertIn('allocation_uuid', data)
def test_get_one_with_json(self):
# Test backward compatibility with guess_content_type_from_ext
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
data = self.get_json(
'/nodes/%s.json' % node.uuid,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(node.uuid, data['uuid'])
def test_get_one_with_json_in_name(self):
# Test that it is possible to name a node ending with .json
node = obj_utils.create_test_node(self.context,
name='node.json',
chassis_id=self.chassis.id)
data = self.get_json(
'/nodes/%s' % node.name,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(node.uuid, data['uuid'])
def test_get_one_with_suffix(self):
# This tests that we don't mess with mime-like suffixes
node = obj_utils.create_test_node(self.context,
name='test.1',
chassis_id=self.chassis.id)
data = self.get_json(
'/nodes/%s' % node.name,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(node.uuid, data['uuid'])
def test_get_one_with_double_json(self):
# Check that .json is only stripped once
node = obj_utils.create_test_node(self.context,
name='node.json',
chassis_id=self.chassis.id)
data = self.get_json(
'/nodes/%s.json' % node.name,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(node.uuid, data['uuid'])
def _test_node_field_hidden_in_lower_version(self, field,
old_version, new_version):
node = obj_utils.create_test_node(self.context)
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: old_version})
self.assertNotIn(field, data)
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: new_version})
self.assertIn(field, data)
def test_node_states_field_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('states', '1.8', '1.14')
def test_node_interface_fields_hidden_in_lower_version(self):
node = obj_utils.create_test_node(self.context)
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.30'})
for field in api_utils.V31_FIELDS:
self.assertNotIn(field, data)
def test_node_storage_interface_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('storage_interface',
'1.32', '1.33')
def test_node_traits_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('traits', '1.36', '1.37')
def test_node_bios_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('bios_interface',
'1.39', '1.40')
def test_node_inspect_wait_state_between_api_versions(self):
node = obj_utils.create_test_node(self.context,
provision_state='inspect wait')
lower_version_data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.38'})
self.assertEqual('inspecting', lower_version_data['provision_state'])
higher_version_data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.39'})
self.assertEqual('inspect wait',
higher_version_data['provision_state'])
def test_node_fault_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('fault',
'1.41', '1.42')
def test_node_deploy_step_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('deploy_step',
'1.43', '1.44')
def test_node_conductor_group_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('conductor_group',
'1.45', '1.46')
def test_node_automated_clean_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('automated_clean',
'1.46', '1.47')
def test_node_automated_clean_null_field(self):
node = obj_utils.create_test_node(self.context, automated_clean=None)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.47'})
self.assertIsNone(data['automated_clean'])
def test_node_automated_clean_true_field(self):
node = obj_utils.create_test_node(self.context, automated_clean=True)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.47'})
self.assertEqual(data['automated_clean'], True)
def test_node_automated_clean_false_field(self):
node = obj_utils.create_test_node(self.context, automated_clean=False)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.47'})
self.assertEqual(data['automated_clean'], False)
def test_node_protected_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('protected',
'1.47', '1.48')
def test_node_protected_reason_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('protected_reason',
'1.47', '1.48')
def test_node_conductor_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('conductor',
'1.48', '1.49')
def test_node_protected(self):
for value in (True, False):
node = obj_utils.create_test_node(self.context, protected=value,
provision_state='active',
uuid=uuidutils.generate_uuid())
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.48'})
self.assertIs(data['protected'], value)
self.assertIsNone(data['protected_reason'])
def test_node_protected_with_reason(self):
node = obj_utils.create_test_node(self.context, protected=True,
provision_state='active',
protected_reason='reason!')
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.48'})
self.assertTrue(data['protected'])
self.assertEqual('reason!', data['protected_reason'])
def test_node_owner_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('owner',
'1.49', '1.50')
def test_node_owner_null_field(self):
node = obj_utils.create_test_node(self.context, owner=None)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.50'})
self.assertIsNone(data['owner'])
def test_node_owner_present(self):
node = obj_utils.create_test_node(self.context,
owner="akindofmagic")
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.50'})
self.assertEqual(data['owner'], "akindofmagic")
def test_node_description_null_field(self):
node = obj_utils.create_test_node(self.context, description=None)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.51'})
self.assertIsNone(data['description'])
def test_node_retired_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('retired',
'1.60', '1.61')
def test_node_retired_reason_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('retired_reason',
'1.60', '1.61')
def test_node_retired(self):
for value in (True, False):
node = obj_utils.create_test_node(self.context, retired=value,
provision_state='active',
uuid=uuidutils.generate_uuid())
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.61'})
self.assertIs(data['retired'], value)
self.assertIsNone(data['retired_reason'])
def test_node_retired_with_reason(self):
node = obj_utils.create_test_node(self.context, retired=True,
provision_state='active',
retired_reason='warranty expired')
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.61'})
self.assertTrue(data['retired'])
self.assertEqual('warranty expired', data['retired_reason'])
def test_node_lessee_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('lessee',
'1.64', '1.65')
def test_node_lessee_null_field(self):
node = obj_utils.create_test_node(self.context, lessee=None)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.65'})
self.assertIsNone(data['lessee'])
def test_node_lessee_present(self):
node = obj_utils.create_test_node(self.context,
lessee="some-lucky-project")
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.65'})
self.assertEqual(data['lessee'], "some-lucky-project")
def test_node_network_data_hidden_in_lower_version(self):
self._test_node_field_hidden_in_lower_version('network_data',
'1.65', '1.66')
def test_node_network_data(self):
node = obj_utils.create_test_node(
self.context, network_data=NETWORK_DATA,
provision_state='active',
uuid=uuidutils.generate_uuid())
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.66'})
self.assertEqual(data['network_data'], NETWORK_DATA)
def test_get_one_custom_fields(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'extra,instance_info'
data = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: str(api_v1.max_version())})
# We always append "links"
self.assertCountEqual(['extra', 'instance_info', 'links'], data)
def test_get_collection_custom_fields(self):
fields = 'uuid,instance_info'
for i in range(3):
obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
instance_uuid=uuidutils.generate_uuid())
data = self.get_json(
'/nodes?fields=%s' % fields,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(3, len(data['nodes']))
for node in data['nodes']:
# We always append "links"
self.assertCountEqual(['uuid', 'instance_info', 'links'], node)
def test_get_custom_fields_invalid_fields(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'uuid,spongebob'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: str(api_v1.max_version())},
expect_errors=True)
self.assertEqual(http_client.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertIn('spongebob', response.json['error_message'])
def test_get_custom_fields_invalid_api_version(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'uuid,extra'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: str(api_v1.min_version())},
expect_errors=True)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_int)
def test_get_one_custom_fields_show_password(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id,
driver_info={'fake_password': 'bar'})
fields = 'driver_info'
data = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: str(api_v1.max_version())})
# We always append "links"
self.assertCountEqual(['driver_info', 'links'], data)
self.assertEqual('******', data['driver_info']['fake_password'])
def test_get_network_interface_fields_invalid_api_version(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'network_interface'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: str('1.19')},
expect_errors=True)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_int)
def test_get_network_interface_fields(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'network_interface'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertIn('network_interface', response)
def test_get_all_interface_fields_invalid_api_version(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields_arg = ','.join(api_utils.V31_FIELDS)
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields_arg),
headers={api_base.Version.string: '1.30'},
expect_errors=True)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_int)
def test_get_all_interface_fields(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields_arg = ','.join(api_utils.V31_FIELDS)
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields_arg),
headers={api_base.Version.string: str(api_v1.max_version())})
for field in api_utils.V31_FIELDS:
self.assertIn(field, response)
def test_get_storage_interface_fields_invalid_api_version(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'storage_interface'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.32'},
expect_errors=True)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_int)
def test_get_storage_interface_fields(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'storage_interface'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertIn('storage_interface', response)
def test_get_traits_fields_invalid_api_version(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'traits'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.36'},
expect_errors=True)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_int)
def test_get_traits_fields(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'traits'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertIn('traits', response)
def test_get_conductor_group_fields_invalid_api_version(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'conductor_group'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.45'},
expect_errors=True)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_int)
def test_get_conductor_group_fields(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'conductor_group'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.46'})
self.assertIn('conductor_group', response)
def test_get_automated_clean_fields(self):
node = obj_utils.create_test_node(self.context,
automated_clean=True)
fields = 'automated_clean'
response = self.get_json('/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.47'})
self.assertIn('automated_clean', response)
def test_get_protected_fields(self):
node = obj_utils.create_test_node(self.context,
protected=True)
response = self.get_json('/nodes/%s?fields=%s' %
(node.uuid, 'protected'),
headers={api_base.Version.string: '1.48'})
self.assertIn('protected', response)
def test_get_conductor_field_invalid_api_version(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'conductor'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.48'},
expect_errors=True)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_int)
def test_get_conductor_field(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
fields = 'conductor'
response = self.get_json(
'/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.49'})
self.assertIn('conductor', response)
def test_get_owner_fields(self):
node = obj_utils.create_test_node(self.context, owner='fred')
fields = 'owner'
response = self.get_json('/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.50'})
self.assertIn('owner', response)
def test_get_description_field(self):
node = obj_utils.create_test_node(self.context,
description='useful piece')
fields = 'description'
response = self.get_json('/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.51'})
self.assertIn('description', response)
def test_get_lessee_field(self):
node = obj_utils.create_test_node(self.context,
lessee='some-lucky-project')
fields = 'lessee'
response = self.get_json('/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.65'})
self.assertIn('lessee', response)
def test_get_with_allocation(self):
allocation = obj_utils.create_test_allocation(self.context)
node = obj_utils.create_test_node(self.context,
allocation_id=allocation.id)
fields = 'allocation_uuid'
response = self.get_json('/nodes/%s?fields=%s' % (node.uuid, fields),
headers={api_base.Version.string: '1.52'})
self.assertEqual(allocation.uuid, response['allocation_uuid'])
def test_get_retired_fields(self):
node = obj_utils.create_test_node(self.context,
retired=True)
response = self.get_json('/nodes/%s?fields=%s' %
(node.uuid, 'retired'),
headers={api_base.Version.string: '1.61'})
self.assertIn('retired', response)
def test_get_one_with_no_agent_secret(self):
node = obj_utils.create_test_node(
self.context,
driver_internal_info={'agent_secret_token': 'abcdefg'})
response = self.get_json('/nodes/%s' % (node.uuid),
headers={api_base.Version.string: '1.52'})
token_value = response['driver_internal_info']['agent_secret_token']
self.assertEqual('******', token_value)
def test_detail(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
data = self.get_json(
'/nodes/detail',
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(node.uuid, data['nodes'][0]["uuid"])
self.assertIn('name', data['nodes'][0])
self.assertIn('driver', data['nodes'][0])
self.assertIn('driver_info', data['nodes'][0])
self.assertIn('extra', data['nodes'][0])
self.assertIn('properties', data['nodes'][0])
self.assertIn('chassis_uuid', data['nodes'][0])
self.assertIn('reservation', data['nodes'][0])
self.assertIn('maintenance', data['nodes'][0])
self.assertIn('console_enabled', data['nodes'][0])
self.assertIn('target_power_state', data['nodes'][0])
self.assertIn('target_provision_state', data['nodes'][0])
self.assertIn('provision_updated_at', data['nodes'][0])
self.assertIn('inspection_finished_at', data['nodes'][0])
self.assertIn('inspection_started_at', data['nodes'][0])
self.assertIn('raid_config', data['nodes'][0])
self.assertIn('target_raid_config', data['nodes'][0])
self.assertIn('network_interface', data['nodes'][0])
self.assertIn('resource_class', data['nodes'][0])
for field in api_utils.V31_FIELDS:
self.assertIn(field, data['nodes'][0])
self.assertIn('storage_interface', data['nodes'][0])
self.assertIn('traits', data['nodes'][0])
self.assertIn('conductor_group', data['nodes'][0])
self.assertIn('automated_clean', data['nodes'][0])
self.assertIn('protected', data['nodes'][0])
self.assertIn('protected_reason', data['nodes'][0])
self.assertIn('owner', data['nodes'][0])
self.assertIn('lessee', data['nodes'][0])
# never expose the chassis_id
self.assertNotIn('chassis_id', data['nodes'][0])
self.assertNotIn('allocation_id', data['nodes'][0])
self.assertIn('allocation_uuid', data['nodes'][0])
self.assertIn('retired', data['nodes'][0])
self.assertIn('retired_reason', data['nodes'][0])
self.assertIn('network_data', data['nodes'][0])
def test_detail_using_query(self):
node = obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
data = self.get_json(
'/nodes?detail=True',
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(node.uuid, data['nodes'][0]["uuid"])
self.assertIn('name', data['nodes'][0])
self.assertIn('driver', data['nodes'][0])
self.assertIn('driver_info', data['nodes'][0])
self.assertIn('extra', data['nodes'][0])
self.assertIn('properties', data['nodes'][0])
self.assertIn('chassis_uuid', data['nodes'][0])
self.assertIn('reservation', data['nodes'][0])
self.assertIn('maintenance', data['nodes'][0])
self.assertIn('console_enabled', data['nodes'][0])
self.assertIn('target_power_state', data['nodes'][0])
self.assertIn('target_provision_state', data['nodes'][0])
self.assertIn('provision_updated_at', data['nodes'][0])
self.assertIn('inspection_finished_at', data['nodes'][0])
self.assertIn('inspection_started_at', data['nodes'][0])
self.assertIn('raid_config', data['nodes'][0])
self.assertIn('target_raid_config', data['nodes'][0])
self.assertIn('network_interface', data['nodes'][0])
self.assertIn('resource_class', data['nodes'][0])
self.assertIn('conductor_group', data['nodes'][0])
self.assertIn('automated_clean', data['nodes'][0])
self.assertIn('protected', data['nodes'][0])
self.assertIn('protected_reason', data['nodes'][0])
self.assertIn('owner', data['nodes'][0])
self.assertIn('lessee', data['nodes'][0])
for field in api_utils.V31_FIELDS:
self.assertIn(field, data['nodes'][0])
# never expose the chassis_id
self.assertNotIn('chassis_id', data['nodes'][0])
self.assertIn('retired', data['nodes'][0])
self.assertIn('retired_reason', data['nodes'][0])
self.assertIn('network_data', data['nodes'][0])
def test_detail_query_false(self):
obj_utils.create_test_node(self.context)
data1 = self.get_json(
'/nodes',
headers={api_base.Version.string: str(api_v1.max_version())})
data2 = self.get_json(
'/nodes?detail=False',
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(data1['nodes'], data2['nodes'])
def test_detail_using_query_false_and_fields(self):
obj_utils.create_test_node(self.context)
data = self.get_json(
'/nodes?detail=False&fields=name',
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertIn('name', data['nodes'][0])
self.assertNotIn('uuid', data['nodes'][0])
def test_detail_using_query_and_fields(self):
obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
response = self.get_json(
'/nodes?detail=True&fields=name',
headers={api_base.Version.string: str(api_v1.max_version())},
expect_errors=True)
self.assertEqual(http_client.BAD_REQUEST, response.status_int)
def test_detail_using_query_old_version(self):
obj_utils.create_test_node(self.context,
chassis_id=self.chassis.id)
response = self.get_json(
'/nodes?detail=True',
headers={api_base.Version.string: str(api_v1.min_version())},
expect_errors=True)
self.assertEqual(http_client.BAD_REQUEST, response.status_int)
def test_detail_against_single(self):
node = obj_utils.create_test_node(self.context)
response = self.get_json('/nodes/%s/detail' % node.uuid,
expect_errors=True)
self.assertEqual(http_client.NOT_FOUND, response.status_int)
@mock.patch.object(policy, 'authorize', spec=True)
def test_detail_forbidden(self, mock_authorize):
def mock_authorize_function(rule, target, creds):
raise exception.HTTPForbidden(resource='fake')
mock_authorize.side_effect = mock_authorize_function
response = self.get_json('/nodes/detail', expect_errors=True,
headers={
api_base.Version.string: '1.50',
'X-Project-Id': '12345'
})
self.assertEqual(http_client.FORBIDDEN, response.status_int)
@mock.patch.object(policy, 'authorize', spec=True)
def test_detail_list_all_forbidden_no_project(self, mock_authorize):
def mock_authorize_function(rule, target, creds):
if rule == 'baremetal:node:list_all':
raise exception.HTTPForbidden(resource='fake')
return True
mock_authorize.side_effect = mock_authorize_function
response = self.get_json('/nodes/detail', expect_errors=True,
headers={
api_base.Version.string: '1.49',
})
self.assertEqual(http_client.FORBIDDEN, response.status_int)
@mock.patch.object(policy, 'authorize', spec=True)
def test_detail_list_all_forbid_project_mismatch(self, mock_authorize):
def mock_authorize_function(rule, target, creds):
if rule == 'baremetal:node:list_all':
raise exception.HTTPForbidden(resource='fake')
return True
mock_authorize.side_effect = mock_authorize_function
response = self.get_json('/nodes/detail?project=54321',
expect_errors=True,
headers={
api_base.Version.string: '1.50',
'X-Project-Id': '12345'
})
self.assertEqual(http_client.FORBIDDEN, response.status_int)
@mock.patch.object(policy, 'authorize', spec=True)
def test_detail_list_all_forbidden(self, mock_authorize):
def mock_authorize_function(rule, target, creds):
if rule == 'baremetal:node:list_all':
raise exception.HTTPForbidden(resource='fake')
return True
mock_authorize.side_effect = mock_authorize_function
nodes = []
for id in range(3):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
owner='12345')
nodes.append(node.uuid)
for id in range(3):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
lessee='12345')
nodes.append(node.uuid)
for id in range(2):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
owner='54321')
for id in range(2):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
lessee='54321')
data = self.get_json('/nodes/detail', headers={
api_base.Version.string: '1.65',
'X-Project-Id': '12345'})
self.assertEqual(len(nodes), len(data['nodes']))
uuids = [n['uuid'] for n in data['nodes']]
self.assertEqual(sorted(nodes), sorted(uuids))
def test_mask_available_state(self):
node = obj_utils.create_test_node(self.context,
provision_state=states.AVAILABLE)
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: str(api_v1.min_version())})
self.assertEqual(states.NOSTATE, data['provision_state'])
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: "1.2"})
self.assertEqual(states.AVAILABLE, data['provision_state'])
def test_hide_fields_in_newer_versions_driver_internal(self):
node = obj_utils.create_test_node(self.context,
driver_internal_info={"foo": "bar"})
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: str(api_v1.min_version())})
self.assertNotIn('driver_internal_info', data)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: "1.3"})
self.assertEqual({"foo": "bar"}, data['driver_internal_info'])
def test_hide_fields_in_newer_versions_name(self):
node = obj_utils.create_test_node(self.context,
name="fish")
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: "1.4"})
self.assertNotIn('name', data)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: "1.5"})
self.assertEqual('fish', data['name'])
def test_hide_fields_in_newer_versions_inspection(self):
some_time = datetime.datetime(2015, 3, 18, 19, 20)
node = obj_utils.create_test_node(self.context,
inspection_started_at=some_time)
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: str(api_v1.min_version())})
self.assertNotIn('inspection_finished_at', data)
self.assertNotIn('inspection_started_at', data)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: "1.6"})
started = timeutils.parse_isotime(
data['inspection_started_at']).replace(tzinfo=None)
self.assertEqual(some_time, started)
self.assertIsNone(data['inspection_finished_at'])
def test_hide_fields_in_newer_versions_clean_step(self):
node = obj_utils.create_test_node(self.context,
clean_step={"foo": "bar"})
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: str(api_v1.min_version())})
self.assertNotIn('clean_step', data)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: "1.7"})
self.assertEqual({"foo": "bar"}, data['clean_step'])
def test_hide_fields_in_newer_versions_network_interface(self):
node = obj_utils.create_test_node(self.context,
network_interface='flat')
data = self.get_json(
'/nodes/detail', headers={api_base.Version.string: '1.19'})
self.assertNotIn('network_interface', data['nodes'][0])
new_data = self.get_json(
'/nodes/detail', headers={api_base.Version.string: '1.20'})
self.assertEqual(node.network_interface,
new_data['nodes'][0]["network_interface"])
def test_hide_fields_in_newer_versions_resource_class(self):
node = obj_utils.create_test_node(self.context,
resource_class='foo')
data = self.get_json(
'/nodes/detail', headers={api_base.Version.string: '1.20'})
self.assertNotIn('resource_class', data['nodes'][0])
new_data = self.get_json(
'/nodes/detail', headers={api_base.Version.string: '1.21'})
self.assertEqual(node.resource_class,
new_data['nodes'][0]["resource_class"])
def test_hide_fields_in_newer_versions_interface_fields(self):
node = obj_utils.create_test_node(self.context)
data = self.get_json(
'/nodes/detail', headers={api_base.Version.string: '1.30'})
for field in api_utils.V31_FIELDS:
self.assertNotIn(field, data['nodes'][0])
new_data = self.get_json(
'/nodes/detail', headers={api_base.Version.string: '1.31'})
for field in api_utils.V31_FIELDS:
self.assertEqual(getattr(node, field),
new_data['nodes'][0][field])
def test_hide_fields_in_newer_versions_volume(self):
node = obj_utils.create_test_node(self.context)
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.31'})
self.assertNotIn('volume', data)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: "1.32"})
self.assertIn('volume', data)
def test_hide_fields_in_newer_versions_storage_interface(self):
node = obj_utils.create_test_node(self.context,
storage_interface='cinder')
data = self.get_json(
'/nodes/detail', headers={api_base.Version.string: '1.32'})
self.assertNotIn('storage_interface', data['nodes'][0])
new_data = self.get_json(
'/nodes/detail', headers={api_base.Version.string: '1.33'})
self.assertEqual(node.storage_interface,
new_data['nodes'][0]["storage_interface"])
def test_hide_fields_in_newer_versions_traits(self):
node = obj_utils.create_test_node(self.context)
objects.TraitList.create(self.context, node.id, ['CUSTOM_1'])
node.refresh()
data = self.get_json(
'/nodes/detail', headers={api_base.Version.string: '1.36'})
self.assertNotIn('traits', data['nodes'][0])
new_data = self.get_json(
'/nodes/detail', headers={api_base.Version.string: '1.37'})
self.assertEqual(['CUSTOM_1'], new_data['nodes'][0]["traits"])
def test_hide_fields_in_newer_versions_description(self):
node = obj_utils.create_test_node(self.context,
description="useful piece")
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: "1.50"})
self.assertNotIn('description', data)
data = self.get_json('/nodes/%s' % node.uuid,
headers={api_base.Version.string: "1.51"})
self.assertEqual('useful piece', data['description'])
def test_many(self):
nodes = []
for id in range(5):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid())
nodes.append(node.uuid)
data = self.get_json('/nodes')
self.assertEqual(len(nodes), len(data['nodes']))
uuids = [n['uuid'] for n in data['nodes']]
self.assertEqual(sorted(nodes), sorted(uuids))
def test_many_have_names(self):
nodes = []
node_names = []
for id in range(5):
name = 'node-%s' % id
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
name=name)
nodes.append(node.uuid)
node_names.append(name)
data = self.get_json('/nodes',
headers={api_base.Version.string: "1.5"})
names = [n['name'] for n in data['nodes']]
self.assertEqual(len(nodes), len(data['nodes']))
self.assertEqual(sorted(node_names), sorted(names))
@mock.patch.object(policy, 'authorize', spec=True)
def test_many_forbidden(self, mock_authorize):
def mock_authorize_function(rule, target, creds):
raise exception.HTTPForbidden(resource='fake')
mock_authorize.side_effect = mock_authorize_function
response = self.get_json('/nodes', expect_errors=True,
headers={
api_base.Version.string: '1.50',
'X-Project-Id': '12345'
})
self.assertEqual(http_client.FORBIDDEN, response.status_int)
@mock.patch.object(policy, 'authorize', spec=True)
def test_many_list_all_forbidden_no_project(self, mock_authorize):
def mock_authorize_function(rule, target, creds):
if rule == 'baremetal:node:list_all':
raise exception.HTTPForbidden(resource='fake')
return True
mock_authorize.side_effect = mock_authorize_function
response = self.get_json('/nodes', expect_errors=True,
headers={
api_base.Version.string: '1.49',
})
self.assertEqual(http_client.FORBIDDEN, response.status_int)
@mock.patch.object(policy, 'authorize', spec=True)
def test_many_list_all_forbid_project_mismatch(self, mock_authorize):
def mock_authorize_function(rule, target, creds):
if rule == 'baremetal:node:list_all':
raise exception.HTTPForbidden(resource='fake')
return True
mock_authorize.side_effect = mock_authorize_function
response = self.get_json('/nodes?project=54321',
expect_errors=True,
headers={
api_base.Version.string: '1.50',
'X-Project-Id': '12345'
})
self.assertEqual(http_client.FORBIDDEN, response.status_int)
@mock.patch.object(policy, 'authorize', spec=True)
def test_many_list_all_forbidden(self, mock_authorize):
def mock_authorize_function(rule, target, creds):
if rule == 'baremetal:node:list_all':
raise exception.HTTPForbidden(resource='fake')
return True
mock_authorize.side_effect = mock_authorize_function
nodes = []
for id in range(3):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
owner='12345')
nodes.append(node.uuid)
for id in range(3):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
lessee='12345')
nodes.append(node.uuid)
for id in range(2):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
owner='54321')
for id in range(2):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
lessee='54321')
data = self.get_json('/nodes', headers={
api_base.Version.string: '1.65',
'X-Project-Id': '12345'})
self.assertEqual(len(nodes), len(data['nodes']))
uuids = [n['uuid'] for n in data['nodes']]
self.assertEqual(sorted(nodes), sorted(uuids))
def _test_links(self, public_url=None):
cfg.CONF.set_override('public_endpoint', public_url, 'api')
uuid = uuidutils.generate_uuid()
obj_utils.create_test_node(self.context, uuid=uuid)
data = self.get_json('/nodes/%s' % uuid)
self.assertIn('links', data)
self.assertEqual(2, len(data['links']))
self.assertIn(uuid, data['links'][0]['href'])
for link in data['links']:
bookmark = link['rel'] == 'bookmark'
self.assertTrue(self.validate_link(link['href'],
bookmark=bookmark))
if public_url is not None:
expected = [{'href': '%s/v1/nodes/%s' % (public_url, uuid),
'rel': 'self'},
{'href': '%s/nodes/%s' % (public_url, uuid),
'rel': 'bookmark'}]
for i in expected:
self.assertIn(i, data['links'])
def test_links(self):
self._test_links()
def test_links_public_url(self):
self._test_links(public_url='http://foo')
def test_collection_links(self):
nodes = []
for id in range(5):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid())
nodes.append(node.uuid)
data = self.get_json('/nodes/?limit=3')
self.assertEqual(3, len(data['nodes']))
next_marker = data['nodes'][-1]['uuid']
self.assertIn(next_marker, data['next'])
def test_collection_links_default_limit(self):
cfg.CONF.set_override('max_limit', 3, 'api')
nodes = []
for id in range(5):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid())
nodes.append(node.uuid)
data = self.get_json('/nodes')
self.assertEqual(3, len(data['nodes']))
next_marker = data['nodes'][-1]['uuid']
self.assertIn(next_marker, data['next'])
def test_collection_links_custom_fields(self):
fields = 'driver_info,uuid'
cfg.CONF.set_override('max_limit', 3, 'api')
nodes = []
for id in range(5):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
driver_info={'fake': 'value'},
properties={'fake': 'bar'})
nodes.append(node.uuid)
data = self.get_json(
'/nodes?fields=%s' % fields,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(3, len(data['nodes']))
next_marker = data['nodes'][-1]['uuid']
self.assertIn(next_marker, data['next'])
self.assertIn('fields', data['next'])
def test_get_collection_pagination_no_uuid(self):
fields = 'name'
limit = 2
nodes = []
for id_ in range(3):
node = obj_utils.create_test_node(
self.context,
uuid=uuidutils.generate_uuid())
nodes.append(node)
data = self.get_json(
'/nodes?fields=%s&limit=%s' % (fields, limit),
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(limit, len(data['nodes']))
self.assertIn('marker=%s' % nodes[limit - 1].uuid, data['next'])
def test_collection_links_instance_uuid_param(self):
cfg.CONF.set_override('max_limit', 1, 'api')
nodes = []
for id in range(2):
node = obj_utils.create_test_node(
self.context,
uuid=uuidutils.generate_uuid(),
instance_uuid=uuidutils.generate_uuid(),
resource_class='tst_resource')
nodes.append(node)
query_str = 'instance_uuid=%s' % nodes[0].instance_uuid
data = self.get_json('/nodes?%s' % query_str)
self.assertEqual(1, len(data['nodes']))
self.assertNotIn('next', data)
def test_sort_key(self):
nodes = []
for id in range(3):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid())
nodes.append(node.uuid)
data = self.get_json('/nodes?sort_key=uuid')
uuids = [n['uuid'] for n in data['nodes']]
self.assertEqual(sorted(nodes), uuids)
def test_sort_key_invalid(self):
invalid_keys_list = ['foo', 'properties', 'driver_info', 'extra',
'instance_info', 'driver_internal_info',
'clean_step', 'traits']
headers = {api_base.Version.string: str(api_v1.max_version())}
for invalid_key in invalid_keys_list:
response = self.get_json('/nodes?sort_key=%s' % invalid_key,
headers=headers,
expect_errors=True)
self.assertEqual(http_client.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertIn(invalid_key, response.json['error_message'])
def _test_sort_key_allowed(self, detail=False):
node_uuids = []
for id in range(3, 0, -1):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
resource_class='rc_%s' % id)
node_uuids.append(node.uuid)
node_uuids.reverse()
headers = {'X-OpenStack-Ironic-API-Version': '1.21'}
detail_str = '/detail' if detail else ''
data = self.get_json('/nodes%s?sort_key=resource_class' % detail_str,
headers=headers)
data_uuids = [n['uuid'] for n in data['nodes']]
self.assertEqual(node_uuids, data_uuids)
def test_sort_key_allowed(self):
self._test_sort_key_allowed()
def test_detail_sort_key_allowed(self):
self._test_sort_key_allowed(detail=True)
def _test_sort_key_not_allowed(self, detail=False):
headers = {'X-OpenStack-Ironic-API-Version': '1.20'}
detail_str = '/detail' if detail else ''
resp = self.get_json('/nodes%s?sort_key=resource_class' % detail_str,
headers=headers, expect_errors=True)
self.assertEqual(http_client.NOT_ACCEPTABLE, resp.status_int)
self.assertEqual('application/json', resp.content_type)
def test_sort_key_not_allowed(self):
self._test_sort_key_not_allowed()
def test_detail_sort_key_not_allowed(self):
self._test_sort_key_not_allowed(detail=True)
def test_ports_subresource_link(self):
node = obj_utils.create_test_node(self.context)
data = self.get_json('/nodes/%s' % node.uuid)
self.assertIn('ports', data)
def test_portgroups_subresource(self):
node = obj_utils.create_test_node(self.context)
headers = {'X-OpenStack-Ironic-API-Version': '1.24'}
for id_ in range(2):
obj_utils.create_test_portgroup(self.context, node_id=node.id,
name="pg-%s" % id_,
uuid=uuidutils.generate_uuid(),
address='52:54:00:cf:2d:3%s' % id_)
data = self.get_json('/nodes/%s/portgroups' % node.uuid,
headers=headers)
self.assertEqual(2, len(data['portgroups']))
self.assertNotIn('next', data)
# Test collection pagination
data = self.get_json('/nodes/%s/portgroups?limit=1' % node.uuid,
headers=headers)
self.assertEqual(1, len(data['portgroups']))
self.assertIn('next', data)
def test_portgroups_subresource_link(self):
node = obj_utils.create_test_node(self.context)
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={'X-OpenStack-Ironic-API-Version': '1.24'})
self.assertIn('portgroups', data)
def test_portgroups_subresource_link_hidden_for_older_versions(self):
node = obj_utils.create_test_node(self.context)
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={'X-OpenStack-Ironic-API-Version': '1.20'})
self.assertNotIn('portgroups', data)
def test_portgroups_subresource_old_api_version(self):
node = obj_utils.create_test_node(self.context)
response = self.get_json(
'/nodes/%s/portgroups' % node.uuid, expect_errors=True,
headers={'X-OpenStack-Ironic-API-Version': '1.23'})
self.assertEqual(http_client.NOT_FOUND, response.status_int)
def test_ports_subresource(self):
node = obj_utils.create_test_node(self.context)
for id_ in range(2):
obj_utils.create_test_port(self.context, node_id=node.id,
uuid=uuidutils.generate_uuid(),
address='52:54:00:cf:2d:3%s' % id_)
data = self.get_json('/nodes/%s/ports' % node.uuid)
self.assertEqual(2, len(data['ports']))
self.assertNotIn('next', data)
# Test collection pagination
data = self.get_json('/nodes/%s/ports?limit=1' % node.uuid)
self.assertEqual(1, len(data['ports']))
self.assertIn('next', data)
def test_ports_subresource_noid(self):
node = obj_utils.create_test_node(self.context)
obj_utils.create_test_port(self.context, node_id=node.id)
# No node id specified
response = self.get_json('/nodes/ports', expect_errors=True)
self.assertEqual(http_client.NOT_FOUND, response.status_int)
def test_ports_subresource_node_not_found(self):
non_existent_uuid = 'eeeeeeee-cccc-aaaa-bbbb-cccccccccccc'
response = self.get_json('/nodes/%s/ports' % non_existent_uuid,
expect_errors=True)
self.assertEqual(http_client.NOT_FOUND, response.status_int)
def test_ports_subresource_invalid_ident(self):
invalid_ident = '123 123'
response = self.get_json('/nodes/%s/ports' % invalid_ident,
expect_errors=True)
self.assertEqual(http_client.BAD_REQUEST, response.status_int)
self.assertIn('Expected UUID or name for node',
response.json['error_message'])
def test_ports_subresource_via_portgroups_subres_not_allowed(self):
node = obj_utils.create_test_node(self.context)
pg = obj_utils.create_test_portgroup(self.context,
node_id=node.id)
response = self.get_json('/nodes/%s/portgroups/%s/ports' % (
node.uuid, pg.uuid), expect_errors=True,
headers={api_base.Version.string: '1.24'})
self.assertEqual(http_client.FORBIDDEN, response.status_int)
def test_volume_subresource_link(self):
node = obj_utils.create_test_node(self.context)
data = self.get_json(
'/nodes/%s' % node.uuid,
headers={api_base.Version.string: '1.32'})
self.assertIn('volume', data)
def test_volume_subresource(self):
node = obj_utils.create_test_node(self.context)
data = self.get_json('/nodes/%s/volume' % node.uuid,
headers={api_base.Version.string: '1.32'})
self.assertIn('connectors', data)
self.assertIn('targets', data)
self.assertIn('/volume/connectors',
data['connectors'][0]['href'])
self.assertIn('/volume/connectors',
data['connectors'][1]['href'])
self.assertIn('/volume/targets',
data['targets'][0]['href'])
self.assertIn('/volume/targets',
data['targets'][1]['href'])
def test_volume_subresource_invalid_api_version(self):
node = obj_utils.create_test_node(self.context)
response = self.get_json('/nodes/%s/volume' % node.uuid,
headers={api_base.Version.string: '1.31'},
expect_errors=True)
self.assertEqual(http_client.NOT_FOUND, response.status_int)
def test_volume_connectors_subresource(self):
node = obj_utils.create_test_node(self.context)
for id_ in range(2):
obj_utils.create_test_volume_connector(
self.context, node_id=node.id, uuid=uuidutils.generate_uuid(),
connector_id='test-connector_id-%s' % id_)
data = self.get_json(
'/nodes/%s/volume/connectors' % node.uuid,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(2, len(data['connectors']))
self.assertNotIn('next', data)
# Test collection pagination
data = self.get_json(
'/nodes/%s/volume/connectors?limit=1' % node.uuid,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(1, len(data['connectors']))
self.assertIn('next', data)
def test_volume_connectors_subresource_noid(self):
node = obj_utils.create_test_node(self.context)
obj_utils.create_test_volume_connector(self.context, node_id=node.id)
# No node_id specified.
response = self.get_json(
'/nodes/volume/connectors',
expect_errors=True,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(http_client.NOT_FOUND, response.status_int)
def test_volume_connectors_subresource_node_not_found(self):
non_existent_uuid = 'eeeeeeee-cccc-aaaa-bbbb-cccccccccccc'
response = self.get_json(
'/nodes/%s/volume/connectors' % non_existent_uuid,
expect_errors=True,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(http_client.NOT_FOUND, response.status_int)
def test_volume_targets_subresource(self):
node = obj_utils.create_test_node(self.context)
for id_ in range(2):
obj_utils.create_test_volume_target(
self.context, node_id=node.id, uuid=uuidutils.generate_uuid(),
boot_index=id_)
data = self.get_json(
'/nodes/%s/volume/targets' % node.uuid,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(2, len(data['targets']))
self.assertNotIn('next', data)
# Test collection pagination
data = self.get_json(
'/nodes/%s/volume/targets?limit=1' % node.uuid,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(1, len(data['targets']))
self.assertIn('next', data)
def test_volume_targets_subresource_noid(self):
node = obj_utils.create_test_node(self.context)
obj_utils.create_test_volume_target(self.context, node_id=node.id)
# No node_id specified.
response = self.get_json(
'/nodes/volume/targets',
expect_errors=True,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(http_client.NOT_FOUND, response.status_int)
def test_volume_targets_subresource_node_not_found(self):
non_existent_uuid = 'eeeeeeee-cccc-aaaa-bbbb-cccccccccccc'
response = self.get_json(
'/nodes/%s/volume/targets' % non_existent_uuid,
expect_errors=True,
headers={api_base.Version.string: str(api_v1.max_version())})
self.assertEqual(http_client.NOT_FOUND, response.status_int)
@mock.patch.object(timeutils, 'utcnow', autospec=True)
def _test_node_states(self, mock_utcnow, api_version=None):
fake_state = 'fake-state'
fake_error = 'fake-error'
fake_config = '{"foo": "bar"}'
test_time = datetime.datetime(2000, 1, 1, 0, 0)
mock_utcnow.return_value = test_time
node = obj_utils.create_test_node(self.context,
power_state=fake_state,
target_power_state=fake_state,
provision_state=fake_state,
target_provision_state=fake_state,
provision_updated_at=test_time,
raid_config=fake_config,
target_raid_config=fake_config,
last_error=fake_error)
headers = {}
if api_version:
headers = {api_base.Version.string: api_version}
data = self.get_json('/nodes/%s/states' % node.uuid, headers=headers)
self.assertEqual(fake_state, data['power_state'])
self.assertEqual(fake_state, data['target_power_state'])
self.assertEqual(fake_state, data['provision_state'])
self.assertEqual(fake_state, data['target_provision_state'])
prov_up_at = timeutils.parse_isotime(
data['provision_updated_at']).replace(tzinfo=None)
self.assertEqual(test_time, prov_up_at)
self.assertEqual(fake_error, data['last_error'])
self.assertFalse(data['console_enabled'])
return data
def test_node_states(self):
self._test_node_states()
def test_node_states_raid(self):
data = self._test_node_states(api_version="1.12")
self.assertEqual({'foo': 'bar'}, data['raid_config'])
self.assertEqual({'foo': 'bar'}, data['target_raid_config'])
@mock.patch.object(timeutils, 'utcnow', autospec=True)
def test_node_states_by_name(self, mock_utcnow):
fake_state = 'fake-state'
fake_error = 'fake-error'
test_time = datetime.datetime(1971, 3, 9, 0, 0)
mock_utcnow.return_value = test_time
node = obj_utils.create_test_node(self.context,
name='eggs',
power_state=fake_state,
target_power_state=fake_state,
provision_state=fake_state,
target_provision_state=fake_state,
provision_updated_at=test_time,
last_error=fake_error)
data = self.get_json('/nodes/%s/states' % node.name,
headers={api_base.Version.string: "1.5"})
self.assertEqual(fake_state, data['power_state'])
self.assertEqual(fake_state, data['target_power_state'])
self.assertEqual(fake_state, data['provision_state'])
self.assertEqual(fake_state, data['target_provision_state'])
prov_up_at = timeutils.parse_isotime(
data['provision_updated_at']).replace(tzinfo=None)
self.assertEqual(test_time, prov_up_at)
self.assertEqual(fake_error, data['last_error'])
self.assertFalse(data['console_enabled'])
def test_node_by_instance_uuid(self):
node = obj_utils.create_test_node(
self.context,
uuid=uuidutils.generate_uuid(),
instance_uuid=uuidutils.generate_uuid())
instance_uuid = node.instance_uuid
data = self.get_json('/nodes?instance_uuid=%s' % instance_uuid,
headers={api_base.Version.string: "1.5"})
self.assertThat(data['nodes'], matchers.HasLength(1))
self.assertEqual(node['instance_uuid'],
data['nodes'][0]["instance_uuid"])
def test_node_by_instance_uuid_wrong_uuid(self):
obj_utils.create_test_node(
self.context, uuid=uuidutils.generate_uuid(),
instance_uuid=uuidutils.generate_uuid())
wrong_uuid = uuidutils.generate_uuid()
data = self.get_json('/nodes?instance_uuid=%s' % wrong_uuid)
self.assertThat(data['nodes'], matchers.HasLength(0))
def test_node_by_instance_uuid_invalid_uuid(self):
response = self.get_json('/nodes?instance_uuid=fake',
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(http_client.BAD_REQUEST, response.status_code)
def test_associated_nodes_insensitive(self):
associated_nodes = (self
._create_association_test_nodes()['associated'])
data = self.get_json('/nodes?associated=true')
data1 = self.get_json('/nodes?associated=True')
uuids = [n['uuid'] for n in data['nodes']]
uuids1 = [n['uuid'] for n in data1['nodes']]
self.assertEqual(sorted(associated_nodes), sorted(uuids1))
self.assertEqual(sorted(associated_nodes), sorted(uuids))
def test_associated_nodes_error(self):
self._create_association_test_nodes()
response = self.get_json('/nodes?associated=blah', expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(http_client.BAD_REQUEST, response.status_code)
self.assertTrue(response.json['error_message'])
def test_unassociated_nodes_insensitive(self):
unassociated_nodes = (
self._create_association_test_nodes()['unassociated'])
data = self.get_json('/nodes?associated=false')
data1 = self.get_json('/nodes?associated=FALSE')
uuids = [n['uuid'] for n in data['nodes']]
uuids1 = [n['uuid'] for n in data1['nodes']]
self.assertEqual(sorted(unassociated_nodes), sorted(uuids1))
self.assertEqual(sorted(unassociated_nodes), sorted(uuids))
def test_unassociated_nodes_with_limit(self):
unassociated_nodes = (
self._create_association_test_nodes()['unassociated'])
data = self.get_json('/nodes?associated=False&limit=2')
self.assertThat(data['nodes'], matchers.HasLength(2))
self.assertIn(data['nodes'][0]['uuid'], unassociated_nodes)
def test_next_link_with_association(self):
self._create_association_test_nodes()
data = self.get_json('/nodes/?limit=3&associated=True')
self.assertThat(data['nodes'], matchers.HasLength(3))
self.assertIn('associated=True', data['next'])
def test_detail_with_association_filter(self):
associated_nodes = (self
._create_association_test_nodes()['associated'])
data = self.get_json('/nodes/detail?associated=true')
self.assertIn('driver', data['nodes'][0])
self.assertEqual(len(associated_nodes), len(data['nodes']))
def test_next_link_with_association_with_detail(self):
self._create_association_test_nodes()
data = self.get_json('/nodes/detail?limit=3&associated=true')
self.assertThat(data['nodes'], matchers.HasLength(3))
self.assertIn('driver', data['nodes'][0])
self.assertIn('associated=True', data['next'])
def test_detail_with_instance_uuid(self):
node = obj_utils.create_test_node(
self.context,
uuid=uuidutils.generate_uuid(),
instance_uuid=uuidutils.generate_uuid(),
chassis_id=self.chassis.id)
instance_uuid = node.instance_uuid
data = self.get_json('/nodes/detail?instance_uuid=%s' % instance_uuid)
self.assertEqual(node['instance_uuid'],
data['nodes'][0]["instance_uuid"])
self.assertIn('driver', data['nodes'][0])
self.assertIn('driver_info', data['nodes'][0])
self.assertIn('extra', data['nodes'][0])
self.assertIn('properties', data['nodes'][0])
self.assertIn('chassis_uuid', data['nodes'][0])
# never expose the chassis_id
self.assertNotIn('chassis_id', data['nodes'][0])
def test_maintenance_nodes(self):
nodes = []
for id in range(5):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
maintenance=id % 2)
nodes.append(node)
data = self.get_json('/nodes?maintenance=true')
uuids = [n['uuid'] for n in data['nodes']]
test_uuids_1 = [n.uuid for n in nodes if n.maintenance]
self.assertEqual(sorted(test_uuids_1), sorted(uuids))
data = self.get_json('/nodes?maintenance=false')
uuids = [n['uuid'] for n in data['nodes']]
test_uuids_0 = [n.uuid for n in nodes if not n.maintenance]
self.assertEqual(sorted(test_uuids_0), sorted(uuids))
def test_maintenance_nodes_error(self):
response = self.get_json('/nodes?associated=true&maintenance=blah',
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(http_client.BAD_REQUEST, response.status_code)
self.assertTrue(response.json['error_message'])
def test_maintenance_nodes_associated(self):
self._create_association_test_nodes()
node = obj_utils.create_test_node(
self.context,
instance_uuid=uuidutils.generate_uuid(),
maintenance=True)
data = self.get_json('/nodes?associated=true&maintenance=false')
uuids = [n['uuid'] for n in data['nodes']]
self.assertNotIn(node.uuid, uuids)
data = self.get_json('/nodes?associated=true&maintenance=true')
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node.uuid, uuids)
data = self.get_json('/nodes?associated=true&maintenance=TruE')
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node.uuid, uuids)
def test_get_nodes_by_provision_state(self):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
provision_state=states.AVAILABLE)
node1 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
provision_state=states.DEPLOYING)
data = self.get_json('/nodes?provision_state=available',
headers={api_base.Version.string: "1.9"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node.uuid, uuids)
self.assertNotIn(node1.uuid, uuids)
data = self.get_json('/nodes?provision_state=deploying',
headers={api_base.Version.string: "1.9"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node1.uuid, uuids)
self.assertNotIn(node.uuid, uuids)
def test_get_nodes_by_invalid_provision_state(self):
response = self.get_json('/nodes?provision_state=test',
headers={api_base.Version.string: "1.9"},
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(http_client.BAD_REQUEST, response.status_code)
self.assertTrue(response.json['error_message'])
def test_get_nodes_by_provision_state_not_allowed(self):
response = self.get_json('/nodes?provision_state=test',
headers={api_base.Version.string: "1.8"},
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_code)
self.assertTrue(response.json['error_message'])
def test_get_nodes_by_driver(self):
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
driver='ipmi')
node1 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
driver='fake-hardware')
data = self.get_json('/nodes?driver=ipmi',
headers={api_base.Version.string: "1.16"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node.uuid, uuids)
self.assertNotIn(node1.uuid, uuids)
data = self.get_json('/nodes?driver=fake-hardware',
headers={api_base.Version.string: "1.16"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node1.uuid, uuids)
self.assertNotIn(node.uuid, uuids)
def test_get_nodes_by_invalid_driver(self):
data = self.get_json('/nodes?driver=test',
headers={api_base.Version.string: "1.16"})
self.assertEqual(0, len(data['nodes']))
def test_get_nodes_by_driver_invalid_api_version(self):
response = self.get_json(
'/nodes?driver=fake',
headers={api_base.Version.string: str(api_v1.min_version())},
expect_errors=True)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_code)
self.assertTrue(response.json['error_message'])
def _test_get_nodes_by_resource_class(self, detail=False):
if detail:
base_url = '/nodes/detail?resource_class=%s'
else:
base_url = '/nodes?resource_class=%s'
node = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
resource_class='foo')
node1 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
resource_class='bar')
data = self.get_json(base_url % 'foo',
headers={api_base.Version.string: "1.21"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node.uuid, uuids)
self.assertNotIn(node1.uuid, uuids)
data = self.get_json(base_url % 'bar',
headers={api_base.Version.string: "1.21"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node1.uuid, uuids)
self.assertNotIn(node.uuid, uuids)
def test_get_nodes_by_resource_class(self):
self._test_get_nodes_by_resource_class(detail=False)
def test_get_nodes_by_resource_class_detail(self):
self._test_get_nodes_by_resource_class(detail=True)
def _test_get_nodes_by_invalid_resource_class(self, detail=False):
if detail:
base_url = '/nodes/detail?resource_class=%s'
else:
base_url = '/nodes?resource_class=%s'
data = self.get_json(base_url % 'test',
headers={api_base.Version.string: "1.21"})
self.assertEqual(0, len(data['nodes']))
def test_get_nodes_by_invalid_resource_class(self):
self._test_get_nodes_by_invalid_resource_class(detail=False)
def test_get_nodes_by_invalid_resource_class_detail(self):
self._test_get_nodes_by_invalid_resource_class(detail=True)
def _test_get_nodes_by_resource_class_invalid_api_version(self,
detail=False):
if detail:
base_url = '/nodes/detail?resource_class=%s'
else:
base_url = '/nodes?resource_class=%s'
response = self.get_json(
base_url % 'fake',
headers={api_base.Version.string: str(api_v1.min_version())},
expect_errors=True)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_code)
self.assertTrue(response.json['error_message'])
def test_get_nodes_by_resource_class_invalid_api_version(self):
self._test_get_nodes_by_resource_class_invalid_api_version(
detail=False)
def test_get_nodes_by_resource_class_invalid_api_version_detail(self):
self._test_get_nodes_by_resource_class_invalid_api_version(detail=True)
def _test_get_nodes_by_traits_not_allowed(self, detail=False):
if detail:
base_url = '/nodes/detail?traits=%s'
else:
base_url = '/nodes?traits=%s'
response = self.get_json(
base_url % 'CUSTOM_TRAIT_1',
headers={api_base.Version.string: str(api_v1.max_version())},
expect_errors=True)
self.assertEqual(http_client.BAD_REQUEST, response.status_code)
self.assertTrue(response.json['error_message'])
def test_get_nodes_by_traits_not_allowed(self):
self._test_get_nodes_by_traits_not_allowed(detail=False)
def test_get_nodes_by_traits_not_allowed_detail(self):
self._test_get_nodes_by_traits_not_allowed(detail=True)
def test_get_nodes_by_fault(self):
node1 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
fault='power failure')
node2 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
fault="clean failure")
for base_url in ('/nodes', '/nodes/detail'):
data = self.get_json(base_url + '?fault=power failure',
headers={api_base.Version.string: "1.42"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node1.uuid, uuids)
self.assertNotIn(node2.uuid, uuids)
data = self.get_json(base_url + '?fault=clean failure',
headers={api_base.Version.string: "1.42"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node2.uuid, uuids)
self.assertNotIn(node1.uuid, uuids)
def test_get_nodes_by_fault_with_invalid_fault(self):
for url in ('/nodes?fault=somefake',
'/nodes/detail?fault=somefake'):
response = self.get_json(
url, headers={api_base.Version.string: "1.42"},
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(http_client.BAD_REQUEST, response.status_code)
self.assertTrue(response.json['error_message'])
def test_get_nodes_by_fault_not_allowed(self):
for url in ('/nodes?fault=power failure',
'/nodes/detail?fault=power failure'):
response = self.get_json(
url, headers={api_base.Version.string: "1.41"},
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_code)
self.assertTrue(response.json['error_message'])
def test_get_nodes_by_conductor_group(self):
node1 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
conductor_group='group1')
node2 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
conductor_group='group2')
for base_url in ('/nodes', '/nodes/detail'):
data = self.get_json(base_url + '?conductor_group=group1',
headers={api_base.Version.string: "1.46"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node1.uuid, uuids)
self.assertNotIn(node2.uuid, uuids)
data = self.get_json(base_url + '?conductor_group=group2',
headers={api_base.Version.string: "1.46"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node2.uuid, uuids)
self.assertNotIn(node1.uuid, uuids)
def test_get_nodes_by_conductor_group_not_allowed(self):
for url in ('/nodes?conductor_group=group1',
'/nodes/detail?conductor_group=group1'):
response = self.get_json(
url, headers={api_base.Version.string: "1.45"},
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_code)
self.assertTrue(response.json['error_message'])
def test_get_nodes_by_conductor_not_allowed(self):
response = self.get_json('/nodes?conductor=rocky.rocks',
headers={api_base.Version.string: "1.48"},
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_code)
self.assertTrue(response.json['error_message'])
def test_get_nodes_by_conductor(self):
node1 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid())
node2 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid())
response = self.get_json('/nodes?conductor=rocky.rocks',
headers={api_base.Version.string: "1.49"})
uuids = [n['uuid'] for n in response['nodes']]
self.assertFalse(uuids)
response = self.get_json('/nodes?conductor=fake.conductor',
headers={api_base.Version.string: "1.49"})
uuids = [n['uuid'] for n in response['nodes']]
self.assertEqual(2, len(uuids))
self.assertIn(node1.uuid, uuids)
self.assertIn(node2.uuid, uuids)
self.mock_get_conductor_for.side_effect = ['rocky.rocks',
'fake.conductor']
response = self.get_json('/nodes?conductor=fake.conductor',
headers={api_base.Version.string: "1.49"})
uuids = [n['uuid'] for n in response['nodes']]
self.assertEqual(1, len(uuids))
self.assertNotIn(node1.uuid, uuids)
self.assertIn(node2.uuid, uuids)
def test_get_nodes_by_conductor_no_valid_host(self):
obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid())
self.mock_get_conductor_for.side_effect = exception.NoValidHost(
reason='hey a conductor just goes vacation')
response = self.get_json('/nodes?conductor=like.shadows',
headers={api_base.Version.string: "1.49"})
self.assertEqual([], response['nodes'])
self.mock_get_conductor_for.side_effect = exception.TemporaryFailure(
reason='this must be conductor strike')
response = self.get_json('/nodes?conductor=like.shadows',
headers={api_base.Version.string: "1.49"})
self.assertEqual([], response['nodes'])
self.mock_get_conductor_for.side_effect = exception.IronicException(
'Some unexpected thing happened')
response = self.get_json('/nodes?conductor=fake.conductor',
headers={api_base.Version.string: "1.49"},
expect_errors=True)
self.assertIn('Some unexpected thing happened',
response.json['error_message'])
def test_get_nodes_by_owner(self):
node1 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
owner='fred')
node2 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
owner='bob')
for base_url in ('/nodes', '/nodes/detail'):
data = self.get_json(base_url + '?owner=fred',
headers={api_base.Version.string: "1.50"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node1.uuid, uuids)
self.assertNotIn(node2.uuid, uuids)
data = self.get_json(base_url + '?owner=bob',
headers={api_base.Version.string: "1.50"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node2.uuid, uuids)
self.assertNotIn(node1.uuid, uuids)
def test_get_nodes_by_owner_not_allowed(self):
for url in ('/nodes?owner=fred',
'/nodes/detail?owner=fred'):
response = self.get_json(
url, headers={api_base.Version.string: "1.48"},
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(http_client.NOT_ACCEPTABLE, response.status_code)
self.assertTrue(response.json['error_message'])
def test_get_nodes_by_description(self):
node1 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
description='some cats here')
node2 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
description='some dogs there')
data = self.get_json('/nodes?description_contains=cat',
headers={api_base.Version.string: '1.51'})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node1.uuid, uuids)
self.assertNotIn(node2.uuid, uuids)
data = self.get_json('/nodes?description_contains=dog',
headers={api_base.Version.string: '1.51'})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node2.uuid, uuids)
self.assertNotIn(node1.uuid, uuids)
def test_get_nodes_by_lessee(self):
node1 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
lessee='project1')
node2 = obj_utils.create_test_node(self.context,
uuid=uuidutils.generate_uuid(),
lessee='project2')
for base_url in ('/nodes', '/nodes/detail'):
data = self.get_json(base_url + '?lessee=project1',
headers={api_base.Version.string: "1.65"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node1.uuid, uuids)
self.assertNotIn(node2.uuid, uuids)
data = self.get_json(base_url + '?lessee=project2',
headers={api_base.Version.string: "1.65"})
uuids = [n['uuid'] for n in data['nodes']]
self.assertIn(node2.uuid, uuids)
self.assertNotIn(node1.uuid, uuids)
def test_get_nodes_by_lessee_not_allowed(self):
for url in ('/nodes?lessee=project1',
'/nodes/detail?lessee=project1'):
response = self.get_json(
url, headers={api_base.Version.string: "1.64"},