runner.py

   1 # Copyright (C) 2010, Thomas Leonard
   2 # Visit http://0install.net for details.
   3
   4 import os, sys, logging
   5 from zeroinstall.injector import driver, model, run, arch, requirements
   6 from zeroinstall.support import tasks
   7 from reporting import format_combo
   8
   9 class VersionRestriction(model.Restriction):
  10         def __init__(self, version):
  11                 self.version = version
  12
  13         def meets_restriction(self, impl):
  14                 return impl.get_version() == self.version
  15
  16         def __repr__(self):
  17                 return "version = %s" % self.version
  18
  19 def run_tests(config, tested_iface, sels, spec):
  20         def _get_implementation_path(impl):
  21                 return impl.local_path or config.iface_cache.stores.lookup_any(impl.digests)
  22
  23         main_command = sels.commands[0] if sels.commands else None
  24
  25         root_impl = sels.selections[tested_iface.uri]
  26         assert root_impl
  27
  28         if spec.test_wrapper:
  29                 tests_dir = None
  30                 # $1 is the main executable, or the root of the package if there isn't one
  31                 # We have to add the slash because otherwise 0launch interprets the path
  32                 # relative to itself...
  33                 if main_command and main_command.path:
  34                         test_main = "/" + main_command.path
  35                 else:
  36                         test_main = "/"
  37         else:
  38                 if main_command is None:
  39                         print >>sys.stderr, "No <command> requested and no test command either!"
  40                         return "skipped"
  41
  42                 test_main = None
  43
  44                 if main_command.path is None:
  45                         tests_dir = _get_implementation_path(root_impl)
  46                 else:
  47                         main_abs = os.path.join(_get_implementation_path(root_impl), main_command.path)
  48                         if not os.path.exists(main_abs):
  49                                 print >>sys.stderr, "Test executable does not exist:", main_abs
  50                                 return "skipped"
  51
  52                         tests_dir = os.path.dirname(main_abs)
  53
  54         child = os.fork()
  55         if child:
  56                 # We are the parent
  57                 pid, status = os.waitpid(child, 0)
  58                 assert pid == child
  59                 print "Status:", hex(status)
  60                 if status == 0:
  61                         return "passed"
  62                 else:
  63                         return "failed"
  64         else:
  65                 # We are the child
  66                 try:
  67                         try:
  68                                 if spec.test_wrapper is None:
  69                                         os.chdir(tests_dir)
  70                                 run.execute_selections(sels, spec.test_args, main = test_main, wrapper = spec.test_wrapper)
  71                                 os._exit(0)
  72                         except model.SafeException, ex:
  73                                 try:
  74                                         print >>sys.stderr, unicode(ex)
  75                                 except:
  76                                         print >>sys.stderr, repr(ex)
  77                         except:
  78                                 import traceback
  79                                 traceback.print_exc()
  80                 finally:
  81                         sys.stdout.flush()
  82                         sys.stderr.flush()
  83                         os._exit(1)
  84
  85 class Results:
  86         def __init__(self, spec):
  87                 self.spec = spec
  88                 self.by_combo = {}              # { set((uri, version)) : status }
  89                 self.by_status = {              # status -> [ selections ]
  90                         'passed': [],
  91                         'skipped': [],
  92                         'failed': [],
  93                 }
  94
  95 def run_test_combinations(config, spec):
  96         r = requirements.Requirements(spec.test_iface)
  97         r.command = spec.command
  98
  99         d = driver.Driver(config = config, requirements = r)
 100         solver = d.solver
 101
 102         # Explore all combinations...
 103
 104         tested_iface = config.iface_cache.get_interface(spec.test_iface)
 105         results = Results(spec)
 106         for combo in spec.get_combos(spec.test_ifaces):
 107                 key = set()
 108                 restrictions = {}
 109                 selections = {}
 110                 for (uri, version) in combo.iteritems():
 111                         iface = config.iface_cache.get_interface(uri)
 112                         selections[iface] = version
 113
 114                         if ',' in version:
 115                                 not_before, before = [model.parse_version(v) if v != "" else None for v in version.split(',')]
 116                                 if (not_before and before) and not_before >= before:
 117                                         raise model.SafeException("Low version >= high version in %s!" % version)
 118                                 restrictions[iface] = [model.VersionRangeRestriction(before, not_before)]
 119                         else:
 120                                 model.parse_version(version)    # Check format
 121                                 restrictions[iface] = [VersionRestriction(version)]
 122                         key.add((uri, version))
 123
 124                 solver.extra_restrictions = restrictions
 125                 solve = d.solve_with_downloads()
 126                 tasks.wait_for_blocker(solve)
 127                 if not solver.ready:
 128                         logging.info("Can't select combination %s: %s", combo, solver.get_failure_reason())
 129                         result = 'skipped'
 130                         for uri, impl in solver.selections.iteritems():
 131                                 if impl is None:
 132                                         selections[uri] = selections.get(uri, None) or '?'
 133                                 else:
 134                                         selections[uri] = impl.get_version()
 135                         if not selections:
 136                                 selections = solver.get_failure_reason()
 137                 else:
 138                         selections = {}
 139                         for iface, impl in solver.selections.iteritems():
 140                                 if impl:
 141                                         version = impl.get_version()
 142                                 else:
 143                                         impl = None
 144                                 selections[iface] = version
 145                         download = d.download_uncached_implementations()
 146                         if download:
 147                                 config.handler.wait_for_blocker(download)
 148
 149                         tested_impl = solver.selections[tested_iface]
 150
 151                         print format_combo(selections)
 152
 153                         result = run_tests(config, tested_iface, solver.selections, spec)
 154
 155                 results.by_status[result].append(selections)
 156                 results.by_combo[frozenset(key)] = (result, selections)
 157
 158         return results