89cd6a0c49
As part of the split of functional and unit tests we need to isolate the unit tests into a separate directory for having multiple test targets in a sane way. Part of bp:functional-tests-for-nova Change-Id: Id42ba373c1bda6a312b673ab2b489ca56da8c628
232 lines
10 KiB
Python
232 lines
10 KiB
Python
# Copyright (c) 2014 VMware, Inc.
|
|
#
|
|
# 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.
|
|
|
|
from nova import exception
|
|
from nova import test
|
|
from nova.virt import diagnostics
|
|
|
|
|
|
class DiagnosticsTests(test.NoDBTestCase):
|
|
|
|
def test_cpu_diagnostics_default(self):
|
|
cpu = diagnostics.CpuDiagnostics()
|
|
self.assertEqual(0, cpu.time)
|
|
|
|
def test_cpu_diagnostics(self):
|
|
cpu = diagnostics.CpuDiagnostics(time=7)
|
|
self.assertEqual(7, cpu.time)
|
|
|
|
def test_nic_diagnostics_default(self):
|
|
nic = diagnostics.NicDiagnostics()
|
|
self.assertEqual('00:00:00:00:00:00', nic.mac_address)
|
|
self.assertEqual(0, nic.rx_octets)
|
|
self.assertEqual(0, nic.rx_errors)
|
|
self.assertEqual(0, nic.rx_drop)
|
|
self.assertEqual(0, nic.rx_packets)
|
|
self.assertEqual(0, nic.tx_octets)
|
|
self.assertEqual(0, nic.tx_errors)
|
|
self.assertEqual(0, nic.tx_drop)
|
|
self.assertEqual(0, nic.tx_packets)
|
|
|
|
def test_nic_diagnostics(self):
|
|
nic = diagnostics.NicDiagnostics(mac_address='00:00:ca:fe:00:00',
|
|
rx_octets=1, rx_errors=2, rx_drop=3, rx_packets=4,
|
|
tx_octets=5, tx_errors=6, tx_drop=7, tx_packets=8)
|
|
self.assertEqual('00:00:ca:fe:00:00', nic.mac_address)
|
|
self.assertEqual(1, nic.rx_octets)
|
|
self.assertEqual(2, nic.rx_errors)
|
|
self.assertEqual(3, nic.rx_drop)
|
|
self.assertEqual(4, nic.rx_packets)
|
|
self.assertEqual(5, nic.tx_octets)
|
|
self.assertEqual(6, nic.tx_errors)
|
|
self.assertEqual(7, nic.tx_drop)
|
|
self.assertEqual(8, nic.tx_packets)
|
|
|
|
def test_disk_diagnostics_default(self):
|
|
disk = diagnostics.DiskDiagnostics()
|
|
self.assertEqual('', disk.id)
|
|
self.assertEqual(0, disk.read_bytes)
|
|
self.assertEqual(0, disk.read_requests)
|
|
self.assertEqual(0, disk.write_bytes)
|
|
self.assertEqual(0, disk.write_requests)
|
|
self.assertEqual(0, disk.errors_count)
|
|
|
|
def test_disk_diagnostics(self):
|
|
disk = diagnostics.DiskDiagnostics(id='fake_disk_id',
|
|
read_bytes=1, read_requests=2,
|
|
write_bytes=3, write_requests=4,
|
|
errors_count=5)
|
|
self.assertEqual('fake_disk_id', disk.id)
|
|
self.assertEqual(1, disk.read_bytes)
|
|
self.assertEqual(2, disk.read_requests)
|
|
self.assertEqual(3, disk.write_bytes)
|
|
self.assertEqual(4, disk.write_requests)
|
|
self.assertEqual(5, disk.errors_count)
|
|
|
|
def test_memory_diagnostics_default(self):
|
|
memory = diagnostics.MemoryDiagnostics()
|
|
self.assertEqual(0, memory.maximum)
|
|
self.assertEqual(0, memory.used)
|
|
|
|
def test_memory_diagnostics(self):
|
|
memory = diagnostics.MemoryDiagnostics(maximum=1, used=2)
|
|
self.assertEqual(1, memory.maximum)
|
|
self.assertEqual(2, memory.used)
|
|
|
|
def test_diagnostics_default(self):
|
|
diags = diagnostics.Diagnostics()
|
|
self.assertIsNone(diags.state)
|
|
self.assertIsNone(diags.driver)
|
|
self.assertIsNone(diags.hypervisor_os)
|
|
self.assertEqual(0, diags.uptime)
|
|
self.assertFalse(diags.config_drive)
|
|
self.assertEqual([], diags.cpu_details)
|
|
self.assertEqual([], diags.nic_details)
|
|
self.assertEqual([], diags.disk_details)
|
|
self.assertEqual(0, diags.memory_details.maximum)
|
|
self.assertEqual(0, diags.memory_details.used)
|
|
self.assertEqual('1.0', diags.version)
|
|
|
|
def test_diagnostics(self):
|
|
cpu_details = [diagnostics.CpuDiagnostics()]
|
|
nic_details = [diagnostics.NicDiagnostics()]
|
|
disk_details = [diagnostics.DiskDiagnostics()]
|
|
diags = diagnostics.Diagnostics(
|
|
state='fake-state', driver='fake-driver',
|
|
hypervisor_os='fake-os',
|
|
uptime=1, cpu_details=cpu_details,
|
|
nic_details=nic_details, disk_details=disk_details,
|
|
config_drive=True)
|
|
self.assertEqual('fake-state', diags.state)
|
|
self.assertEqual('fake-driver', diags.driver)
|
|
self.assertEqual('fake-os', diags.hypervisor_os)
|
|
self.assertEqual(1, diags.uptime)
|
|
self.assertTrue(diags.config_drive)
|
|
self.assertEqual(1, len(diags.cpu_details))
|
|
self.assertEqual(1, len(diags.nic_details))
|
|
self.assertEqual(1, len(diags.disk_details))
|
|
self.assertEqual(0, diags.memory_details.maximum)
|
|
self.assertEqual(0, diags.memory_details.used)
|
|
self.assertEqual('1.0', diags.version)
|
|
|
|
def test_add_cpu(self):
|
|
diags = diagnostics.Diagnostics()
|
|
self.assertEqual([], diags.cpu_details)
|
|
diags.add_cpu(time=7)
|
|
self.assertEqual(1, len(diags.cpu_details))
|
|
self.assertEqual(7, diags.cpu_details[0].time)
|
|
|
|
def test_add_nic(self):
|
|
diags = diagnostics.Diagnostics()
|
|
self.assertEqual([], diags.nic_details)
|
|
diags.add_nic(mac_address='00:00:ca:fe:00:00',
|
|
rx_octets=1, rx_errors=2, rx_drop=3, rx_packets=4,
|
|
tx_octets=5, tx_errors=6, tx_drop=7, tx_packets=8)
|
|
self.assertEqual(1, len(diags.nic_details))
|
|
self.assertEqual('00:00:ca:fe:00:00', diags.nic_details[0].mac_address)
|
|
self.assertEqual(1, diags.nic_details[0].rx_octets)
|
|
self.assertEqual(2, diags.nic_details[0].rx_errors)
|
|
self.assertEqual(3, diags.nic_details[0].rx_drop)
|
|
self.assertEqual(4, diags.nic_details[0].rx_packets)
|
|
self.assertEqual(5, diags.nic_details[0].tx_octets)
|
|
self.assertEqual(6, diags.nic_details[0].tx_errors)
|
|
self.assertEqual(7, diags.nic_details[0].tx_drop)
|
|
self.assertEqual(8, diags.nic_details[0].tx_packets)
|
|
|
|
def test_add_disk(self):
|
|
diags = diagnostics.Diagnostics()
|
|
self.assertEqual([], diags.disk_details)
|
|
diags.add_disk(id='fake_disk_id',
|
|
read_bytes=1, read_requests=2,
|
|
write_bytes=3, write_requests=4,
|
|
errors_count=5)
|
|
self.assertEqual(1, len(diags.disk_details))
|
|
self.assertEqual('fake_disk_id', diags.disk_details[0].id)
|
|
self.assertEqual(1, diags.disk_details[0].read_bytes)
|
|
self.assertEqual(2, diags.disk_details[0].read_requests)
|
|
self.assertEqual(3, diags.disk_details[0].write_bytes)
|
|
self.assertEqual(4, diags.disk_details[0].write_requests)
|
|
self.assertEqual(5, diags.disk_details[0].errors_count)
|
|
|
|
def test_diagnostics_serialize_default(self):
|
|
diags = diagnostics.Diagnostics()
|
|
expected = {'config_drive': False,
|
|
'cpu_details': [],
|
|
'disk_details': [],
|
|
'driver': None,
|
|
'hypervisor_os': None,
|
|
'memory_details': {'maximum': 0, 'used': 0},
|
|
'nic_details': [],
|
|
'state': None,
|
|
'uptime': 0,
|
|
'version': '1.0'}
|
|
result = diags.serialize()
|
|
self.assertEqual(expected, result)
|
|
|
|
def test_diagnostics_serialize(self):
|
|
cpu_details = [diagnostics.CpuDiagnostics()]
|
|
nic_details = [diagnostics.NicDiagnostics()]
|
|
disk_details = [diagnostics.DiskDiagnostics()]
|
|
diags = diagnostics.Diagnostics(
|
|
state='fake-state', driver='fake-driver',
|
|
hypervisor_os='fake-os',
|
|
uptime=1, cpu_details=cpu_details,
|
|
nic_details=nic_details, disk_details=disk_details,
|
|
config_drive=True)
|
|
expected = {'config_drive': True,
|
|
'cpu_details': [{'time': 0}],
|
|
'disk_details': [{'errors_count': 0,
|
|
'id': '',
|
|
'read_bytes': 0,
|
|
'read_requests': 0,
|
|
'write_bytes': 0,
|
|
'write_requests': 0}],
|
|
'driver': 'fake-driver',
|
|
'hypervisor_os': 'fake-os',
|
|
'memory_details': {'maximum': 0, 'used': 0},
|
|
'nic_details': [{'mac_address': '00:00:00:00:00:00',
|
|
'rx_drop': 0,
|
|
'rx_errors': 0,
|
|
'rx_octets': 0,
|
|
'rx_packets': 0,
|
|
'tx_drop': 0,
|
|
'tx_errors': 0,
|
|
'tx_octets': 0,
|
|
'tx_packets': 0}],
|
|
'state': 'fake-state',
|
|
'uptime': 1,
|
|
'version': '1.0'}
|
|
result = diags.serialize()
|
|
self.assertEqual(expected, result)
|
|
|
|
def test_diagnostics_invalid_input(self):
|
|
self.assertRaises(exception.InvalidInput,
|
|
diagnostics.Diagnostics,
|
|
cpu_details='invalid type')
|
|
self.assertRaises(exception.InvalidInput,
|
|
diagnostics.Diagnostics,
|
|
cpu_details=['invalid entry'])
|
|
self.assertRaises(exception.InvalidInput,
|
|
diagnostics.Diagnostics,
|
|
nic_details='invalid type')
|
|
self.assertRaises(exception.InvalidInput,
|
|
diagnostics.Diagnostics,
|
|
nic_details=['invalid entry'])
|
|
self.assertRaises(exception.InvalidInput,
|
|
diagnostics.Diagnostics,
|
|
disk_details='invalid type')
|
|
self.assertRaises(exception.InvalidInput,
|
|
diagnostics.Diagnostics,
|
|
disk_details=['invalid entry'])
|