phablet-tools 1.2+16.04.20160317-0ubuntu1 / usr / bin / phablet-click-test-setup

#! /usr/bin/python
# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
# Copyright (C) 2013 Canonical Ltd.
# Author: Sergio Schvezov <sergio.schvezov@canonical.com>

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; version 3 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from __future__ import print_function

from launchpadlib.launchpad import Launchpad
from phabletutils.device import AndroidBridge
from os import path
from subprocess import (check_call, check_output)

import argparse
import atexit
import json
import os
import shutil
import tempfile
import urllib2
import sys


class LP(object):

    @property
    def arch_series(self):
        return self._arch_series

    @property
    def archive(self):
        return self._archive

    @property
    def series(self):
        return self._series

    def __init__(self, distribution, release, arch):
        self.lp = Launchpad.login_anonymously('phablet-click-test-setup',
                                              'production',
                                              '~/.launchpadlib/cache',
                                              version="devel")
        distro = self.lp.distributions[distribution]
        self._series = distro.getSeries(name_or_version=release)
        self._arch_series = self._series.getDistroArchSeries(archtag=arch)
        self._archive = self.lp.distributions[distribution].main_archive


lp = None

basic_packages = (
)


class UbuntuDevice(AndroidBridge):

    def __init__(self, device=None):
        super(UbuntuDevice, self).__init__(device=device)

    def get_package_version(self, package):
        cmd = 'shell dpkg-query --show --showformat \'\${Version}\' %s ' % \
            package
        return check_output(self._cmd % cmd, shell=True)

    def get_click_manifest(self, user):
        cmd = 'shell click list --user=%s --manifest' % user
        return json.loads(check_output(self._cmd % cmd, shell=True))


def parse_arguments():
    parser = argparse.ArgumentParser(
        description='Sets up a device to be tested.')
    parser.add_argument('--manifest',
                        help='Manifest with package dependencies')
    parser.add_argument('-s', '--serial',
                        help='''Device serial. Use when more than
                                one device is connected.''',
                        )
    parser.add_argument('--click', default=None,
                        help='Specific click package to setup for.'
                        )
    parser.add_argument('--user', default='phablet',
                        help='User on device to use')
    parser.add_argument('--wipe', action='store_true',
                        help='Clean up previous setup on device')
    parser.add_argument('--distribution', default='ubuntu',
                        help='Selects distribution to use')
    parser.add_argument('--series', default=None,
                        help='Selects series to use')
    parser.add_argument('--depends', dest='depends', default=None, nargs='+',
                        help='Packages to copy to the device. Packages cannot \
                        contain or depend on architecture specifics.')
    return parser.parse_args()


def cleanup(directory):
    if os.path.exists(directory) and os.path.isdir(directory):
        print('Removing directory %s' % directory)
        shutil.rmtree(directory)


def get_python_binary_package(package, target_dir):
    tmp_dir = tempfile.mkdtemp()
    atexit.register(cleanup, tmp_dir)
    print('Fetching %s - into %s' % (package, tmp_dir))
    bpph = lp.archive.getPublishedBinaries(
        binary_name=package,
        distro_arch_series=lp.arch_series,
        status="Published", pocket='Release',
        exact_match=True)
    version = bpph[0].binary_package_version
    bin_url = bpph[0].binaryFileUrls()[0]
    print('Downloading %s version %s' % (package, version))
    url = urllib2.urlopen(bin_url)
    data = url.read()
    package_name = "%s_%s_%s.deb" % (package, version, 'armhf')
    target = path.join(tmp_dir, package_name)
    with open(target, "wb") as package:
        package.write(data)
    extract_dir = path.join(tmp_dir, 'deb')
    check_call(['dpkg-deb', '-x', target, extract_dir])
    python3_modules_dir = os.path.join(extract_dir,
                                       'usr/lib/python3/dist-packages')
    if os.path.exists(python3_modules_dir):
        for f in os.listdir(python3_modules_dir):
            source = path.join(python3_modules_dir, f)
            if path.islink(source):
                source = path.join(python3_modules_dir, os.readlink(source))
            print('Moving %s to %s' % (source, target_dir))
            shutil.move(source, target_dir)


def get_source_package_tests(package, version, target_dir):
    tmp_dir = tempfile.mkdtemp()
    atexit.register(cleanup, tmp_dir)
    print('Fetching %s - into %s' % (package, tmp_dir))
    try:
        sp = lp.archive.getPublishedSources(
            source_name=package,
            version=version,
            distro_series=lp.series,
            exact_match=True)
        # TODO needs filtering
        source_url = sp[0].sourceFileUrls()[0]
        print('Downloading %s version %s from %s' % (package,
                                                     version, source_url))
        url = urllib2.urlopen(source_url)
        data = url.read()

        target = path.join(tmp_dir, 'source_file')

        with open(target, "wb") as p:
            p.write(data)
        check_call(['tar', '-xf', target], cwd=tmp_dir)

        print('Keeping tests from obtained package')
        package_source = filter((lambda x: x.startswith(package)),
                                os.listdir(tmp_dir))
        # Just let an exception be thrown if more than one match which means
        # there's a problem somewhere
        test_base_dir = path.join(tmp_dir,
                                  package_source[0],
                                  'tests',
                                  'autopilot')
        test_dirs = filter((lambda x: path.isdir(path.join(test_base_dir, x))),
                           os.listdir(test_base_dir))
        for test_dir in test_dirs:
            test_dir = path.join(test_base_dir, test_dir)
            print('Moving %s to %s' % (test_dir, target_dir))
            shutil.move(test_dir, target_dir)
    except IndexError:
        print('package %s, version %s not found in %s' % (package,
                                                          version, lp.series))
        print('Use --distribution and --series if needed')
        raise
    except:
        print('Unexpected error: ', sys.exc_info()[0])
        raise


def get_bzr_tests(branch, revision, autopilot_dir, target_dir):
    tmp_dir = tempfile.mkdtemp()
    atexit.register(cleanup, tmp_dir)
    print('Checking out %s to %s' % (branch, path.join(tmp_dir, 'work')))
    check_call(['bzr', 'checkout', '--lightweight', branch, '-r', revision,
               'work'], cwd=tmp_dir)
    test_base_dir = path.join(tmp_dir, 'work', autopilot_dir)
    test_dirs = filter((lambda x: path.isdir(path.join(test_base_dir, x))),
                       os.listdir(test_base_dir))
    for test_dir in test_dirs:
        test_dir = path.join(test_base_dir, test_dir)
        print('Moving %s to %s' % (test_dir, target_dir))
        shutil.move(test_dir, target_dir)


def fetch_test_base(adb, test_dir, arch, python_packages=None):
    if python_packages:
        for package in python_packages:
            get_python_binary_package(package, test_dir)
    for package in basic_packages:
        binaries = package['binary']
        binary = binaries[arch] if arch in binaries else binaries['all']
        version = adb.get_package_version(binary)
        get_source_package_tests(package['source'], version, test_dir)


def fetch_click_tests(adb, test_dir, user, click=None):
    manifest = adb.get_click_manifest(user)
    if click:
        print('Only setting up for %s' % click)
        manifest = [entry for entry in manifest if click == entry['name']]
    print('Only keeping entries with x-source')
    manifest = [entry for entry in manifest if 'x-source' in entry]
    for entry in manifest:
        get_bzr_tests(entry['x-source']['vcs-bzr'],
                      entry['x-source']['vcs-bzr-revno'],
                      'tests/autopilot',
                      test_dir)


def main():
    global lp
    args = parse_arguments()
    adb = UbuntuDevice(args.serial)
    adb.start()
    arch = adb.shell('sudo -iu phablet dpkg --print-architecture').strip()
    if not args.series:
        args.series = adb.shell('lsb_release -c -s').strip()
    lp = LP(args.distribution, args.series, arch)
    test_dir = tempfile.mkdtemp()
    atexit.register(cleanup, test_dir)
    fetch_test_base(adb, test_dir, arch, args.depends)
    fetch_click_tests(adb, test_dir, args.user, args.click)
    destination = path.join('/home', args.user, 'autopilot')
    if args.wipe:
        print('Clearing previous test setup in %s' % destination)
        adb.shell('rm -Rf %s' % destination)
    adb.push(test_dir, destination)
    adb.chown(args.user, destination)
    print('Files setup in %s on device' % destination)


if __name__ == '__main__':
    main()