270 lines
9.6 KiB
Python
270 lines
9.6 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright 2011 University of Southern California
|
|
# 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.
|
|
|
|
import __builtin__
|
|
|
|
import functools
|
|
import mox
|
|
import StringIO
|
|
|
|
from nova.compute import power_state
|
|
from nova import exception
|
|
from nova import flags
|
|
from nova.openstack.common import jsonutils
|
|
from nova import test
|
|
from nova.tests import fake_utils
|
|
|
|
from nova.virt.baremetal import dom
|
|
from nova.virt.baremetal import driver
|
|
|
|
|
|
FLAGS = flags.FLAGS
|
|
|
|
|
|
# Same fake_domains is used by different classes,
|
|
# but different fake_file is used by different classes for unit test.
|
|
fake_domains = [{'status': 1, 'name': 'instance-00000001',
|
|
'memory_kb': 16777216, 'kernel_id': '1896115634',
|
|
'ramdisk_id': '', 'image_id': '1552326678',
|
|
'vcpus': 1, 'node_id': 6,
|
|
'mac_address': '02:16:3e:01:4e:c9',
|
|
'ip_address': '10.5.1.2'}]
|
|
|
|
|
|
class DomainReadWriteTestCase(test.TestCase):
|
|
|
|
def setUp(self):
|
|
super(DomainReadWriteTestCase, self).setUp()
|
|
self.flags(baremetal_driver='fake')
|
|
|
|
def test_read_domain_with_empty_list(self):
|
|
"""Read a file that contains no domains"""
|
|
|
|
self.mox.StubOutWithMock(__builtin__, 'open')
|
|
fake_file = StringIO.StringIO('[]')
|
|
open('/tftpboot/test_fake_dom_file', 'r').AndReturn(fake_file)
|
|
|
|
self.mox.ReplayAll()
|
|
|
|
domains = dom.read_domains('/tftpboot/test_fake_dom_file')
|
|
|
|
self.assertEqual(domains, [])
|
|
|
|
def test_read_domain(self):
|
|
"""Read a file that contains at least one domain"""
|
|
fake_file = StringIO.StringIO('''[{"status": 1,
|
|
"image_id": "1552326678", "vcpus": 1, "node_id": 6,
|
|
"name": "instance-00000001", "memory_kb": 16777216,
|
|
"mac_address": "02:16:3e:01:4e:c9", "kernel_id": "1896115634",
|
|
"ramdisk_id": "", "ip_address": "10.5.1.2"}]''')
|
|
|
|
self.mox.StubOutWithMock(__builtin__, 'open')
|
|
open('/tftpboot/test_fake_dom_file', 'r').AndReturn(fake_file)
|
|
|
|
self.mox.ReplayAll()
|
|
|
|
domains = dom.read_domains('/tftpboot/test_fake_dom_file')
|
|
|
|
self.assertEqual(domains, fake_domains)
|
|
|
|
def test_read_no_file(self):
|
|
"""Try to read when the file does not exist
|
|
|
|
This should through and IO exception"""
|
|
|
|
self.mox.StubOutWithMock(__builtin__, 'open')
|
|
open('/tftpboot/test_fake_dom_file',
|
|
'r').AndRaise(IOError(2, 'No such file or directory',
|
|
'/tftpboot/test_fake_dom_file'))
|
|
|
|
self.mox.ReplayAll()
|
|
|
|
self.assertRaises(exception.NotFound, dom.read_domains,
|
|
'/tftpboot/test_fake_dom_file')
|
|
|
|
def assertJSONEquals(self, x, y):
|
|
"""Check if two json strings represent the equivalent Python object"""
|
|
self.assertEquals(jsonutils.loads(x), jsonutils.loads(y))
|
|
return jsonutils.loads(x) == jsonutils.loads(y)
|
|
|
|
def test_write_domain(self):
|
|
"""Write the domain to file"""
|
|
self.mox.StubOutWithMock(__builtin__, 'open')
|
|
mock_file = self.mox.CreateMock(file)
|
|
expected_json = '''[{"status": 1,
|
|
"image_id": "1552326678", "vcpus": 1, "node_id": 6,
|
|
"name": "instance-00000001", "memory_kb": 16777216,
|
|
"mac_address": "02:16:3e:01:4e:c9", "kernel_id": "1896115634",
|
|
"ramdisk_id": "", "ip_address": "10.5.1.2"}]'''
|
|
open('/tftpboot/test_fake_dom_file', 'w').AndReturn(mock_file)
|
|
|
|
# Check if the argument to file.write() represents the same
|
|
# Python object as expected_json
|
|
# We can't do an exact string comparison
|
|
# because of ordering and whitespace
|
|
mock_file.write(mox.Func(functools.partial(self.assertJSONEquals,
|
|
expected_json)))
|
|
mock_file.close()
|
|
|
|
self.mox.ReplayAll()
|
|
|
|
dom.write_domains('/tftpboot/test_fake_dom_file', fake_domains)
|
|
|
|
|
|
class BareMetalDomTestCase(test.TestCase):
|
|
|
|
def setUp(self):
|
|
super(BareMetalDomTestCase, self).setUp()
|
|
self.flags(baremetal_driver='fake')
|
|
# Stub out utils.execute
|
|
fake_utils.stub_out_utils_execute(self.stubs)
|
|
|
|
def tearDown(self):
|
|
super(BareMetalDomTestCase, self).tearDown()
|
|
|
|
# Reset the singleton state
|
|
dom.BareMetalDom._instance = None
|
|
dom.BareMetalDom._is_init = False
|
|
|
|
def test_read_domain_only_once(self):
|
|
"""Confirm that the domain is read from a file only once,
|
|
even if the object is instantiated multiple times"""
|
|
self.mox.StubOutWithMock(dom, 'read_domains')
|
|
self.mox.StubOutWithMock(dom, 'write_domains')
|
|
|
|
dom.read_domains('/tftpboot/test_fake_dom_file').AndReturn([])
|
|
dom.write_domains('/tftpboot/test_fake_dom_file', [])
|
|
|
|
self.mox.ReplayAll()
|
|
|
|
# Instantiate multiple instances
|
|
x = dom.BareMetalDom()
|
|
x = dom.BareMetalDom()
|
|
x = dom.BareMetalDom()
|
|
|
|
def test_init_no_domains(self):
|
|
|
|
# Create the mock objects
|
|
self.mox.StubOutWithMock(dom, 'read_domains')
|
|
self.mox.StubOutWithMock(dom, 'write_domains')
|
|
|
|
dom.read_domains('/tftpboot/test_fake_dom_file').AndReturn([])
|
|
dom.write_domains('/tftpboot/test_fake_dom_file', [])
|
|
|
|
self.mox.ReplayAll()
|
|
|
|
# Code under test
|
|
bmdom = dom.BareMetalDom()
|
|
|
|
# Expectd values
|
|
self.assertEqual(bmdom.fake_dom_nums, 0)
|
|
|
|
def test_init_remove_non_running_domain(self):
|
|
"""Check to see that all entries in the domain list are removed
|
|
except for the one that is in the running state"""
|
|
|
|
domains = [dict(node_id=1, name='i-00000001',
|
|
status=power_state.NOSTATE),
|
|
dict(node_id=2, name='i-00000002', status=power_state.RUNNING),
|
|
dict(node_id=3, name='i-00000003', status=power_state.PAUSED),
|
|
dict(node_id=5, name='i-00000004', status=power_state.SHUTDOWN),
|
|
dict(node_id=7, name='i-00000005', status=power_state.CRASHED),
|
|
dict(node_id=8, name='i-00000006', status=power_state.SUSPENDED),
|
|
dict(node_id=9, name='i-00000007', status=power_state.NOSTATE)]
|
|
|
|
# Create the mock objects
|
|
self.mox.StubOutWithMock(dom, 'read_domains')
|
|
self.mox.StubOutWithMock(dom, 'write_domains')
|
|
dom.read_domains('/tftpboot/test_fake_dom_file').AndReturn(domains)
|
|
dom.write_domains('/tftpboot/test_fake_dom_file', domains)
|
|
|
|
self.mox.ReplayAll()
|
|
|
|
# Code under test
|
|
bmdom = dom.BareMetalDom()
|
|
|
|
self.assertEqual(bmdom.domains, [{'node_id': 2,
|
|
'name': 'i-00000002',
|
|
'status': power_state.RUNNING}])
|
|
self.assertEqual(bmdom.fake_dom_nums, 1)
|
|
|
|
def test_find_domain(self):
|
|
domain = {'status': 1, 'name': 'instance-00000001',
|
|
'memory_kb': 16777216, 'kernel_id': '1896115634',
|
|
'ramdisk_id': '', 'image_id': '1552326678',
|
|
'vcpus': 1, 'node_id': 6,
|
|
'mac_address': '02:16:3e:01:4e:c9',
|
|
'ip_address': '10.5.1.2'}
|
|
|
|
# Create the mock objects
|
|
self.mox.StubOutWithMock(dom, 'read_domains')
|
|
self.mox.StubOutWithMock(dom, 'write_domains')
|
|
|
|
# Expected calls
|
|
dom.read_domains('/tftpboot/'
|
|
'test_fake_dom_file').AndReturn(fake_domains)
|
|
dom.write_domains('/tftpboot/test_fake_dom_file', fake_domains)
|
|
|
|
self.mox.ReplayAll()
|
|
|
|
# Code under test
|
|
bmdom = dom.BareMetalDom()
|
|
|
|
# Expected values
|
|
self.assertEquals(bmdom.find_domain('instance-00000001'), domain)
|
|
|
|
|
|
class BareMetalTestCase(test.TestCase):
|
|
|
|
test_ip = '10.11.12.13'
|
|
test_instance = {'memory_kb': '1024000',
|
|
'basepath': '/some/path',
|
|
'bridge_name': 'br100',
|
|
'mac_address': '02:12:34:46:56:67',
|
|
'vcpus': 2,
|
|
'project_id': 'fake',
|
|
'bridge': 'br101',
|
|
'image_ref': '123456',
|
|
'instance_type_id': '5'} # m1.small
|
|
|
|
def setUp(self):
|
|
super(BareMetalTestCase, self).setUp()
|
|
self.flags(baremetal_driver='fake')
|
|
fake_utils.stub_out_utils_execute(self.stubs)
|
|
|
|
def test_get_info(self):
|
|
# Create the mock objects
|
|
self.mox.StubOutWithMock(dom, 'read_domains')
|
|
self.mox.StubOutWithMock(dom, 'write_domains')
|
|
|
|
# Expected calls
|
|
dom.read_domains('/tftpboot/'
|
|
'test_fake_dom_file').AndReturn(fake_domains)
|
|
dom.write_domains('/tftpboot/test_fake_dom_file', fake_domains)
|
|
|
|
self.mox.ReplayAll()
|
|
|
|
# Code under test
|
|
conn = driver.BareMetalDriver(None, True)
|
|
# TODO(mikalstill): this is not a very good fake instance
|
|
info = conn.get_info({'name': 'instance-00000001'})
|
|
|
|
# Expected values
|
|
self.assertEquals(info['mem'], 16777216)
|
|
self.assertEquals(info['state'], 1)
|
|
self.assertEquals(info['num_cpu'], 1)
|
|
self.assertEquals(info['cpu_time'], 100)
|
|
self.assertEquals(info['max_mem'], 16777216)
|