2 from basetest
import BaseTest
3 import os
, sys
, subprocess
5 from StringIO
import StringIO
7 sys
.path
.insert(0, '..')
9 from zeroinstall
.injector
import policy
, run
, namespaces
10 from zeroinstall
import SafeException
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')
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
26 def testCommandBindings(self
):
27 if 'SELF_COMMAND' in os
.environ
:
28 del os
.environ
['SELF_COMMAND']
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
34 sys
.stdout
= StringIO()
35 run
.execute_selections(p
.solver
.selections
, [], main
= 'runnable/go.sh', dry_run
= True, stores
= self
.config
.stores
)
37 sys
.stdout
= old_stdout
38 assert 'local' in os
.environ
['LOCAL'], os
.environ
['LOCAL']
39 assert 'SELF_COMMAND' in os
.environ
41 def testAbsMain(self
):
42 p
= policy
.Policy(command_feed
, config
= self
.config
)
43 self
.config
.handler
.wait_for_blocker(p
.solve_with_downloads())
45 old_stdout
= sys
.stdout
47 sys
.stdout
= StringIO()
48 run
.execute_selections(p
.solver
.selections
, [], main
= '/runnable/runner', dry_run
= True, stores
= self
.config
.stores
)
50 sys
.stdout
= old_stdout
53 old_stdout
= sys
.stdout
55 sys
.stdout
= StringIO()
56 run
.execute_selections(p
.solver
.selections
, [], main
= '/runnable/not-there', dry_run
= True, stores
= self
.config
.stores
)
58 sys
.stdout
= old_stdout
59 except SafeException
as ex
:
60 assert 'not-there' in unicode(ex
)
63 p
= policy
.Policy(runnable
, config
= self
.config
)
64 self
.config
.handler
.wait_for_blocker(p
.solve_with_downloads())
65 old_stdout
= sys
.stdout
67 sys
.stdout
= StringIO()
68 run
.execute_selections(p
.solver
.selections
, [], dry_run
= True, stores
= self
.config
.stores
)
69 out
= sys
.stdout
.getvalue()
71 sys
.stdout
= old_stdout
72 assert 'runner-arg' in out
, out
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
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()
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
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
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
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
:
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
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
118 if __name__
== '__main__':