1# Lint as: python2, python3 2# Copyright 2018 The Chromium OS Authors. All rights reserved. 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5 6import contextlib 7import json 8import logging 9import os 10import time 11 12from autotest_lib.client.common_lib import error, utils 13 14 15class ChartFixture: 16 """Sets up chart tablet to display placeholder scene image.""" 17 DISPLAY_SCRIPT = '/usr/local/autotest/bin/display_chart.py' 18 OUTPUT_LOG = '/tmp/chart_service.log' 19 20 def __init__(self, chart_host, scene_uri, job=None): 21 self.host = chart_host 22 self.scene_uri = scene_uri 23 self.job = job 24 self.display_pid = None 25 self.host.run(['rm', '-f', self.OUTPUT_LOG], ignore_status=True) 26 27 def initialize(self): 28 """Prepare scene file and display it on chart host.""" 29 logging.info('Prepare scene file') 30 chart_version = self.host.run( 31 'cat /etc/lsb-release | grep CHROMEOS_RELEASE_BUILDER_PATH') 32 logging.info('Chart version: %s', chart_version) 33 if utils.is_in_container(): 34 # Reboot chart to clean the dirty state from last test. See 35 # b/201032899. 36 version = self.host.get_release_builder_path() 37 self.job.run_test('provision_QuickProvision', 38 host=self.host, 39 value=version, 40 force_update_engine=True) 41 42 tmpdir = self.host.get_tmp_dir() 43 scene_path = os.path.join( 44 tmpdir, self.scene_uri[self.scene_uri.rfind('/') + 1:]) 45 self.host.run('wget', args=('-O', scene_path, self.scene_uri)) 46 47 logging.info('Display scene file') 48 self.display_pid = self.host.run_background( 49 'python {script} {scene} >{log} 2>&1'.format( 50 script=self.DISPLAY_SCRIPT, 51 scene=scene_path, 52 log=self.OUTPUT_LOG)) 53 54 logging.info( 55 'Poll for "is ready" message for ensuring chart is ready.') 56 timeout = 60 57 poll_time_step = 0.1 58 while timeout > 0: 59 if self.host.run( 60 'grep', 61 args=('-q', 'Chart is ready.', self.OUTPUT_LOG), 62 ignore_status=True).exit_status == 0: 63 break 64 time.sleep(poll_time_step) 65 timeout -= poll_time_step 66 else: 67 raise error.TestError('Timeout waiting for chart ready') 68 69 def cleanup(self): 70 """Cleanup display script.""" 71 if self.display_pid is not None: 72 self.host.run( 73 'kill', 74 args=('-2', str(self.display_pid)), 75 ignore_status=True) 76 self.host.get_file(self.OUTPUT_LOG, '.') 77 78 79def get_chart_address(host_address, args): 80 """Get address of chart tablet from commandline args or mapping logic in 81 test lab. 82 83 @param host_address: a list of hostname strings. 84 @param args: a dict parse from commandline args. 85 @return: 86 A list of strings for chart tablet addresses. 87 """ 88 address = utils.args_to_dict(args).get('chart') 89 if address is not None: 90 return address.split(',') 91 elif utils.is_in_container(): 92 return [utils.get_lab_chart_address(host) for host in host_address] 93 else: 94 return None 95 96 97class DUTFixture: 98 """Sets up camera filter for target camera facing on DUT.""" 99 TEST_CONFIG_PATH = '/var/cache/camera/test_config.json' 100 CAMERA_SCENE_LOG = '/tmp/scene.jpg' 101 102 def __init__(self, test, host, facing): 103 self.test = test 104 self.host = host 105 self.facing = facing 106 107 @contextlib.contextmanager 108 def _set_selinux_permissive(self): 109 selinux_mode = self.host.run_output('getenforce') 110 self.host.run('setenforce 0') 111 yield 112 self.host.run('setenforce', args=(selinux_mode, )) 113 114 def _write_file(self, filepath, content, permission=None, owner=None): 115 """Write content to filepath on remote host. 116 @param permission: set permission to 0xxx octal number of remote file. 117 @param owner: set owner of remote file. 118 """ 119 tmp_path = os.path.join(self.test.tmpdir, os.path.basename(filepath)) 120 with open(tmp_path, 'w') as f: 121 f.write(content) 122 if permission is not None: 123 os.chmod(tmp_path, permission) 124 self.host.send_file(tmp_path, filepath, delete_dest=True) 125 if owner is not None: 126 self.host.run('chown', args=(owner, filepath)) 127 128 def initialize(self): 129 """Filter out camera other than target facing on DUT.""" 130 self._write_file( 131 self.TEST_CONFIG_PATH, 132 json.dumps({ 133 'enable_back_camera': self.facing == 'back', 134 'enable_front_camera': self.facing == 'front', 135 'enable_external_camera': False 136 }), 137 owner='arc-camera') 138 139 # cros_camera_service will reference the test config to filter out 140 # undesired cameras. 141 logging.info('Restart camera service with filter option') 142 self.host.upstart_restart('cros-camera') 143 144 # arc_setup will reference the test config to filter out the media 145 # profile of undesired cameras. 146 logging.info('Restart ARC++ container with camera test config') 147 self.host.run('restart ui') 148 149 @contextlib.contextmanager 150 def _stop_camera_service(self): 151 # Ensure camera service is running or the 152 # upstart_stop()/upstart_restart() may failed due to in 153 # "start|post-stop" sleep for respawning state. See b/183904344 for 154 # detail. 155 logging.info('Wait for presence of camera service') 156 self.host.wait_for_service('cros-camera') 157 158 self.host.upstart_stop('cros-camera') 159 yield 160 self.host.upstart_restart('cros-camera') 161 162 def log_camera_scene(self): 163 """Capture an image from camera as the log for debugging scene related 164 problem.""" 165 166 gtest_filter = ( 167 'Camera3StillCaptureTest/' 168 'Camera3DumpSimpleStillCaptureTest.DumpCaptureResult/0') 169 with self._stop_camera_service(): 170 self.host.run( 171 'sudo', 172 args=('--user=arc-camera', 'cros_camera_test', 173 '--gtest_filter=' + gtest_filter, 174 '--camera_facing=' + self.facing, 175 '--dump_still_capture_path=' + 176 self.CAMERA_SCENE_LOG)) 177 178 self.host.get_file(self.CAMERA_SCENE_LOG, '.') 179 180 def cleanup(self): 181 """Cleanup camera filter.""" 182 logging.info('Remove filter option and restore camera service') 183 with self._stop_camera_service(): 184 self.host.run('rm', args=('-f', self.TEST_CONFIG_PATH)) 185 186 logging.info('Restore camera profile in ARC++ container') 187 self.host.run('restart ui') 188
