blob: 0d9c46294e8dcd83fc270ae6815ef0178402fc0d [file] [log] [blame]
# -*- coding: utf-8 -*-
# Copyright (c) 2013 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 os
import subprocess as sub
import sys
import xml.parsers.expat
This script can be used to download chromeos test images from the the cloud
storage. We assume that the cloud storage path remains similar all the time.
eg: gs://chromeos-releases/dev-channel/link/4162.0.0/
So passing the above value to 'gstuil cp ' will download
the relevant CrOS image.
After downloading the file we uzip the file and wait for the user
to give the path to which the image should be copied.
eg: /dev/sdb
This script will create a build_defaults.xml file with the defaults info in
the scripts folder. So please run this from a folder with a write permissions.
def run_command(command, shell=False):
""" Run a given command using subprocess.
@param command: The command to run. Should be a list.
@param shell: Flag to use bash to run the command.
@return out: Output of the process.
process = sub.Popen(command, stdout=sub.PIPE,
stderr=sub.PIPE, shell=shell)
(out, err) = process.communicate()
if err and 'Copying gs://chromeos-releases/' not in err:
raise Exception('We got an error %s' % err)
return out
class defaults():
# TODO: Have a defaults file to read/write into it.
channel_list = ['dev', 'beta', 'stable']
device_list = ['x86-mario', 'x86-zgb', 'x86-alex', 'daisy', 'stout',
'link', 'butterfly', 'stout', 'stumpy', 'parrot', 'lumpy',]
default_dict = {'channel': None, 'device': None,
'branch': None, 'version': None}
defaults_file = 'build_defaults.xml'
def __init__(self):
if not os.path.exists(self.defaults_file):
def set_defaults(self, channel, device, version, branch):
"""Set the new defaults.
@param channel: default channel to be saved.
@param device: defalut device to be saved.
@param version: default version to be saved.
@param branch: default branch to be saved.
self.default_dict['channel'] = channel
self.default_dict['branch'] = branch
self.default_dict['version'] = version
self.default_dict['device'] = device
def _create_defaults_element(self, initial=True):
"""Create a defaults element to be added to the xml file.
@param initial: Flag to check if we are creating or updating the
@return defaults_element: Element with defaults attributes to be added.
defaults_element = '<defaults '
for key in self.default_dict.keys():
if initial or self.default_dict[key]:
defaults_element += '%s="%s" ' % (key, self.default_dict[key])
defaults_element += ('></defaults>\n')
return defaults_element
def _update_defaults(self):
"""Update the defaults list in database/xml."""
defaults_element = self._create_defaults_element(initial=False)
lines = open(self.defaults_file, 'r').readlines()
fin = open(self.defaults_file, 'w')
for line in lines:
if '<defaults ' in line:
break # this will always be the last line
def _create_defaults(self):
"""Create all the defaults."""
print 'Creating Defaults file.'
fout = open(self.defaults_file, 'wb')
root = '<feed xmlns=\'\' xml:lang=\'en\'>\n'
channel_element = '<channel>'
for channel in self.channel_list:
channel_element += '%s, ' % channel
channel_element += ('</channel>\n')
device_element = '<device>'
for device in self.device_list:
device_element += '%s, ' % device
device_element += ('</device>\n')
defaults_element = self._create_defaults_element()
def previous_defaults(self):
"""Get the default values."""
# Parse and read the xml data
return self.default_dict
def start_element(self, name, attrs):
if name == 'defaults':
self.default_dict = attrs
def char_data(self, data):
if 'dev' in data:
if 'link' in data:
def read_xml(self):
"""Read and parse the xml file."""
fin = open(self.defaults_file, 'r')
parse_data =
parser = xml.parsers.expat.ParserCreate()
parser.StartElementHandler = self.start_element
parser.CharacterDataHandler = self.char_data
def channel(self):
"""Read the default channel from the logs"""
return self.channel_list
def add_default_channel(self, data):
"""Add a channel if it does not exist in defaults."""
for channel in data.split(','):
if channel and (channel not in self.channel_list):
def add_default_devices(self, data):
"""Add a device if it does not exist in defaults."""
for device in data.split(','):
if device and device not in self.device_list:
def element_in_list(self, element, element_list):
for old_element in element_list:
if element in old_element:
return False
return True
class download_image():
# This script will work until the image path remains unchanged. If the
# cloud team decides to change the default path to the images, change this
# path constant accordingly. This is used build_source.
image_path = 'gs://chromeos-releases/%s-channel/%s/%s/'
def file_check(self, dest, folder=True, action=False):
""" Check if a file or folder exists.
This is used in two places, first to check if we have the test
folder and second to check if we already have the file to download.
@param dest: The path to look for. Should be a string.
@param folder: Flag to indicate if the path is a directory or not.
@param action: Flag to indicate if an action is required.
eg: Creation or deletion.
if folder and not os.path.exists(dest):
if action:
print 'Created directory %s' % dest
raise Exception('The path %s does not exist' % dest)
elif folder and not os.path.isdir(dest):
raise Exception('The path %s exists and is not a folder.' % dest)
elif not folder and os.path.exists(dest):
zip_file = os.path.basename(dest)
file_type = 'file'
if os.path.isdir(zip_file):
file_type = 'folder'
if action:
print ('The %s %s exists or is downloaded in %s' %
(os.path.basename(dest), file_type, os.path.dirname(dest)))
delete_flag = raw_input('Do you want to remove the existing?'
' (Y/N): ')
if delete_flag.lower() == 'y':
print 'Removing the file %s' % dest
print 'Continuing with the download.'
def check_download(self, dest, source):
"""Check if the download was successful. Verify the checksum and the
size of the downloaded file.
@param dest: The complete path to the downloaded file
@param source: The complete path to the file in cloudstorage.
out = run_command(['gsutil', 'ls', '-L', source])
checksum = size = ''
for line in out.split('\n'):
if 'ETag:' in line:
checksum = line.split(':')[1].strip('\t')
elif 'Content-Length:' in line:
size = line.split(':')[1].strip('\t')
sout = run_command(['ls -Sl %s' % dest], shell=True)
md5out = run_command(['md5sum', dest])
if not size in sout:
raise Exception ('File did not download completely. '
'It should be %s and we got %s' % (size, sout))
elif not checksum in md5out:
raise Exception ('The downloaded file is corrupted. '
'The checksum should %s and we got %s' %
(checksum, md5out))
def download(self, device=None, channel=None, version=None, branch=None):
"""Download a zip file containing the test image. The user is
expected to provide a clean input. A blank input means that
we use defaults. Defaults will be the last build
that the script attempted to download.
@param device : The device under test. eg: x86-mario, link. This should
be the exact name of the device as shown in the
@param channel: The channel under which we can get the test build.
eg: dev, beta, stable.
@param version: The version of the build. eg:3701.0.0, 3701.71.0.
@param branch : The branch under which we can find the version.
eg: R26, R27.
@return final_dest: The complete downloaded file path.
# TODO: download stops at 4109631488, check why
build_info = self.build_source(device=device, channel=channel,
version=version, branch=branch)
source = build_info[0]
dest = '/tmp/%s/test' % build_info[1]
final_dest = os.path.join(dest, os.path.basename(source))
self.file_check(dest, folder=True, action=True)
self.file_check(final_dest, folder=False, action=True)
print 'Downloading %s, Please wait ...' % source
run_command(['gsutil', 'cp', source, dest])
self.file_check(final_dest, folder=False)
self.check_download(final_dest, source)
run_command(['nautilus', dest])
return final_dest
def build_source(self, device=None, channel=None,
version=None, branch=None):
"""Multiple option parser. Allows user to give arguments or options
or call the function from their module to create the source path
of the CrOS test build zip file to be downloaded. The user is
expected to provide a clean input. A blank input means that we
use defaults. Defaults will be the last build
that the script attempted to download.
@param device : The device under test. eg: x86-mario, link. This should
be the exact name of the device as shown in the
@param channel: The channel under which we can get the test build.
eg: dev, beta, stable.
@param version: The version of the build. eg:3701.0.0, 3701.71.0.
@param branch : The branch under which we can find the version.
eg: R26, R27.
default = defaults()
# We save the previous configuration as default
prev_values = default.previous_defaults()
defa_channel = prev_values['channel']
defa_version = prev_values['version']
defa_branch = prev_values['branch']
defa_device = prev_values['device']
if not (channel and branch and version and device):
channel = raw_input('Enter the channel(default: %s): ' %
device = raw_input('Enter the device(default: %s): ' %
branch = raw_input('Enter the branch(default: %s): ' %
version = raw_input('Enter the version(default: %s): ' %
# Check the inputs again
if not (channel and branch and version and device):
if not (defa_channel and defa_branch and defa_version and
raise Exception ('Insufficient input to download the build. '
'Please use \" --help\" to get '
'more information about passing arguments.')
if not channel: # If no input, we use defaults
channel = defa_channel
if not branch:
branch = defa_branch
if not version:
version = defa_version
if not device:
device = defa_device
path = (self.image_path % (channel, device, version, branch,
version, device))
default.set_defaults(channel=channel, device=device,
version=version, branch=branch)
return [path, device]
class unzip_burn():
def unzip_file(self, zip_file):
""" Unzip the given file into a folder with the same name.
@param zip_file: The file to be unzipped.
@return dest: The folder with unzipped files.
dest = os.path.splitext(zip_file)[0]
# We avoid all the unnecesssary files to reduce the unzip time
command = ['unzip', zip_file, '-d', dest, '-x',
'autotest.tar.bz2', 'chromiumos_base_image.bin',
large_zips = ['stout', 'link']
for board in large_zips:
if board in zip_file:
if run_command(command):
return dest
def burn(self, image, drive):
""" Burn the image to the given drive.
@param image: The complete path to the image.
@param drive: The complete path to the drive.
print 'Burning the image %s on to drive %s ...' % (image, drive)
if run_command(['sudo dd if=%s of=%s' % (image, drive)],
print 'Image is now on %s drive.'
def main():
download = download_image()
image =
burn = unzip_burn()
print 'Unzipping the folder, please wait ...'
dest = burn.unzip_file(image)
if dest:
download_path = os.path.join(dest, 'chromiumos_test_image.bin')
print 'The test image is in %s' % download_path
drive_path = raw_input('Which drive to use?(default: /dev/sdc): ')
if '/dev' not in drive_path:
print 'Using the default drive path /dev/sdc.'
drive_path = '/dev/sdc'
if os.path.exists(download_path) and os.path.exists(drive_path):
burn.burn(download_path, drive_path)
if __name__ == '__main__':