# Copyright 2018 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 contextlib
import json
import logging
from lxml import etree
import os
import StringIO
from autotest_lib.client.common_lib import utils
class ChartFixture:
"""Sets up chart tablet to display dummy scene image."""
DISPLAY_SCRIPT = '/usr/local/autotest/bin/display_chart.py'
def __init__(self, chart_host, scene_uri):
self.host = chart_host
self.scene_uri = scene_uri
self.display_pid = None
def initialize(self):
"""Prepare scene file and display it on chart host."""
logging.info('Prepare scene file')
chart_dir = self.host.get_tmp_dir()
scene_path = os.path.join(
chart_dir, self.scene_uri[self.scene_uri.rfind('/') + 1:])
self.host.run('wget', args=('-O', scene_path, self.scene_uri))
self.host.run('chmod', args=('-R', '755', chart_dir))
logging.info('Display scene file')
self.display_pid = self.host.run_background(
'python %s %s' % (self.DISPLAY_SCRIPT, scene_path))
# TODO(inker): Suppose chart should be displayed very soon. Or require
# of waiting until chart actually displayed.
def cleanup(self):
"""Cleanup display script."""
if self.display_pid is not None:
self.host.run('kill', args=('-2', str(self.display_pid)))
def get_chart_address(host_address, args):
"""Get address of chart tablet from commandline args or mapping logic in
test lab.
@param host_address: a list of hostname strings.
@param args: a dict parse from commandline args.
@return:
A list of strings for chart tablet addresses.
"""
address = utils.args_to_dict(args).get('chart')
if address is not None:
return address.split(',')
elif utils.is_in_container():
return [
utils.get_lab_chart_address(host)
for host in host_address
]
else:
return None
class DUTFixture:
"""Sets up camera filter for target camera facing on DUT."""
TEST_CONFIG_PATH = '/var/cache/camera/test_config.json'
GENERATE_CAMERA_PROFILE = os.path.join('/usr', 'bin',
'generate_camera_profile')
GENERATE_CAMERA_PROFILE_BACKUP = GENERATE_CAMERA_PROFILE + '.bak'
CAMERA_PROFILE_PATH = ('/mnt/stateful_partition/encrypted/var/cache/camera'
'/media_profiles.xml')
def __init__(self, test, host, facing):
self.test = test
self.host = host
self.facing = facing
@contextlib.contextmanager
def _set_selinux_permissive(self):
selinux_mode = self.host.run_output('getenforce')
self.host.run('setenforce 0')
yield
self.host.run('setenforce', args=(selinux_mode, ))
def _filter_camera_profile(self, content, facing):
"""Filter camera profile of target facing from content of camera
profile.
@return:
New camera profile with only target facing, camera ids are
renumbered from 0.
"""
tree = etree.parse(
StringIO.StringIO(content),
parser=etree.XMLParser(compact=False))
root = tree.getroot()
profiles = root.findall('CamcorderProfiles')
logging.debug('%d number of camera(s) found in camera profile',
len(profiles))
assert 1 <= len(profiles) <= 2
if len(profiles) == 2:
cam_id = 0 if facing == 'back' else 1
for p in profiles:
if cam_id == int(p.attrib['cameraId']):
p.attrib['cameraId'] = '0'
else:
root.remove(p)
else:
with self.test._login_chrome(
board=self.test._get_board_name(),
reboot=False), self._set_selinux_permissive():
has_front_camera = (
'feature:android.hardware.camera.front' in self.host.
run_output('android-sh -c "pm list features"'))
logging.debug('has_front_camera=%s', has_front_camera)
if (facing == 'front') != has_front_camera:
root.remove(profiles[0])
return etree.tostring(
tree, xml_declaration=True, encoding=tree.docinfo.encoding)
def _read_file(self, filepath):
"""Read content of filepath from host."""
tmp_path = os.path.join(self.test.tmpdir, os.path.basename(filepath))
self.host.get_file(filepath, tmp_path, delete_dest=True)
with open(tmp_path) as f:
return f.read()
def _write_file(self, filepath, content, permission=None, owner=None):
"""Write content to filepath on remote host.
@param permission: set permission to 0xxx octal number of remote file.
@param owner: set owner of remote file.
"""
tmp_path = os.path.join(self.test.tmpdir, os.path.basename(filepath))
with open(tmp_path, 'w') as f:
f.write(content)
if permission is not None:
os.chmod(tmp_path, permission)
self.host.send_file(tmp_path, filepath, delete_dest=True)
if owner is not None:
self.host.run('chown', args=(owner, filepath))
def initialize(self):
"""Filter out camera other than target facing on DUT."""
logging.info('Restart camera service with filter option')
self._write_file(
self.TEST_CONFIG_PATH,
json.dumps({
'enable_back_camera': self.facing == 'back',
'enable_front_camera': self.facing == 'front',
'enable_external_camera': False
}),
owner='arc-camera')
self.host.run('restart cros-camera')
# To replace camera profile in ARC++ container, arc_setup will run
# GENERATE_CAMERA_PROFILE and mount its generated profile under
# CAMERA_PROFILE_PATH into container.
logging.info('Replace camera profile in ARC++ container')
self.host.run(
'mv',
args=(self.GENERATE_CAMERA_PROFILE,
self.GENERATE_CAMERA_PROFILE_BACKUP))
self._write_file(
self.GENERATE_CAMERA_PROFILE,
'''\
#!/bin/bash
# Put this executable file to %r to make sure arc-setup knows
# we're using dynamic media_profiles.xml copy from host path
# %r''' % (self.GENERATE_CAMERA_PROFILE, self.CAMERA_PROFILE_PATH),
permission=0755)
profile = self._read_file(self.CAMERA_PROFILE_PATH)
new_profile = self._filter_camera_profile(profile, self.facing)
self._write_file(self.CAMERA_PROFILE_PATH, new_profile)
self.host.run('restart ui')
def cleanup(self):
"""Cleanup camera filter."""
logging.info('Remove filter option and restore camera service')
self.host.run('rm', args=('-f', self.TEST_CONFIG_PATH))
self.host.run('restart cros-camera')
# Restore GENERATE_CAMERA_PROFILE to regenerate camera profile on DUT.
logging.info('Restore camera profile in ARC++ container')
self.host.run(
'mv',
args=(self.GENERATE_CAMERA_PROFILE_BACKUP,
self.GENERATE_CAMERA_PROFILE),
ignore_status=True)
self.host.run('restart ui')