/usr/bin/adt-virt-xenlvm

#!/usr/bin/python
#
# adt-virt-xenlvm is part of autopkgtest
# autopkgtest is a tool for testing Debian binary packages
#
# autopkgtest is Copyright (C) 2006-2007 Canonical Ltd.
#
# 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 2 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, write to the Free Software
# Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#
# See the file CREDITS for a full list of credits information (often
# installed as /usr/share/doc/autopkgtest/CREDITS).

import sys
import os
import subprocess
import fnmatch
import signal
from optparse import OptionParser

try:
    our_base = os.environ['AUTOPKGTEST_BASE'] + '/lib'
except KeyError:
    our_base = '/usr/share/autopkgtest/python'
sys.path.insert(1, our_base)

import VirtSubproc

withholder = None
with_testbed = None
downtmp = '/root/adt-downtmp'
pauses = None
event_counters = {}


def check_pause(kind):
    if kind not in event_counters:
        event_counters[kind] = 0
    event = '%s#%d' % (kind, event_counters[kind])
    event_counters[kind] += 1
    for pattern in pauses:
        if not '#' in pattern:
            matched = kind == pattern
        else:
            matched = fnmatch.fnmatchcase(event, pattern)
        if matched:
            break
    if not matched:
        return
    print >> sys.stderr, ('----- adt-virt-xenlvm pause (event=%s),'
                          ' stopping -----' % event)
    os.kill(0, signal.SIGSTOP)


def parse_args():
    global with_testbed, pauses

    usage = '%prog <options> [-- <adt-xenlvm options>]'
    parser = OptionParser(usage=usage)

    parser.add_option('-r', '--gain-root')
    parser.add_option('-d', '--debug', action='store_true')
    parser.add_option('--userv', action='store_true')
    parser.add_option('--distro')
    parser.add_option('--nominum')
    parser.add_option('--pause', default='')

    (opts, xlargs) = parser.parse_args()
    VirtSubproc.debuglevel = opts.debug
    xargs_userv = []
    xargs_direct = []

    for k in ['distro', 'nominum']:
        v = getattr(opts, k)
        if v is None:
            continue
        xargs_direct.append('--%s=%s' % (k, v))
        xargs_userv.append('-D%s=%s' % (k, v))

    if not opts.userv:
        if opts.gain_root is None:
            gain_root = []
        else:
            gain_root = opts.gain_root.split()
        with_testbed = (gain_root + ['adt-xenlvm-with-testbed'] +
                        xargs_direct + xlargs +
                        ['--', 'sh', '-ec', 'echo y; exec cat'])
        VirtSubproc.down = (gain_root + ['adt-xenlvm-on-testbed'] +
                            xargs_direct + xlargs + ['--'])
        VirtSubproc.downkind = 'shstring'
    else:
        if opts.gain_root:
            parser.error('--userv and --gain-root are not compatible')
        basis = (['userv'] + xargs_userv + xlargs +
                 ['root', 'adt-xenlvm-testbed'])
        with_testbed = basis + ['with']
        get_down = subprocess.Popen(basis + ['pon0'],
                                    stdin=file('/dev/null'),
                                    stdout=subprocess.PIPE,
                                    stderr=None)
        (pon0, _) = get_down.communicate()
        if get_down.returncode:
            VirtSubproc.bomb('failed to check userv service provision'
                             ' and subcommand details (code=%d)' %
                             get_down.returncode)
        VirtSubproc.down = pon0.split('\0')
        VirtSubproc.downkind = 'auxverb'
    pauses = opts.pause.split(',')


def do_open():
    global withholder
    assert(withholder is None)
    withholder = subprocess.Popen(
        with_testbed,
        stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    l = withholder.stdout.readline(2)
    rc = withholder.poll()
    if rc is not None:
        withholder.stdin.close()
        withholder.stdout.close()
        withholder = None
        VirtSubproc.bomb('with-testbed failed, code %d' % rc)
    if l != 'y\n':
        VirtSubproc.bomb("with-testbed sh gave wrong output `%s', not `l'"
                         % l.rstrip('\n'))


def do_close():
    global withholder
    check_pause('close')
    withholder.stdin.close()
    withholder.stdout.close()
    rc = withholder.wait()
    withholder = None
    if rc:
        VirtSubproc.bomb('with-testbed failed when closing/reverting,'
                         ' code %d' % rc)


def hook_forked_inchild():
    if withholder is not None:
        withholder.stdin.close()
        withholder.stdout.close()


def hook_open():
    hook_cleanup()
    return do_open()


def hook_downtmp():
    VirtSubproc.execute('mkdir %s' % downtmp, downp=True)
    return downtmp


def hook_revert():
    check_pause('revert')
    do_close()
    return do_open()


def hook_cleanup():
    check_pause('cleanup')
    global withholder
    if withholder is not None:
        do_close()


def hook_capabilities():
    return ['revert', 'root-on-testbed', 'suggested-normal-user=adtxenu']

parse_args()
VirtSubproc.main()
autopkgtest 2.14.1 / usr / bin / adt-virt-xenlvm
