| # Copyright 2015 The Chromium 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 dbus |
| import json |
| import logging |
| import os |
| import shutil |
| import tempfile |
| import time |
| |
| from autotest_lib.client.cros import dbus_util |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.common_lib import utils |
| from autotest_lib.client.common_lib.cros.fake_device_server import \ |
| fake_gcd_helper |
| from autotest_lib.client.common_lib.cros.fake_device_server.client_lib import \ |
| commands |
| from autotest_lib.client.common_lib.cros.fake_device_server.client_lib import \ |
| devices |
| from autotest_lib.client.common_lib.cros.fake_device_server.client_lib import \ |
| fail_control |
| from autotest_lib.client.common_lib.cros.fake_device_server.client_lib import \ |
| oauth |
| from autotest_lib.client.common_lib.cros.fake_device_server.client_lib import \ |
| registration |
| from autotest_lib.client.common_lib.cros.tendo import buffet_config |
| from autotest_lib.client.common_lib.cros.tendo import buffet_dbus_helper |
| |
| |
| TEST_NAME = 'test_name ' |
| TEST_DESCRIPTION = 'test_description ' |
| TEST_LOCATION = 'test_location ' |
| |
| TEST_COMMAND_CATEGORY = 'registration_test' |
| TEST_COMMAND_NAME = '_TestEcho' |
| TEST_COMMAND_PARAM = 'message' |
| TEST_COMMAND_DEFINITION = { |
| TEST_COMMAND_CATEGORY: { |
| TEST_COMMAND_NAME: { |
| 'parameters': { TEST_COMMAND_PARAM: { 'type': 'string' } }, |
| 'results': {}, |
| 'name': 'Test Echo Command', |
| } |
| } |
| } |
| |
| STATUS_UNCONFIGURED = 'unconfigured' |
| STATUS_CONNECTING = 'connecting' |
| STATUS_CONNECTED = 'connected' |
| STATUS_INVALID_CREDENTIALS = 'invalid_credentials' |
| |
| def _assert_has(resource, key, value, resource_description): |
| if resource is None: |
| raise error.TestFail('Wanted %s[%s]=%r, but %s is None.' % |
| (resource_description, key, value)) |
| if key not in resource: |
| raise error.TestFail('%s not in %s' % (key, resource_description)) |
| |
| if resource[key] != value: |
| raise error.TestFail('Wanted %s[%s]=%r, but got %r' % |
| (resource_description, key, value, resource[key])) |
| |
| |
| class BuffetTester(object): |
| """Helper class for buffet tests.""" |
| |
| |
| def __init__(self): |
| """Initialization routine.""" |
| # We're going to confirm buffet is polling by issuing commands to |
| # the mock GCD server, then checking that buffet gets them. The |
| # commands are test.TestEcho commands with a single parameter |
| # |message|. |self._expected_messages| is a list of these messages. |
| self._expected_messages = [] |
| # We store our command definitions under this root. |
| self._temp_dir_path = None |
| # Spin up our mock server. |
| self._gcd = fake_gcd_helper.FakeGCDHelper() |
| self._gcd.start() |
| # Create the command definition we want to use. |
| self._temp_dir_path = tempfile.mkdtemp() |
| commands_dir = os.path.join(self._temp_dir_path, 'commands') |
| os.mkdir(commands_dir) |
| command_definition_path = os.path.join( |
| commands_dir, '%s.json' % TEST_COMMAND_CATEGORY) |
| with open(command_definition_path, 'w') as f: |
| f.write(json.dumps(TEST_COMMAND_DEFINITION)) |
| utils.run('chown -R buffet:buffet %s' % self._temp_dir_path) |
| logging.debug('Created test commands definition: %s', |
| command_definition_path) |
| # Create client proxies for interacting with oyr fake server. |
| self._registration_client = registration.RegistrationClient( |
| server_url=buffet_config.LOCAL_SERVICE_URL, |
| api_key=buffet_config.TEST_API_KEY) |
| self._device_client = devices.DevicesClient( |
| server_url=buffet_config.LOCAL_SERVICE_URL, |
| api_key=buffet_config.TEST_API_KEY) |
| self._oauth_client = oauth.OAuthClient( |
| server_url=buffet_config.LOCAL_SERVICE_URL, |
| api_key=buffet_config.TEST_API_KEY) |
| self._fail_control_client = fail_control.FailControlClient( |
| server_url=buffet_config.LOCAL_SERVICE_URL, |
| api_key=buffet_config.TEST_API_KEY) |
| self._command_client = commands.CommandsClient( |
| server_url=buffet_config.LOCAL_SERVICE_URL, |
| api_key=buffet_config.TEST_API_KEY) |
| self._config = buffet_config.BuffetConfig( |
| log_verbosity=3, |
| test_definitions_dir=self._temp_dir_path) |
| |
| |
| def check_buffet_status_is(self, expected_status, |
| expected_device_id='', |
| timeout_seconds=0): |
| """Assert that buffet has the given registration status. |
| |
| Optionally, a timeout can be specified to wait until the |
| status changes. |
| |
| @param expected_device_id: device id created during registration. |
| @param expected_status: the status to wait for. |
| @param timeout_seconds: number of seconds to wait for status to change. |
| |
| """ |
| buffet = buffet_dbus_helper.BuffetDBusHelper() |
| start_time = time.time() |
| while True: |
| actual_status = buffet.status |
| actual_device_id = buffet.device_id |
| if (actual_status == expected_status and |
| actual_device_id == expected_device_id): |
| return |
| time_spent = time.time() - start_time |
| if time_spent > timeout_seconds: |
| if actual_status != expected_status: |
| raise error.TestFail('Buffet should be %s, but is %s ' |
| '(waited %.1f seconds).' % |
| (expected_status, actual_status, |
| time_spent)) |
| if actual_device_id != expected_device_id: |
| raise error.TestFail('Device ID should be %s, but is %s ' |
| '(waited %.1f seconds).' % |
| (expected_device_id, actual_device_id, |
| time_spent)) |
| time.sleep(0.5) |
| |
| |
| def check_buffet_is_polling(self, device_id, timeout_seconds=30): |
| """Assert that buffet is polling for new commands. |
| |
| @param device_id: string device id created during registration. |
| @param timeout_seconds: number of seconds to wait for polling |
| to start. |
| |
| """ |
| new_command_message = ('This is message %d' % |
| len(self._expected_messages)) |
| command_resource = { |
| 'name': '%s.%s' % (TEST_COMMAND_CATEGORY, TEST_COMMAND_NAME), |
| 'deviceId': device_id, |
| 'parameters': {TEST_COMMAND_PARAM: new_command_message} |
| } |
| self._expected_messages.append(new_command_message) |
| self._command_client.create_command(device_id, command_resource) |
| # Confirm that the command eventually appears on buffet. |
| buffet = buffet_dbus_helper.BuffetDBusHelper() |
| polling_interval_seconds = 0.5 |
| start_time = time.time() |
| while time.time() - start_time < timeout_seconds: |
| objects = dbus_util.dbus2primitive( |
| buffet.object_manager.GetManagedObjects()) |
| cmds = [interfaces[buffet_dbus_helper.COMMAND_INTERFACE] |
| for path, interfaces in objects.iteritems() |
| if buffet_dbus_helper.COMMAND_INTERFACE in interfaces] |
| messages = [cmd['Parameters'][TEST_COMMAND_PARAM] for cmd in cmds |
| if (cmd['Name'] == '%s.%s' % (TEST_COMMAND_CATEGORY, |
| TEST_COMMAND_NAME))] |
| # |cmds| is a list of property sets |
| if len(messages) != len(self._expected_messages): |
| # Still waiting for our pending command to show up. |
| time.sleep(polling_interval_seconds) |
| continue |
| logging.debug('Finally saw the right number of commands over ' |
| 'DBus: %r', cmds) |
| if sorted(messages) != sorted(self._expected_messages): |
| raise error.TestFail( |
| 'Expected commands with messages=%r but got %r.' % |
| (self._expected_messages, messages)) |
| logging.info('Buffet has DBus proxies for commands with ' |
| 'messages: %r', self._expected_messages) |
| return |
| raise error.TestFail('Timed out waiting for Buffet to expose ' |
| 'pending commands with messages: %r' % |
| self._expected_messages) |
| |
| |
| def register_with_server(self): |
| """Make buffet register with the cloud server. |
| |
| This includes the whole registration flow and ends with buffet |
| obtained an access token for future interactions. The status |
| is guaranteed to be STATUS_CONNECTED when this |
| method returns. |
| |
| @return string: the device_id obtained during registration. |
| |
| """ |
| ticket = self._registration_client.create_registration_ticket() |
| logging.info('Created ticket: %r', ticket) |
| buffet = buffet_dbus_helper.BuffetDBusHelper() |
| buffet.manager.UpdateDeviceInfo(dbus.String(TEST_NAME), |
| dbus.String(TEST_DESCRIPTION), |
| dbus.String(TEST_LOCATION)) |
| device_id = dbus_util.dbus2primitive( |
| buffet.manager.RegisterDevice(dbus.String(ticket['id']))) |
| # Confirm that registration has populated some fields. |
| device_resource = self._device_client.get_device(device_id) |
| logging.debug('Got device resource=%r', device_resource) |
| _assert_has(device_resource, 'name', TEST_NAME, |
| 'device resource') |
| _assert_has(device_resource, 'modelManifestId', 'AATST', |
| 'device resource') |
| logging.info('Registration successful') |
| self.check_buffet_status_is(STATUS_CONNECTED, |
| expected_device_id=device_id, |
| timeout_seconds=5) |
| return device_id |
| |
| |
| def restart_buffet(self, reset_state): |
| """Function for restarting the buffet daemon. |
| |
| @param reset_state: If True, all local buffet state will be deleted. |
| """ |
| self._config.restart_with_config(clean_state=reset_state) |
| |
| |
| def close(self): |
| """Cleanup to be used when done with this instance.""" |
| buffet_config.naive_restart() |
| self._gcd.close() |
| if self._temp_dir_path is not None: |
| shutil.rmtree(self._temp_dir_path, True) |
