Merge lp:~canonical-platform-qa/ubuntu-ota-tests/thinkingoutloud-wrapper-reflash-script into lp:ubuntu-ota-tests

[Christopher Lee]

Wrapper script to reflash the device to an old revision prior to running each ota update test with adt-run.

WIP thoughts around how to flash the device between test runs.
Currently it's really thinking out load:

 A script that runs the individual dep8 tests and reflashes the device in-between.

Some issues incl. the --testname argument not working as I expected it to and it still runs all the tests in d/t/control :-)

[Brendan Donegan (brendan-donegan)]

Some inline comments. Also the assumption of test files starting with a particular prefix is a little brittle. Is there any way we can parse the test names from debian/tests/control instead? If not that's ok

review: Needs Fixing

[Christopher Lee (veebers)]

> Some inline comments. Also the assumption of test files starting with a
> particular prefix is a little brittle. Is there any way we can parse the test
> names from debian/tests/control instead? If not that's ok

For a first pass this was quick and hacky. I agree that it's brittle, I'll update the branch with a solution if I can get it done in time. Otherwise we'll use what's here and iterate on it.

[Christopher Lee (veebers)]

Not sure what to do about the phablet-network setting as it requires sudo perms. Running the whole script is dumb as adt attempts to do ssh stuff and the perms are all screwy (and will change the perms of .ssh/known_hosts for instance).
At the moment it'll prompt you for the password, and it's likely that the test will take long enough that the next time it's needed you'll get prompted again.

review: Needs Fixing

[Richard Huddie (rhuddie)]

I've made updates to now use nmcli to setup the wifi connection. This is called using adt-run --setup-command option. During testing I found that if the network connection already exists, adt-run would fail becaue nmcli reported a failure, although it was not an error that would cause any issues. So that is why the nmcli command always reports true. But under normal use this should never be an issue as the device is just reflashed. I also set the default revision to -1 and added a 10 minute timeout for each test execution.

[lp:~canonical-platform-qa/ubuntu-ota-tests/thinkingoutloud-wrapper-reflash-script]

33. By Richard Huddie on 2015-03-23

Fix self tests and add new ones for wifi and timeout parameters.

[Christopher Lee (veebers)]

Code looks good to me, currently trying to make a run on the device but it's playing up. If someone can confirm this run on a device this turns into an Approve from me.

[Federico Gimenez (fgimenez)]

Works great, only two issues:

* The test_apply_update_noop is triggering the update because of it has an update available after reflashing to rev-1; we should add a guard as in [1], something like:

  @unittest.skipIf(is_update_available(),
                  'Update available, would be applied')

* There's a flake8 error in ./ubuntu_ota_tests/hooks.py:31, a file that hasn't been modified by this mp, don't know how it got there :)

Thanks,

[1] https://code.launchpad.net/~canonical-platform-qa/ubuntu-ota-tests/dbus-upgrade/+merge/253423

review: Needs Fixing

[lp:~canonical-platform-qa/ubuntu-ota-tests/thinkingoutloud-wrapper-reflash-script]

34. By Richard Huddie on 2015-03-24

Fix review comments: skip test_apply_update_noop if update available and fix flake8. Revert logging to use print which was not previously working.

[Richard Huddie (rhuddie)]

Thanks Federico, I have made those updates.

[Federico Gimenez (fgimenez)]

Ok thanks, +1

review: Approve

[Richard Huddie (rhuddie)]

Hi Chris, I confirm it works. There was an issue with test_apply_update_noop, which was causing an actual reboot and update in the selftests because the version was flashed to -1. But this test is now skipped if there is a real update available.
Thanks.

[Leo Arias (elopio)]

federico + veebers, that's +2. Top-approving...

[PS Jenkins bot (ps-jenkins)]

review: Approve (continuous-integration)

=== added file 'run-ota-tests'
--- run-ota-tests	1970-01-01 00:00:00 +0000
+++ run-ota-tests	2015-03-24 11:32:34 +0000
@@ -0,0 +1,79 @@
+#!/usr/bin/python3
+
+#
+# Ubuntu OTA Tests
+# Copyright (C) 2015 Canonical
+#
+# 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# 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/>.
+#
+
+import sys
+
+from ubuntu_ota_tests import command_line
+
+
+def main(argv):
+    """
+    Individually run each test found in debian/tests/control file.
+    For each test:
+    1) Flash the device to the required starting revision.
+    2) Run each individual ota upgrade test with adt-run and store results.
+    3) Amalgamate results into a single directory.
+
+    Usage:
+        run-ota-tests --wifi-ssid <SSID> --wifi-password <PASSWORD> [options]
+    Required:
+        --wifi-ssid         SSID of wifi network to connect device to
+        --wifi-password     Password of wifi network to connect device to
+    Optional:
+        --revision        Image revision, default: latest -1
+        --channel         Channel, default: ubuntu-touch/devel-proposed
+        --password        Device password, default: 0000
+        --output_dir      Results directory, default: /tmp/otatests-{ID}
+        --ssh-overwrite-script  SSH virt script
+                                  adb or adb-reboot-to-recovery
+                                  default adb
+        [-h]                Display help message
+
+    """
+    args = command_line.parse_cmdline_args(argv)
+    adt_script = args.ssh_overwrite_script or 'adb'
+    test_names = command_line.find_adt_test_file_names()
+
+    if not test_names:
+        print('No tests found')
+        raise RuntimeError('No tests found')
+
+    print('Found tests: {}'.format(', '.join(test_names)))
+    output_dir_path = command_line.get_output_dir(args.output_dir)
+
+    for test in test_names:
+        print('{ui}Flashing device for test: {test}'.format(
+              ui='*** ', test=test))
+        command_line.flash_device(args.revision, args.channel, args.password)
+        print('{ui}Running test: {test}'.format(ui='*** ', test=test))
+        command_line.adt_run_test(
+            test, output_dir_path, adt_script, args.password,
+            args.wifi_ssid, args.wifi_password)
+        print('Test completed.')
+
+    command_line.amalgamate_results(output_dir_path)
+
+    print('{ui}\nResults can be found: {path}\n{ui}'.format(
+        path=output_dir_path,
+        ui='='*80))
+
+
+if __name__ == '__main__':
+    main(sys.argv[1:])

=== added file 'ubuntu_ota_tests/command_line.py'
--- ubuntu_ota_tests/command_line.py	1970-01-01 00:00:00 +0000
+++ ubuntu_ota_tests/command_line.py	2015-03-24 11:32:34 +0000
@@ -0,0 +1,290 @@
+#
+# Ubuntu OTA Tests
+# Copyright (C) 2015 Canonical
+#
+# 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# 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/>.
+#
+
+import debian.deb822
+import fixtures
+import glob
+import os
+import subprocess
+
+from argparse import ArgumentParser
+from datetime import datetime
+
+
+ONE_MIN_TIMEOUT = 60
+FIVE_MIN_TIMEOUT = 300
+TEN_MIN_TIMEOUT = 600
+THIRTY_MIN_TIMEOUT = 1800
+
+
+def find_adt_test_file_names():
+    """Return list of test names for test files.
+
+    :return: List of strings of the file name of each test that is defined in
+    debian/tests/control
+
+    """
+    test_dir = get_test_directory(os.getcwd())
+    test_control = os.path.join(test_dir, 'control')
+    test_paras = get_test_stanza_strings(test_control)
+
+    all_tests = []
+    for test_string in test_paras:
+        all_tests.extend(test_string.replace(',', ' ').split())
+
+    # Perhaps we want to filter on name so we can pick what gets run now and
+    # later (i.e. self tests etc.)
+    return all_tests
+
+
+def get_test_stanza_strings(test_control_file):
+    """Parse debian/tests/control file and read out names of tests.
+
+    :param test_control_file: Path to debian/tests/control file
+    :return: List of test names to run
+
+    """
+    with open(test_control_file, 'r') as control_file:
+        control_paras = debian.deb822.Deb822.iter_paragraphs(
+            control_file.readlines(),
+            fields=['Tests']
+        )
+    # Reduce all test paragraphs into one string.
+    # and split out into individual names
+    return [p['Tests'] for p in control_paras]
+
+
+def get_test_directory(base_dir):
+    """Return the path to debian/tests folder from root directory.
+
+    :param base_dir: Root directory of project
+    :return: Path to debian/tests folder from root directory
+
+    """
+    test_dir = os.path.join(base_dir, 'debian/tests/')
+    if not os.path.exists(test_dir):
+        raise RuntimeError(
+            'Cannot find test dir. This script is designed to be run from the '
+            'branch root.')
+    return test_dir
+
+
+def get_output_dir(output_dir):
+    """Create unique output directory path and ensure it exists.
+
+    :param output_dir: Directory to use
+    :return: output_dir or unique path
+
+    """
+    path = output_dir or os.path.join(
+        '/tmp/',
+        'otatests-{date}'.format(
+            date=datetime.now().strftime('%H%M%S')))
+    if not os.path.exists(path):
+        os.makedirs(path)
+    return path
+
+
+def amalgamate_results(output_dir_path):
+    """Moves all the results from the per-run dirs into the main.
+
+    :param output_dir_path: Directory to store results
+
+    """
+    summary_files = glob.glob('{}/*/summary'.format(output_dir_path))
+    main_summary_file = os.path.join(output_dir_path, 'ota_summary')
+    with open(main_summary_file, 'w') as summary_file:
+        for summary in summary_files:
+            with open(summary, 'r') as f:
+                summary_file.write(f.read())
+
+
+def adt_run_test(test_name, output_dir, adt_script, device_password,
+                 wifi_ssid, wifi_password, timeout=TEN_MIN_TIMEOUT):
+    """Run the singular test storing the results in output directory.
+
+    :param test_name: Name of test to run
+    :param output_dir: Directory to store results
+    :param adt_script: Script for ssh connection, or 'adb'
+    :param password: Password for device
+    :param wifi_ssid: SSID of wifi network to connect device to
+    :param wifi_password: Password for wifi network to connect device to
+
+    """
+    # Results for each test in their own dir.
+    run_output_dir = os.path.join(output_dir, test_name)
+    os.mkdir(run_output_dir)
+
+    adt_command = get_adtrun_commands(
+        test_name,
+        run_output_dir,
+        adt_script,
+        device_password,
+        wifi_ssid,
+        wifi_password,
+        timeout)
+
+    return subprocess.call(adt_command)
+
+
+def get_adtrun_commands(test_name, output_dir, adt_script, password,
+                        wifi_ssid, wifi_password, timeout):
+    """Return adt-run command using specified parameters.
+
+    :param test_name: Name of test to run
+    :param output_dir: Directory to store results
+    :param adt_script: Script for ssh connection, or 'adb'
+    :param password: Password for device
+    :param wifi_ssid: SSID of wifi network to connect device to
+    :param wifi_password: Password for wifi network to connect device to
+    :param timeout: Timeout value for test exeuction
+    :return: List containing adt-run command and options
+
+    """
+    adt_run_cmds = [
+        'adt-run',
+        '-B',
+        '--testname',
+        test_name,
+        '--unbuilt-tree={}'.format(os.getcwd()),
+        '--output-dir',
+        output_dir,
+        '--setup-commands={}'.format(get_device_network_command(
+            wifi_ssid, wifi_password)),
+        '--timeout-test={}'.format(timeout)
+    ]
+
+    virt_cmds = ['ssh', '-s', adt_script, '--', '-p', password]
+
+    return adt_run_cmds + ['---'] + virt_cmds
+
+
+def flash_device(revision, channel, password):
+    """Flash the connected device with the passed revision.
+
+    :param revision: Revision of image to flash
+    :param channel: Name of channel
+    :param password: Password for device
+
+    """
+    flash_cmd = get_udflash_command(revision, channel, password)
+    print('{ui}\nFlashing device to revno: {rev} of {channel} . . .\n'.format(
+          rev=revision, channel=channel, ui='-' * 80))
+    timed_command_runner(flash_cmd, THIRTY_MIN_TIMEOUT)
+
+    print('*** Waiting for device to boot . . .')
+    wait_for_boot()
+    print('*** Device booted.')
+
+
+def get_udflash_command(revision, channel, password):
+    """Return command to flash specified revision and channel to device.
+
+    :param revision: Revision of image to flash
+    :param channel: Name of channel
+    :param password: Password for device
+    :return: List of commands
+
+    """
+    return [
+        'ubuntu-device-flash',
+        '--revision={}'.format(revision),
+        'touch',
+        '--developer-mode',
+        '--password',
+        password,
+        '--channel={}'.format(channel),
+        '--wipe'
+        ]
+
+
+def wait_for_boot():
+    """Wait for device to boot up."""
+    adb_wait = ['adb', 'wait-for-device']
+    timed_command_runner(adb_wait, timeout=TEN_MIN_TIMEOUT)
+
+
+def get_device_network_command(wifi_ssid, wifi_password):
+    """Return command to configure wifi network on device.
+
+    :param wifi_ssid: SSID of required wifi network
+    :param wifi_password: Password of required wifi network
+
+    """
+    cmd_fmt = ('nmcli device wifi connect '
+               '\'{ssid}\' password \'{password}\' 2>/dev/null || true')
+    return cmd_fmt.format(ssid=wifi_ssid, password=wifi_password)
+
+
+def timed_command_runner(cmd_list, timeout=ONE_MIN_TIMEOUT):
+    """Return command return code if successfully executes.
+
+    :param cmd_list: List of commands to execute
+    :param timeout: Timeout period for command to execute
+
+    :raises RuntimeError: if the command takes longer than timeout seconds to
+      complete
+
+    """
+    try:
+        with fixtures.Timeout(timeout, gentle=True):
+            return subprocess.call(cmd_list)
+    except fixtures.TimeoutException:
+        raise RuntimeError('Command:\n$ {command}\ntook longer than {timeout} '
+                           'seconds to run.'.format(
+                               command=' '.join(cmd_list),
+                               timeout=timeout))
+
+
+def parse_cmdline_args(argv):
+    """Return options and values read from command line."""
+    parser = _get_parser()
+    return parser.parse_args(args=argv)
+
+
+def _get_parser():
+    """Specify parameter options for running the tests."""
+    main_args = ArgumentParser()
+    main_args.add_argument(
+        '--wifi-ssid',
+        required=True,
+        help='SSID of the wi-fi network to connect device to.')
+    main_args.add_argument(
+        '--wifi-password',
+        required=True,
+        help='Password of the wi-fi network to connect device to.')
+    main_args.add_argument(
+        '--revision',
+        default='-1',
+        help='Revision of the image to flash, default is latest revision -1.')
+    main_args.add_argument(
+        '--channel',
+        default='ubuntu-touch/devel-proposed',
+        help='Channel to flash.')
+    main_args.add_argument(
+        '--password', default='0000',
+        help='Device password.')
+    main_args.add_argument(
+        '--output_dir',
+        help='Results will be placed here')
+    main_args.add_argument(
+        '--ssh-overwrite-script',
+        default=None,
+        help='Specific ssh virt script to run (i.e. adb-reboot-to-recovery)')
+
+    return main_args

=== modified file 'ubuntu_ota_tests/hooks.py'
--- ubuntu_ota_tests/hooks.py	2015-03-18 12:05:27 +0000
+++ ubuntu_ota_tests/hooks.py	2015-03-24 11:32:34 +0000
@@ -27,7 +27,6 @@
     from systemimage.reboot import BaseReboot as Base
 
 
-
 log = logging.getLogger('systemimage')
 
 

=== added file 'ubuntu_ota_tests/selftests/test_commandline.py'
--- ubuntu_ota_tests/selftests/test_commandline.py	1970-01-01 00:00:00 +0000
+++ ubuntu_ota_tests/selftests/test_commandline.py	2015-03-24 11:32:34 +0000
@@ -0,0 +1,138 @@
+#
+# Ubuntu OTA Tests
+# Copyright (C) 2015 Canonical
+#
+# 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# 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/>.
+#
+
+import os
+import unittest
+import testtools
+import tempfile
+
+from unittest.mock import patch
+from testtools import matchers
+from ubuntu_ota_tests import command_line as cmdline
+
+
+class CommandLineGetTestDirectoryTests(unittest.TestCase):
+
+    def test_raises_runtimeerror_when_dir_doesnt_exist(self):
+        base_dir = '/tmp'
+        self.assertRaises(RuntimeError, cmdline.get_test_directory, base_dir)
+
+    @patch.object(cmdline.os.path, 'exists', new=lambda x: True)
+    def test_returns_correct_path(self):
+        base_dir = '/tmp'
+        expected_output = os.path.join(base_dir, 'debian/tests/')
+        self.assertEqual(expected_output, cmdline.get_test_directory(base_dir))
+
+
+class CommandLineGetOutputDirTests(unittest.TestCase):
+
+    def test_returns_same_path(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            self.assertEqual(tmp_dir, cmdline.get_output_dir(tmp_dir))
+
+    def test_creates_dir_if_nonexistent(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            non_dir = os.path.join(tmp_dir, 'testing')
+            self.assertFalse(os.path.exists(non_dir))
+
+            cmdline.get_output_dir(non_dir)
+
+            self.assertTrue(os.path.exists(non_dir))
+
+    @patch.object(cmdline.os.path, 'exists', new=lambda x: True)
+    def test_create_custom_dir_if_passed_None(self):
+        output_path = cmdline.get_output_dir(None)
+        self.assertTrue(output_path.startswith('/tmp/otatests-'))
+
+
+class CommandLineGetFlashCommand(testtools.TestCase):
+
+    def test_revision_argument_uses_passed_value(self):
+        revision = 'revision-test'
+        channel = 'channel-test'
+        password = 'password-test'
+        command_list = cmdline.get_udflash_command(revision, channel, password)
+        self.assertThat(
+            command_list,
+            matchers.Contains('--revision=revision-test'))
+
+        self.assertThat(
+            command_list,
+            matchers.Contains('--channel=channel-test'))
+
+        self.assertThat(
+            command_list,
+            matchers.ContainsAll(['--password', 'password-test']))
+
+
+class CommandLineGetAdtrunCommand(testtools.TestCase):
+
+    def test_ssh_commands_contain_password(self):
+        adt_commands = cmdline.get_adtrun_commands(
+            '', '', '', 'test-password', '', '', '')
+
+        ssh_command_list = adt_commands[adt_commands.index('---'):]
+
+        self.assertThat(
+            ssh_command_list,
+            matchers.ContainsAll(['-p', 'test-password']))
+
+    def test_testname_argument_used(self):
+            adt_commands = cmdline.get_adtrun_commands(
+                'test-name', '', '', '', '', '', '')
+
+            self.assertThat(
+                adt_commands,
+                matchers.ContainsAll(['--testname', 'test-name']))
+
+    def test_wifi_ssid_argument_used(self):
+            adt_commands = cmdline.get_adtrun_commands(
+                '', '', '', '', 'wifi-ssid-name', '', '')
+            command_str = ' '.join(adt_commands)
+            self.assertThat(
+                command_str,
+                matchers.Contains('wifi-ssid-name'))
+
+    def test_wifi_password_argument_used(self):
+            adt_commands = cmdline.get_adtrun_commands(
+                '', '', '', '', '', 'wifi-password-value', '')
+            command_str = ' '.join(adt_commands)
+            self.assertThat(
+                command_str,
+                matchers.Contains('wifi-password-value'))
+
+    def test_timeout_test_argument_used(self):
+            adt_commands = cmdline.get_adtrun_commands(
+                '', '', '', '', '', '', '111')
+            self.assertThat(
+                adt_commands,
+                matchers.ContainsAll(['--timeout-test=111']))
+
+
+class CommandLineTimedCommandRunTests(testtools.TestCase):
+
+    def test_raises_no_exception_executed_within_timeout(self):
+        self.assertEqual(0, cmdline.timed_command_runner(['true'], timeout=1))
+
+    def test_raises_exception_executed_goes_over_timeout(self):
+        # Not sure if there is a better way to make this test quicker because
+        # it takes at least 1 sec. so we've gone from 0.004 seconds to 1.004s
+        self.assertRaises(
+            RuntimeError,
+            cmdline.timed_command_runner,
+            ['sleep', '2'], timeout=1)

=== modified file 'ubuntu_ota_tests/selftests/test_services.py'
--- ubuntu_ota_tests/selftests/test_services.py	2015-03-18 13:54:26 +0000
+++ ubuntu_ota_tests/selftests/test_services.py	2015-03-24 11:32:34 +0000
@@ -64,6 +64,8 @@
         utcnow = datetime.utcnow().replace(microsecond=0).isoformat(' ')
         self.assertLessEqual(status.last_update_date, utcnow)
 
+    @unittest.skipIf(is_update_available(),
+                     'Update available, would be applied')
     def test_apply_update_noop(self):
         # make sure that this will be a noop by cancelling any previous update
         sys_image_iface = services.get_system_image_interface()