x/compass-core
Code Issues Proposed changes

finish test_host.py, fix bugs in host.py

Browse Source 
Change-Id: Iebbeb3f59cb0de9b00c770005cf9c1158ad204c6
This commit is contained in:
Lei Lei 2014-10-01 12:00:09 -07:00
parent 5f1717cac8
commit f1f3860ee7
2 changed files with 970 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
compass/db/api/host.py
View File

@ -438,7 +438,7 @@ def patch_host_config(session, updater, host_id, **kwargs):
)
return patch_config_internal(
session, updater, host, **kwargs
host, **kwargs
)
@ -842,7 +842,7 @@ def poweron_host(
"""power on host."""
from compass.tasks import client as celery_client
host = utils.get_db_object(
session, models.host, id=host_id
session, models.Host, id=host_id
)
is_host_validated(session, host)
celery_client.celery.send_task(
@ -870,7 +870,7 @@ def poweroff_host(
"""power off host."""
from compass.tasks import client as celery_client
host = utils.get_db_object(
session, models.host, id=host_id
session, models.Host, id=host_id
)
is_host_validated(session, host)
celery_client.celery.send_task(
@ -898,7 +898,7 @@ def reset_host(
"""reset host."""
from compass.tasks import client as celery_client
host = utils.get_db_object(
session, models.host, id=host_id
session, models.Host, id=host_id
)
is_host_validated(session, host)
celery_client.celery.send_task(

966
compass/tests/db/api/test_host.py
View File

@ -15,6 +15,7 @@
import datetime
import logging
import mock
import os
import sys
import unittest2
@ -164,6 +165,66 @@ class HostTestCase(unittest2.TestCase):
subnet_id=self.subnet_ids[1],
is_promiscuous=True
)
# add log history
filenames = ['log1', 'log2']
for filename in filenames:
host.add_host_log_history(
self.user_object,
self.host_ids[0],
filename=filename
)
self.os_configs = {
'general': {
'language': 'EN',
'timezone': 'UTC',
'http_proxy': 'http://127.0.0.1:3128',
'https_proxy': 'http://127.0.0.1:3128',
'no_proxy': [
'127.0.0.1',
'compass'
],
'ntp_server': '127.0.0.1',
'dns_servers': [
'127.0.0.1'
],
'domain': 'ods.com',
'search_path': [
'ods.com'
],
'default_gateway': '127.0.0.1',
},
'server_credentials': {
'username': 'root',
'password': 'root',
},
'partition': {
'/var': {
'max_size': '100G',
'percentage': 10,
'size': '1G'
}
}
}
self.package_configs = {
'security': {
'service_credentials': {
'$service': {
'username': 'root',
'password': 'root'
}
},
'console_credentials': {
'$console': {
'username': 'root',
'password': 'root'
}
}
},
'network_mapping': {
'$interface_type': 'eth0'
}
}
class TestListHosts(HostTestCase):
@ -347,6 +408,911 @@ class TestUpdateHost(HostTestCase):
)
class TestUpdateHosts(HostTestCase):
"""Test update hosts."""
def setUp(self):
super(TestUpdateHosts, self).setUp()
def tearDown(self):
super(TestUpdateHosts, self).tearDown()
def test_update_hosts(self):
update_hosts = host.update_hosts(
self.user_object,
data=[
{
'host_id': self.host_ids[0],
'name': 'test_update1'
},
{
'host_id': self.host_ids[1],
'name': 'test_update2'
}
]
)
results = []
for update_host in update_hosts:
results.append(update_host['name'])
for result in results:
self.assertIn(result, ['test_update1', 'test_update2'])
class TestDelHost(HostTestCase):
"""Test delete host."""
def setUp(self):
super(TestDelHost, self).setUp()
def tearDown(self):
super(TestDelHost, self).tearDown()
def test_del_host(self):
host.del_host(
self.user_object,
self.host_ids[0]
)
del_host = host.list_hosts(
self.user_object
)
ids = []
for item in del_host:
ids.append(item['id'])
self.assertNotIn(self.host_ids[0], ids)
def test_is_host_editable(self):
host.update_host_state(
self.user_object,
self.host_ids[0],
state='INSTALLING'
)
self.assertRaises(
exception.Forbidden,
host.del_host,
self.user_object,
self.host_ids[0]
)
class TestGetHostConfig(HostTestCase):
"""Test get host config."""
def setUp(self):
super(TestGetHostConfig, self).setUp()
host.update_host_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
def tearDown(self):
super(TestGetHostConfig, self).tearDown()
def test_get_host_config(self):
os_configs = host.get_host_config(
self.user_object,
self.host_ids[0]
)
self.assertItemsEqual(self.os_configs, os_configs['os_config'])
class TestGetHostDeployedConfig(HostTestCase):
"""Test get host deployed config."""
def setUp(self):
super(TestGetHostDeployedConfig, self).setUp()
host.update_host_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
cluster.update_cluster_config(
self.user_object,
self.cluster_id,
os_config=self.os_configs,
package_config=self.package_configs
)
cluster.update_cluster_host(
self.user_object,
self.cluster_id,
self.host_ids[0],
roles=['allinone-compute']
)
cluster.review_cluster(
self.user_object,
self.cluster_id,
review={
'hosts': [self.host_ids[0]]
}
)
host.update_host_deployed_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
def tearDown(self):
super(TestGetHostDeployedConfig, self).tearDown()
def test_get_host_deployed_config(self):
os_configs = host.get_host_deployed_config(
self.user_object,
self.host_ids[0]
)
self.assertItemsEqual(
os_configs['deployed_os_config'],
self.os_configs
)
class TestUpdateHostDeployedConfig(HostTestCase):
"""Test update host deployed config."""
def setUp(self):
super(TestUpdateHostDeployedConfig, self).setUp()
host.update_host_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
cluster.update_cluster_config(
self.user_object,
self.cluster_id,
os_config=self.os_configs,
package_config=self.package_configs
)
cluster.update_cluster_host(
self.user_object,
self.cluster_id,
self.host_ids[0],
roles=['allinone-compute']
)
cluster.review_cluster(
self.user_object,
self.cluster_id,
review={
'hosts': [self.host_ids[0]]
}
)
def tearDown(self):
super(TestUpdateHostDeployedConfig, self).tearDown()
def test_update_host_deployed_config(self):
host.update_host_deployed_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
os_configs = host.get_host_deployed_config(
self.user_object,
self.host_ids[0]
)
self.assertItemsEqual(
os_configs['deployed_os_config'],
self.os_configs
)
def test_is_host_editable(self):
host.update_host_state(
self.user_object,
self.host_ids[0],
state='INSTALLING'
)
self.assertRaises(
exception.Forbidden,
host.update_host_deployed_config,
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
class TestUpdateHostConfig(HostTestCase):
"""Test update host config."""
def setUp(self):
super(TestUpdateHostConfig, self).setUp()
def tearDown(self):
super(TestUpdateHostConfig, self).tearDown()
def test_update_host_config(self):
host.update_host_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
os_configs = host.get_host_config(
self.user_object,
self.host_ids[0]
)
self.assertItemsEqual(self.os_configs, os_configs['os_config'])
def test_is_host_editable(self):
host.update_host_state(
self.user_object,
self.host_ids[0],
state='INSTALLING'
)
self.assertRaises(
exception.Forbidden,
host.update_host_config,
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
class TestPatchHostConfig(HostTestCase):
"""Test patch host config."""
def setUp(self):
super(TestPatchHostConfig, self).setUp()
def tearDown(self):
super(TestPatchHostConfig, self).tearDown()
def test_patch_host_config(self):
host.patch_host_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
os_configs = host.get_host_config(
self.user_object,
self.host_ids[0]
)
self.assertItemsEqual(self.os_configs, os_configs['os_config'])
def test_is_host_editable(self):
host.update_host_state(
self.user_object,
self.host_ids[0],
state='INSTALLING'
)
self.assertRaises(
exception.Forbidden,
host.patch_host_config,
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
class TestDelHostConfig(HostTestCase):
"""Test delete host config."""
def setUp(self):
super(TestDelHostConfig, self).setUp()
host.update_host_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
def tearDown(self):
super(TestDelHostConfig, self).tearDown()
def test_del_host_config(self):
host.del_host_config(
self.user_object,
self.host_ids[0]
)
os_configs = host.get_host_config(
self.user_object,
self.host_ids[0]
)
self.assertEqual(os_configs['os_config'], {})
def test_is_host_editable(self):
host.update_host_state(
self.user_object,
self.host_ids[0],
state='INSTALLING'
)
self.assertRaises(
exception.Forbidden,
host.del_host_config,
self.user_object,
self.host_ids[0]
)
class TestListHostNetworks(HostTestCase):
"""Test list host networks."""
def setUp(self):
super(TestListHostNetworks, self).setUp()
host.add_host_network(
self.user_object,
self.host_ids[0],
interface='eth1',
ip='10.145.88.10',
subnet_id=self.subnet_ids[0],
is_promiscuous=True
)
def tearDown(self):
super(TestListHostNetworks, self).tearDown()
def test_list_host_networs(self):
host_networks = host.list_host_networks(
self.user_object,
self.host_ids[0]
)
results = []
for host_network in host_networks:
results.append(host_network['ip'])
for result in results:
self.assertIn(result, ['10.145.88.10', '10.145.88.0'])
class TestListHostnetworks(HostTestCase):
"""Test list hostnetworks."""
def setUp(self):
super(TestListHostnetworks, self).setUp()
def tearDown(self):
super(TestListHostnetworks, self).tearDown()
def test_list_hostnetworks(self):
host_networks = host.list_hostnetworks(
self.user_object,
)
results = []
for host_network in host_networks:
results.append(host_network['ip'])
for result in results:
self.assertIn(result, ['10.145.88.0', '192.168.100.0'])
class TestGetHostNetwork(HostTestCase):
"""Test get host network."""
def setUp(self):
super(TestGetHostNetwork, self).setUp()
def tearDown(self):
super(TestGetHostNetwork, self).tearDown()
def test_get_host_network(self):
host_network = host.get_host_network(
self.user_object,
self.host_ids[0],
self.host_ids[0]
)
self.assertEqual(host_network['ip'], '10.145.88.0')
def test_record_not_exists(self):
self.assertRaises(
exception.RecordNotExists,
host.get_host_network,
self.user_object,
2,
self.host_ids[0]
)
class TestGetHostnetwork(HostTestCase):
"""Test get hostnetwork."""
def setUp(self):
super(TestGetHostnetwork, self).setUp()
def tearDown(self):
super(TestGetHostnetwork, self).tearDown()
def test_get_hostnetwork(self):
host_network = host.get_hostnetwork(
self.user_object,
self.host_ids[0]
)
self.assertEqual(host_network['ip'], '10.145.88.0')
class TestAddHostNetwork(HostTestCase):
"""Test add host network."""
def setUp(self):
super(TestAddHostNetwork, self).setUp()
def tearDown(self):
super(TestAddHostNetwork, self).tearDown()
def test_add_host_network(self):
host.add_host_network(
self.user_object,
self.host_ids[0],
interface='eth1',
ip='10.145.88.20',
subnet_id=self.subnet_ids[0],
is_mgmt=True
)
host_network = host.list_host_networks(
self.user_object,
self.host_ids[0]
)
result = []
for item in host_network:
result.append(item['ip'])
self.assertIn('10.145.88.20', result)
def test_invalid_parameter(self):
self.assertRaises(
exception.InvalidParameter,
host.add_host_network,
self.user_object,
self.host_ids[0],
interface='eth3',
ip='10.145.88.0',
subnet_id=self.subnet_ids[0]
)
class TestAddHostNetworks(HostTestCase):
"""Test add host networks."""
def setUp(self):
super(TestAddHostNetworks, self).setUp()
def tearDown(self):
super(TestAddHostNetworks, self).tearDown()
def test_addhost_networks(self):
host_networks = host.add_host_networks(
self.user_object,
data=[
{
'host_id': self.host_ids[0],
'networks': [
{
'interface': 'eth2',
'ip': '10.145.88.20',
'subnet_id': self.subnet_ids[0],
'is_mgmt': True
},
{
'interface': 'eth3',
'ip': '10.145.88.0',
'subnet_id': self.subnet_ids[0],
'is_mgmt': True
}
]
}
]
)
ip = []
for host_network in host_networks['hosts']:
for item in host_network['networks']:
ip.append(item['ip'])
fail_ip = []
for fail_host in host_networks['failed_hosts']:
for item in fail_host['networks']:
fail_ip.append(item['ip'])
self.assertIn('10.145.88.20', ip)
self.assertIn('10.145.88.0', fail_ip)
class TestUpdateHostNetwork(HostTestCase):
"""Test update host network."""
def setUp(self):
super(TestUpdateHostNetwork, self).setUp()
def tearDown(self):
super(TestUpdateHostNetwork, self).tearDown()
def test_update_host_network(self):
host.update_host_network(
self.user_object,
self.host_ids[0],
self.host_ids[0],
interface='eth10',
ip='10.145.88.100'
)
host_networks = host.list_host_networks(
self.user_object,
self.host_ids[0]
)
interface = None
ip = None
for host_network in host_networks:
interface = host_network['interface']
ip = host_network['ip']
self.assertEqual(interface, 'eth10')
self.assertEqual(ip, '10.145.88.100')
def test_record_not_exists(self):
self.assertRaises(
exception.RecordNotExists,
host.update_host_network,
self.user_object,
self.host_ids[0],
2
)
class TestUpdateHostnetwork(HostTestCase):
"""Test update hostnetwork."""
def setUp(self):
super(TestUpdateHostnetwork, self).setUp()
def tearDown(self):
super(TestUpdateHostnetwork, self).tearDown()
def test_update_hostnetwork(self):
host.update_hostnetwork(
self.user_object,
self.host_ids[0],
interface='eth10',
ip='10.145.88.100'
)
host_networks = host.list_host_networks(
self.user_object,
self.host_ids[0]
)
interface = None
ip = None
for host_network in host_networks:
interface = host_network['interface']
ip = host_network['ip']
self.assertEqual(interface, 'eth10')
self.assertEqual(ip, '10.145.88.100')
def test_invalid_parameter(self):
host.add_host_network(
self.user_object,
self.host_ids[0],
interface='eth11',
ip='10.145.88.101',
subnet_id=self.subnet_ids[0],
is_promiscuous=True
)
self.assertRaises(
exception.InvalidParameter,
host.update_hostnetwork,
self.user_object,
self.host_ids[0],
interface='eth11'
)
self.assertRaises(
exception.InvalidParameter,
host.update_hostnetwork,
self.user_object,
self.host_ids[0],
ip='10.145.88.101'
)
class TestDelHostNetwork(HostTestCase):
"""Test delete host network."""
def setUp(self):
super(TestDelHostNetwork, self).setUp()
def tearDown(self):
super(TestDelHostNetwork, self).tearDown()
def test_del_host_network(self):
host.del_host_network(
self.user_object,
self.host_ids[0],
self.host_ids[0]
)
host_network = host.list_host_networks(
self.user_object,
self.host_ids[0]
)
self.assertEqual(host_network, [])
def test_record_not_exists(self):
self.assertRaises(
exception.RecordNotExists,
host.del_host_network,
self.user_object,
100,
self.host_ids[0]
)
class TestDelHostnetwork(HostTestCase):
"""Test delete hostnetwork."""
def setUp(self):
super(TestDelHostnetwork, self).setUp()
def tearDown(self):
super(TestDelHostnetwork, self).tearDown()
def test_del_hostnetwork(self):
host.del_hostnetwork(
self.user_object,
self.host_ids[0]
)
host_network = host.list_host_networks(
self.user_object,
self.host_ids[0]
)
self.assertEqual(host_network, [])
class TestGetHostState(HostTestCase):
"""Test get host state."""
def setUp(self):
super(TestGetHostState, self).setUp()
def tearDown(self):
super(TestGetHostState, self).tearDown()
def test_get_host_state(self):
host_states = host.get_host_state(
self.user_object,
self.host_ids[0]
)
self.assertEqual(host_states['state'], 'UNINITIALIZED')
class TestUpdateHostState(HostTestCase):
"""Test update host state."""
def setUp(self):
super(TestUpdateHostState, self).setUp()
def tearDown(self):
super(TestUpdateHostState, self).tearDown()
def test_update_host_state(self):
host.update_host_state(
self.user_object,
self.host_ids[0],
state='INSTALLING'
)
host_states = host.get_host_state(
self.user_object,
self.host_ids[0]
)
self.assertEqual(host_states['state'], 'INSTALLING')
class TestGetHostLogHistories(HostTestCase):
"""Test get host log histories."""
def setUp(self):
super(TestGetHostLogHistories, self).setUp()
def tearDown(self):
super(TestGetHostLogHistories, self).tearDown()
def test_get_host_log_histories(self):
logs = host.get_host_log_histories(
self.user_object,
self.host_ids[0]
)
filenames = []
for log in logs:
filenames.append(log['filename'])
for filename in filenames:
self.assertIn(filename, ['log1', 'log2'])
class TestGetHostLogHistory(HostTestCase):
"""Test get host log history."""
def setUp(self):
super(TestGetHostLogHistory, self).setUp()
def tearDown(self):
super(TestGetHostLogHistory, self).tearDown()
def test_get_host_log_history(self):
log = host.get_host_log_history(
self.user_object,
self.host_ids[0],
'log1'
)
self.assertEqual(log['filename'], 'log1')
class TestUpdateHostLogHistory(HostTestCase):
"""Test update host log history."""
def setUp(self):
super(TestUpdateHostLogHistory, self).setUp()
def tearDown(self):
super(TestUpdateHostLogHistory, self).tearDown()
def test_update_host_log_history(self):
host.update_host_log_history(
self.user_object,
self.host_ids[0],
'log1',
severity='WARNING',
message='update log'
)
logs = host.get_host_log_histories(
self.user_object,
self.host_ids[0]
)
result = []
for log in logs:
result.append(log['severity'])
result.append(log['message'])
expects = ['WARNING', 'update log']
for expect in expects:
self.assertIn(expect, result)
class TestAddHostLogHistory(HostTestCase):
"""Test add host log history."""
def setUp(self):
super(TestAddHostLogHistory, self).setUp()
def tearDown(self):
super(TestAddHostLogHistory, self).tearDown()
def test_add_host_log_history(self):
host.add_host_log_history(
self.user_object,
self.host_ids[0],
filename='add_log'
)
logs = host.get_host_log_histories(
self.user_object,
self.host_ids[0]
)
result = []
for log in logs:
result.append(log['filename'])
self.assertIn('add_log', result)
class TestPoweronHost(HostTestCase):
"""Test poweron host."""
def setUp(self):
super(TestPoweronHost, self).setUp()
host.update_host_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
cluster.update_cluster_config(
self.user_object,
self.cluster_id,
os_config=self.os_configs,
package_config=self.package_configs
)
cluster.update_cluster_host(
self.user_object,
self.cluster_id,
self.host_ids[0],
roles=['allinone-compute']
)
cluster.review_cluster(
self.user_object,
self.cluster_id,
review={
'hosts': [self.host_ids[0]]
}
)
def tearDown(self):
super(TestPoweronHost, self).tearDown()
def test_poweron_host(self):
from compass.tasks import client as celery_client
celery_client.celery.send_task = mock.Mock()
poweron_host = host.poweron_host(
self.user_object,
self.host_ids[0],
poweron={'poweron': True}
)
self.assertEqual(
poweron_host['status'],
'poweron newname1 action sent'
)
class TestPoweroffHost(HostTestCase):
"""Test poweroff host."""
def setUp(self):
super(TestPoweroffHost, self).setUp()
host.update_host_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
cluster.update_cluster_config(
self.user_object,
self.cluster_id,
os_config=self.os_configs,
package_config=self.package_configs
)
cluster.update_cluster_host(
self.user_object,
self.cluster_id,
self.host_ids[0],
roles=['allinone-compute']
)
cluster.review_cluster(
self.user_object,
self.cluster_id,
review={
'hosts': [self.host_ids[0]]
}
)
def tearDown(self):
super(TestPoweroffHost, self).tearDown()
def test_poweroff_host(self):
from compass.tasks import client as celery_client
celery_client.celery.send_task = mock.Mock()
poweroff_host = host.poweroff_host(
self.user_object,
self.host_ids[0],
poweroff={'poweroff': True}
)
self.assertEqual(
poweroff_host['status'],
'poweroff newname1 action sent'
)
class TestResetHost(HostTestCase):
"""Test reset host."""
def setUp(self):
super(TestResetHost, self).setUp()
host.update_host_config(
self.user_object,
self.host_ids[0],
os_config=self.os_configs
)
cluster.update_cluster_config(
self.user_object,
self.cluster_id,
os_config=self.os_configs,
package_config=self.package_configs
)
cluster.update_cluster_host(
self.user_object,
self.cluster_id,
self.host_ids[0],
roles=['allinone-compute']
)
cluster.review_cluster(
self.user_object,
self.cluster_id,
review={
'hosts': [self.host_ids[0]]
}
)
def tearDown(self):
super(TestResetHost, self).tearDown()
def test_reset_host(self):
from compass.tasks import client as celery_client
celery_client.celery.send_task = mock.Mock()
reset_host = host.reset_host(
self.user_object,
self.host_ids[0],
reset={'reset': True}
)
self.assertEqual(
reset_host['status'],
'reset newname1 action sent'
)
if __name__ == '__main__':
flags.init()
logsetting.init()