#!/usr/bin/python
#
# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import unittest
import common
from autotest_lib.client.common_lib.test_utils import unittest
from autotest_lib.frontend import setup_django_environment
from autotest_lib.frontend.afe import frontend_test_utils
from autotest_lib.frontend.afe import rdb_model_extensions as rdb_models
from autotest_lib.scheduler import rdb_hosts
from autotest_lib.scheduler import rdb_testing_utils
from autotest_lib.scheduler import rdb_utils
class RDBHostTests(unittest.TestCase, frontend_test_utils.FrontendTestMixin):
"""Unittests for RDBHost objects."""
def setUp(self):
self.db_helper = rdb_testing_utils.DBHelper()
self._database = self.db_helper.database
# Runs syncdb setting up initial database conditions
self._frontend_common_setup()
def tearDown(self):
self._database.disconnect()
self._frontend_common_teardown()
def testWireFormat(self):
"""Test that we can create a client host with the server host's fields.
Get the wire_format fields of an RDBServerHostWrapper and use them to
create an RDBClientHostWrapper.
@raises AssertionError: If the labels and acls don't match up after
going through the complete wire_format conversion, of the bare
wire_format conversion also converts labels and acls.
@raises RDBException: If some critical fields were lost during
wire_format conversion, as we won't be able to construct the
RDBClientHostWrapper.
"""
labels = set(['a', 'b', 'c'])
acls = set(['d', 'e'])
server_host = rdb_hosts.RDBServerHostWrapper(
self.db_helper.create_host('h1', deps=labels, acls=acls))
acl_ids = set([aclgroup.id for aclgroup in
self.db_helper.get_acls(name__in=acls)])
label_ids = set([label.id for label in
self.db_helper.get_labels(name__in=labels)])
# The RDBServerHostWrapper keeps ids of labels/acls to perform
# comparison operations within the rdb, but converts labels to
# strings because this is the format the scheduler expects them in.
self.assertTrue(set(server_host.labels) == label_ids and
set(server_host.acls) == acl_ids)
formatted_server_host = server_host.wire_format()
client_host = rdb_hosts.RDBClientHostWrapper(**formatted_server_host)
self.assertTrue(set(client_host.labels) == labels and
set(client_host.acls) == acl_ids)
bare_formatted_server_host = server_host.wire_format(
unwrap_foreign_keys=False)
self.assertTrue(bare_formatted_server_host.get('labels') is None and
bare_formatted_server_host.get('acls') is None)
def testLeasing(self):
"""Test that leasing a leased host raises an exception.
@raises AssertionError: If double leasing a host doesn't raise
an RDBException, or the leased bits are not set after the
first attempt at leasing it.
@raises RDBException: If the host is created with the leased bit set.
"""
hostname = 'h1'
server_host = rdb_hosts.RDBServerHostWrapper(
self.db_helper.create_host(hostname))
server_host.lease()
host = self.db_helper.get_host(hostname=hostname)[0]
self.assertTrue(host.leased and server_host.leased)
self.assertRaises(rdb_utils.RDBException, server_host.lease)
def testPlatformAndLabels(self):
"""Test that a client host returns the right platform and labels.
@raises AssertionError: If client host cannot return the right platform
and labels.
"""
platform_name = 'x86'
label_names = ['a', 'b']
self.db_helper.create_label(name=platform_name, platform=True)
server_host = rdb_hosts.RDBServerHostWrapper(
self.db_helper.create_host(
'h1', deps=set(label_names + [platform_name])))
client_host = rdb_hosts.RDBClientHostWrapper(
**server_host.wire_format())
platform, labels = client_host.platform_and_labels()
self.assertTrue(platform == platform_name)
self.assertTrue(set(labels) == set(label_names))
def testClientUpdateSave(self):
"""Test that a client host is capable of saving its attributes.
Create a client host, set its attributes and verify that the attributes
are saved properly by recreating a server host and checking them.
@raises AssertionError: If the server host has the wrong attributes.
"""
hostname = 'h1'
db_host = self.db_helper.create_host(hostname, leased=True)
server_host_dict = rdb_hosts.RDBServerHostWrapper(db_host).wire_format()
client_host = rdb_hosts.RDBClientHostWrapper(**server_host_dict)
host_data = {'hostname': hostname, 'id': db_host.id}
default_values = rdb_models.AbstractHostModel.provide_default_values(
host_data)
for k, v in default_values.iteritems():
self.assertTrue(server_host_dict[k] == v)
updated_client_fields = {
'locked': True,
'leased': False,
'status': 'FakeStatus',
'invalid': True,
'protection': 1,
'dirty': True,
}
client_host.__dict__.update(updated_client_fields)
client_host.save()
updated_server_host = rdb_hosts.RDBServerHostWrapper(
self.db_helper.get_host(hostname=hostname)[0]).wire_format()
for k, v in updated_client_fields.iteritems():
self.assertTrue(updated_server_host[k] == v)
def testUpdateField(self):
"""Test that update field on the client host works as expected.
@raises AssertionError: If a bad update is processed without an
exception, of a good update isn't processed as expected.
"""
hostname = 'h1'
db_host = self.db_helper.create_host(hostname, dirty=False)
server_host_dict = rdb_hosts.RDBServerHostWrapper(db_host).wire_format()
client_host = rdb_hosts.RDBClientHostWrapper(**server_host_dict)
self.assertRaises(rdb_utils.RDBException, client_host.update_field,
*('id', 'fakeid'))
self.assertRaises(rdb_utils.RDBException, client_host.update_field,
*('Nonexist', 'Nonexist'))
client_host.update_field('dirty', True)
self.assertTrue(
self.db_helper.get_host(hostname=hostname)[0].dirty == True and
client_host.dirty == True)
new_status = 'newstatus'
client_host.set_status(new_status)
self.assertTrue(
self.db_helper.get_host(hostname=hostname)[0].status ==
new_status and client_host.status == new_status)
if __name__ == '__main__':
unittest.main()