michael@0: # This Source Code Form is subject to the terms of the Mozilla Public
michael@0: # License, v. 2.0. If a copy of the MPL was not distributed with this
michael@0: # file, You can obtain one at http://mozilla.org/MPL/2.0/.
michael@0: 
michael@0: import json
michael@0: import os
michael@0: import subprocess
michael@0: import sys
michael@0: import tempfile
michael@0: import threading
michael@0: import zipfile
michael@0: 
michael@0: from ConfigParser import ConfigParser
michael@0: 
michael@0: this_dir = os.path.abspath(os.path.dirname(__file__))
michael@0: marionette_dir = os.path.dirname(this_dir)
michael@0: marionette_client_dir = os.path.join(marionette_dir, 'client', 'marionette')
michael@0: 
michael@0: def find_b2g():
michael@0:     sys.path.append(marionette_client_dir)
michael@0:     from b2gbuild import B2GBuild
michael@0:     return B2GBuild()
michael@0: 
michael@0: class DictObject(dict):
michael@0:     def __getattr__(self, item):
michael@0:         try:
michael@0:             return self.__getitem__(item)
michael@0:         except KeyError:
michael@0:             raise AttributeError(item)
michael@0: 
michael@0:     def __getitem__(self, item):
michael@0:         value = dict.__getitem__(self, item)
michael@0:         if isinstance(value, dict):
michael@0:             return DictObject(value)
michael@0:         return value
michael@0: 
michael@0: class SmokeTestError(Exception):
michael@0:     pass
michael@0: 
michael@0: class SmokeTestConfigError(SmokeTestError):
michael@0:     def __init__(self, message):
michael@0:         SmokeTestError.__init__(self, 'smoketest-config.json: ' + message)
michael@0: 
michael@0: class SmokeTestConfig(DictObject):
michael@0:     TOP_LEVEL_REQUIRED = ('devices', 'public_key', 'private_key')
michael@0:     DEVICE_REQUIRED    = ('system_fs_type', 'system_location', 'data_fs_type',
michael@0:                           'data_location', 'sdcard', 'sdcard_recovery',
michael@0:                           'serials')
michael@0: 
michael@0:     def __init__(self, build_dir):
michael@0:         self.top_dir = build_dir
michael@0:         self.build_data = {}
michael@0:         self.flash_template = None
michael@0: 
michael@0:         with open(os.path.join(build_dir, 'smoketest-config.json')) as f:
michael@0:             DictObject.__init__(self, json.loads(f.read()))
michael@0: 
michael@0:         for required in self.TOP_LEVEL_REQUIRED:
michael@0:             if required not in self:
michael@0:                 raise SmokeTestConfigError('No "%s" found' % required)
michael@0: 
michael@0:         if len(self.devices) == 0:
michael@0:             raise SmokeTestConfigError('No devices found')
michael@0: 
michael@0:         for name, device in self.devices.iteritems():
michael@0:             for required in self.DEVICE_REQUIRED:
michael@0:                 if required not in device:
michael@0:                     raise SmokeTestConfigError('No "%s" found in device "%s"' % (required, name))
michael@0: 
michael@0:     def get_build_data(self, device, build_id):
michael@0:         if device in self.build_data:
michael@0:             if build_id in self.build_data[device]:
michael@0:                 return self.build_data[device][build_id]
michael@0:         else:
michael@0:             self.build_data[device] = {}
michael@0: 
michael@0:         build_dir = os.path.join(self.top_dir, device, build_id)
michael@0:         flash_zip = os.path.join(build_dir, 'flash.zip')
michael@0:         with zipfile.ZipFile(flash_zip) as zip:
michael@0:             app_ini = ConfigParser()
michael@0:             app_ini.readfp(zip.open('system/b2g/application.ini'))
michael@0:             platform_ini = ConfigParser()
michael@0:             platform_ini.readfp(zip.open('system/b2g/platform.ini'))
michael@0: 
michael@0:         build_data = self.build_data[device][build_id] = DictObject({
michael@0:             'app_version': app_ini.get('App', 'version'),
michael@0:             'app_build_id': app_ini.get('App', 'buildid'),
michael@0:             'platform_build_id': platform_ini.get('Build', 'buildid'),
michael@0:             'platform_milestone': platform_ini.get('Build', 'milestone'),
michael@0:             'complete_mar': os.path.join(build_dir, 'complete.mar'),
michael@0:             'flash_script': os.path.join(build_dir, 'flash.sh')
michael@0:         })
michael@0: 
michael@0:         return build_data
michael@0: 
michael@0: class SmokeTestRunner(object):
michael@0:     DEVICE_TIMEOUT = 30
michael@0: 
michael@0:     def __init__(self, config, b2g, run_dir=None):
michael@0:         self.config = config
michael@0:         self.b2g = b2g
michael@0:         self.run_dir = run_dir or tempfile.mkdtemp()
michael@0: 
michael@0:         update_tools = self.b2g.import_update_tools()
michael@0:         self.b2g_config = update_tools.B2GConfig()
michael@0: 
michael@0:     def run_b2g_update_test(self, serial, testvars, tests):
michael@0:         b2g_update_test = os.path.join(marionette_client_dir,
michael@0:                                        'venv_b2g_update_test.sh')
michael@0: 
michael@0:         if not tests:
michael@0:             tests = [os.path.join(marionette_client_dir, 'tests',
michael@0:                                   'update-tests.ini')]
michael@0: 
michael@0:         args = ['bash', b2g_update_test, sys.executable,
michael@0:                 '--homedir', self.b2g.homedir,
michael@0:                 '--address', 'localhost:2828',
michael@0:                 '--type', 'b2g+smoketest',
michael@0:                 '--device', serial,
michael@0:                 '--testvars', testvars]
michael@0:         args.extend(tests)
michael@0: 
michael@0:         print ' '.join(args)
michael@0:         subprocess.check_call(args)
michael@0: 
michael@0:     def build_testvars(self, device, start_id, finish_id):
michael@0:         run_dir = os.path.join(self.run_dir, device, start_id, finish_id)
michael@0:         if not os.path.exists(run_dir):
michael@0:             os.makedirs(run_dir)
michael@0: 
michael@0:         start_data = self.config.get_build_data(device, start_id)
michael@0:         finish_data = self.config.get_build_data(device, finish_id)
michael@0: 
michael@0:         partial_mar = os.path.join(run_dir, 'partial.mar')
michael@0:         if not os.path.exists(partial_mar):
michael@0:             build_gecko_mar = os.path.join(self.b2g.update_tools,
michael@0:                                            'build-gecko-mar.py')
michael@0:             subprocess.check_call([sys.executable, build_gecko_mar,
michael@0:                                    '--from', start_data.complete_mar,
michael@0:                                    '--to', finish_data.complete_mar,
michael@0:                                    partial_mar])
michael@0:         finish_data['partial_mar'] = partial_mar
michael@0: 
michael@0:         testvars = os.path.join(run_dir, 'testvars.json')
michael@0:         if not os.path.exists(testvars):
michael@0:             open(testvars, 'w').write(json.dumps({
michael@0:                 'start': start_data,
michael@0:                 'finish': finish_data
michael@0:             }))
michael@0: 
michael@0:         return testvars
michael@0: 
michael@0:     def wait_for_device(self, device):
michael@0:         for serial in self.config.devices[device].serials:
michael@0:             proc = subprocess.Popen([self.b2g.adb_path, '-s', serial,
michael@0:                                      'wait-for-device'])
michael@0:             def wait_for_adb():
michael@0:                 proc.communicate()
michael@0: 
michael@0:             thread = threading.Thread(target=wait_for_adb)
michael@0:             thread.start()
michael@0:             thread.join(self.DEVICE_TIMEOUT)
michael@0: 
michael@0:             if thread.isAlive():
michael@0:                 print >>sys.stderr, '%s device %s is not recognized by ADB, ' \
michael@0:                                     'trying next device' % (device, serial)
michael@0:                 proc.kill()
michael@0:                 thread.join()
michael@0:                 continue
michael@0: 
michael@0:             return serial
michael@0:         return None
michael@0: 
michael@0:     def run_smoketests_for_device(self, device, start_id, finish_id, tests):
michael@0:         testvars = self.build_testvars(device, start_id, finish_id)
michael@0:         serial = self.wait_for_device(device)
michael@0:         if not serial:
michael@0:             raise SmokeTestError('No connected serials for device "%s" could ' \
michael@0:                                  'be found' % device)
michael@0: 
michael@0:         try:
michael@0:             self.run_b2g_update_test(serial, testvars, tests)
michael@0:         except subprocess.CalledProcessError:
michael@0:             print >>sys.stderr, 'SMOKETEST-FAIL | START=%s | FINISH=%s | ' \
michael@0:                                 'DEVICE=%s/%s | %s' % (start_id, finish_id,
michael@0:                                                        device, serial, testvars)
michael@0: 
michael@0:     def run_smoketests(self, build_ids, tests):
michael@0:         build_ids.sort()
michael@0: 
michael@0:         latest_build_id = build_ids.pop(-1)
michael@0:         for build_id in build_ids:
michael@0:             for device in self.config.devices:
michael@0:                 self.run_smoketests_for_device(device, build_id,
michael@0:                                                latest_build_id, tests)