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, namespaces
  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 runexec = os.path.join(mydir, 'runnable', 'RunExec.xml')
  16 recursive_runner = os.path.join(mydir, 'runnable', 'RecursiveRunner.xml')
  17 command_feed = os.path.join(mydir, 'Command.xml')
  18 package_selections = os.path.join(mydir, 'package-selection.xml')
  19
  20 class TestRun(BaseTest):
  21         def testRunnable(self):
  22                 child = subprocess.Popen([local_0launch, '--', runnable, 'user-arg'], stdout = subprocess.PIPE)
  23                 stdout, _ = child.communicate()
  24                 assert 'Runner: script=A test script: args=command-arg -- user-arg' in stdout, stdout
  25
  26         def testCommandBindings(self):
  27                 if 'SELF_COMMAND' in os.environ:
  28                         del os.environ['SELF_COMMAND']
  29
  30                 p = policy.Policy(command_feed, config = self.config)
  31                 self.config.handler.wait_for_blocker(p.solve_with_downloads())
  32                 old_stdout = sys.stdout
  33                 try:
  34                         sys.stdout = StringIO()
  35                         run.execute_selections(p.solver.selections, [], main = 'runnable/go.sh', dry_run = True, stores = self.config.stores)
  36                 finally:
  37                         sys.stdout = old_stdout
  38                 assert 'local' in os.environ['LOCAL'], os.environ['LOCAL']
  39                 assert 'SELF_COMMAND' in os.environ
  40
  41         def testAbsMain(self):
  42                 p = policy.Policy(command_feed, config = self.config)
  43                 self.config.handler.wait_for_blocker(p.solve_with_downloads())
  44
  45                 old_stdout = sys.stdout
  46                 try:
  47                         sys.stdout = StringIO()
  48                         run.execute_selections(p.solver.selections, [], main = '/runnable/runner', dry_run = True, stores = self.config.stores)
  49                 finally:
  50                         sys.stdout = old_stdout
  51
  52                 try:
  53                         old_stdout = sys.stdout
  54                         try:
  55                                 sys.stdout = StringIO()
  56                                 run.execute_selections(p.solver.selections, [], main = '/runnable/not-there', dry_run = True, stores = self.config.stores)
  57                         finally:
  58                                 sys.stdout = old_stdout
  59                 except SafeException as ex:
  60                         assert 'not-there' in unicode(ex)
  61
  62         def testArgs(self):
  63                 p = policy.Policy(runnable, config = self.config)
  64                 self.config.handler.wait_for_blocker(p.solve_with_downloads())
  65                 old_stdout = sys.stdout
  66                 try:
  67                         sys.stdout = StringIO()
  68                         run.execute_selections(p.solver.selections, [], dry_run = True, stores = self.config.stores)
  69                         out = sys.stdout.getvalue()
  70                 finally:
  71                         sys.stdout = old_stdout
  72                 assert 'runner-arg' in out, out
  73
  74         def testWrapper(self):
  75                 p = policy.Policy(runnable, config = self.config)
  76                 self.config.handler.wait_for_blocker(p.solve_with_downloads())
  77                 old_stdout = sys.stdout
  78                 try:
  79                         sys.stdout = StringIO()
  80                         run.execute_selections(p.solver.selections, [], wrapper = 'echo', dry_run = True, stores = self.config.stores)
  81                         out = sys.stdout.getvalue()
  82                 finally:
  83                         sys.stdout = old_stdout
  84                 assert '/bin/sh -c echo "$@"' in out, out
  85                 assert 'runner-arg' in out, out
  86                 assert 'script' in out, out
  87
  88         def testRecursive(self):
  89                 child = subprocess.Popen([local_0launch, '--', recursive_runner, 'user-arg'], stdout = subprocess.PIPE)
  90                 stdout, _ = child.communicate()
  91                 assert 'Runner: script=A test script: args=command-arg -- arg-for-runnable recursive-arg -- user-arg' in stdout, stdout
  92
  93         def testExecutable(self):
  94                 child = subprocess.Popen([local_0launch, '--', runexec, 'user-arg-run'], stdout = subprocess.PIPE)
  95                 stdout, _ = child.communicate()
  96                 assert 'Runner: script=A test script: args=foo-arg -- var user-arg-run' in stdout, stdout
  97                 assert 'Runner: script=A test script: args=command-arg -- path user-arg-run' in stdout, stdout
  98
  99                 # Check runenv.py is updated correctly
 100                 from zeroinstall.support import basedir
 101                 runenv = basedir.load_first_cache(namespaces.config_site, namespaces.config_prog, 'runenv.py')
 102                 os.chmod(runenv, 0700)
 103                 with open(runenv, 'wb') as s:
 104                         s.write('#!/\n')
 105
 106                 child = subprocess.Popen([local_0launch, '--', runexec, 'user-arg-run'], stdout = subprocess.PIPE)
 107                 stdout, _ = child.communicate()
 108                 assert 'Runner: script=A test script: args=foo-arg -- var user-arg-run' in stdout, stdout
 109                 assert 'Runner: script=A test script: args=command-arg -- path user-arg-run' in stdout, stdout
 110
 111         def testRunPackage(self):
 112                 if 'TEST' in os.environ:
 113                         del os.environ['TEST']
 114                 child = subprocess.Popen([local_0launch, '--wrapper', 'echo $TEST #', '--', package_selections], stdout = subprocess.PIPE)
 115                 stdout, _ = child.communicate()
 116                 assert stdout.strip() == 'OK', stdout
 117
 118 if __name__ == '__main__':
 119         unittest.main()