tests/testrun.py

   1 #!/usr/bin/env python
   2 from basetest import BaseTest
   3 import os, sys, subprocess
   4 import unittest
   5 from StringIO import StringIO
   6
   7 sys.path.insert(0, '..')
   8
   9 from zeroinstall.injector import policy, run
  10 from zeroinstall import SafeException
  11
  12 mydir = os.path.abspath(os.path.dirname(__file__))
  13 local_0launch = os.path.join(os.path.dirname(mydir), '0launch')
  14 runnable = os.path.join(mydir, 'runnable', 'Runnable.xml')
  15 recursive_runner = os.path.join(mydir, 'runnable', 'RecursiveRunner.xml')
  16 command_feed = os.path.join(mydir, 'Command.xml')
  17
  18 class TestRun(BaseTest):
  19         def testRunnable(self):
  20                 child = subprocess.Popen([local_0launch, '--', runnable, 'user-arg'], stdout = subprocess.PIPE)
  21                 stdout, _ = child.communicate()
  22                 assert 'Runner: script=A test script: args=command-arg -- user-arg' in stdout, stdout
  23
  24         def testCommandBindings(self):
  25                 p = policy.Policy(command_feed, config = self.config)
  26                 self.config.handler.wait_for_blocker(p.solve_with_downloads())
  27                 old_stdout = sys.stdout
  28                 try:
  29                         sys.stdout = StringIO()
  30                         run.execute_selections(p.solver.selections, [], main = 'runner', dry_run = True, stores = self.config.stores)
  31                 finally:
  32                         sys.stdout = old_stdout
  33                 assert 'local' in os.environ['LOCAL'], os.environ['LOCAL']
  34
  35         def testAbsMain(self):
  36                 p = policy.Policy(command_feed, config = self.config)
  37                 self.config.handler.wait_for_blocker(p.solve_with_downloads())
  38
  39                 old_stdout = sys.stdout
  40                 try:
  41                         sys.stdout = StringIO()
  42                         run.execute_selections(p.solver.selections, [], main = '/runnable/runner', dry_run = True, stores = self.config.stores)
  43                 finally:
  44                         sys.stdout = old_stdout
  45
  46                 try:
  47                         old_stdout = sys.stdout
  48                         try:
  49                                 sys.stdout = StringIO()
  50                                 run.execute_selections(p.solver.selections, [], main = '/runnable/not-there', dry_run = True, stores = self.config.stores)
  51                         finally:
  52                                 sys.stdout = old_stdout
  53                 except SafeException, ex:
  54                         assert 'not-there' in unicode(ex)
  55
  56         def testArgs(self):
  57                 p = policy.Policy(runnable, config = self.config)
  58                 self.config.handler.wait_for_blocker(p.solve_with_downloads())
  59                 old_stdout = sys.stdout
  60                 try:
  61                         sys.stdout = StringIO()
  62                         run.execute_selections(p.solver.selections, [], dry_run = True, stores = self.config.stores)
  63                         out = sys.stdout.getvalue()
  64                 finally:
  65                         sys.stdout = old_stdout
  66                 assert 'runner-arg' in out, out
  67
  68         def testWrapper(self):
  69                 p = policy.Policy(runnable, config = self.config)
  70                 self.config.handler.wait_for_blocker(p.solve_with_downloads())
  71                 old_stdout = sys.stdout
  72                 try:
  73                         sys.stdout = StringIO()
  74                         run.execute_selections(p.solver.selections, [], wrapper = 'echo', dry_run = True, stores = self.config.stores)
  75                         out = sys.stdout.getvalue()
  76                 finally:
  77                         sys.stdout = old_stdout
  78                 assert '/bin/sh -c echo "$@"' in out, out
  79                 assert 'runner-arg' in out, out
  80                 assert 'script' in out, out
  81
  82         def testRecursive(self):
  83                 child = subprocess.Popen([local_0launch, '--', recursive_runner, 'user-arg'], stdout = subprocess.PIPE)
  84                 stdout, _ = child.communicate()
  85                 assert 'Runner: script=A test script: args=command-arg -- arg-for-runnable recursive-arg -- user-arg' in stdout, stdout
  86
  87 if __name__ == '__main__':
  88         unittest.main()