| # 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) |
| |