test_image.py revision fec1349a3e70eb41dce8bc07a8c63563d23d64b2 
1# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5"""Module for discovering Chrome OS test images and payloads."""
6
7import logging
8import os
9import re
10
11import common
12from autotest_lib.client.common_lib import global_config
13
14try:
15  from devserver import gsutil_util
16except ImportError:
17  # Make this easy for users to automatically import the devserver if not found.
18  from autotest_lib.utils import build_externals, external_packages
19  tot = external_packages.find_top_of_autotest_tree()
20  install_dir = os.path.join(tot, build_externals.INSTALL_DIR)
21  build_externals.build_and_install_packages(
22      [external_packages.DevServerRepo()], install_dir)
23  from devserver import gsutil_util
24
25
26# A string indicating a zip-file boundary within a URI path. This string must
27# end with a '/', in order for standard basename code to work correctly for
28# zip-encapsulated paths.
29ZIPFILE_BOUNDARY = '//'
30ARCHIVE_URL_FORMAT = '%(archive_base)s/%(board)s-release/%(branch)s-%(release)s'
31
32
33class TestImageError(BaseException):
34    """Raised on any error in this module."""
35    pass
36
37
38def get_archive_url(board, branch, release):
39    """Returns the gs archive_url for the respective arguments.
40
41    @param board: the platform name (string)
42    @param release: the release version (string), without milestone and
43                    attempt/build counters
44    @param branch: the release's branch name
45    """
46    archive_base = global_config.global_config.get_config_value(
47            'CROS', 'image_storage_server')
48    archive_base = archive_base.rstrip('/') # Remove any trailing /'s.
49
50    # TODO(garnold) adjustment to -he variant board names; should be removed
51    # once we switch to using artifacts from gs://chromeos-images/
52    # (see chromium-os:38222)
53    board = re.sub('-he$', '_he', board)
54    return ARCHIVE_URL_FORMAT % dict(
55            archive_base=archive_base, board=board, branch=branch,
56            release=release)
57
58
59def gs_ls(pattern, archive_url, single):
60    """Returns a list of URIs that match a given pattern.
61
62    @param pattern: a regexp pattern to match (feeds into re.match).
63    @param archive_url: the gs uri where to search (see ARCHIVE_URL_FORMAT).
64    @param single: if true, expect a single match and return it.
65
66    @return A list of URIs (possibly an empty list).
67
68    """
69    try:
70        logging.debug('Searching for pattern %s from url %s', pattern,
71                      archive_url)
72        uri_list = gsutil_util.GetGSNamesWithWait(
73                pattern, archive_url, err_str=__name__, single_item=single,
74                timeout=1)
75        # Convert to the format our clients expect (full archive path).
76        if uri_list:
77            return ['/'.join([archive_url, u]) for u in uri_list]
78
79        return []
80    except gsutil_util.PatternNotSpecific as e:
81        raise TestImageError(str(e))
82    except gsutil_util.GSUtilError:
83        return []
84
85
86def find_payload_uri(archive_url, delta=False, single=False):
87    """Finds test payloads corresponding to a given board/release.
88
89    @param archive_url: Archive_url directory to find the payload.
90    @param delta: if true, seek delta payloads to the given release
91    @param single: if true, expect a single match and return it, otherwise
92           None
93
94    @return A (possibly empty) list of URIs, or a single (possibly None) URI if
95            |single| is True.
96
97    @raise TestImageError if an error has occurred.
98
99    """
100    if delta:
101        pattern = '.*_delta_.*'
102    else:
103        pattern = '.*_full_.*'
104
105    payload_uri_list = gs_ls(pattern, archive_url, single)
106    if not payload_uri_list:
107        return None if single else []
108
109    return payload_uri_list[0] if single else payload_uri_list
110
111
112def find_image_uri(archive_url):
113    """Returns a URI to a test image.
114
115    @param archive_url: archive_url directory to find the payload.
116
117    @return A URI to the desired image if found, None otherwise. It will most
118            likely be a file inside an image archive (image.zip), in which case
119            we'll be using ZIPFILE_BOUNDARY ('//') to denote a zip-encapsulated
120            file, for example:
121            gs://chromeos-image-archive/.../image.zip//chromiumos_test_image.bin
122
123    @raise TestImageError if an error has occurred.
124
125    """
126    image_archive = gs_ls('image.zip', archive_url, single=True)
127    if not image_archive:
128        return None
129
130    return (image_archive[0] + ZIPFILE_BOUNDARY + 'chromiumos_test_image.bin')
131