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