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