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
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')
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
25 def testCommandBindings(self
):
26 p
= policy
.Policy(command_feed
, config
= self
.config
)
27 self
.config
.handler
.wait_for_blocker(p
.solve_with_downloads())
28 old_stdout
= sys
.stdout
30 sys
.stdout
= StringIO()
31 run
.execute_selections(p
.solver
.selections
, [], main
= 'runner', dry_run
= True, stores
= self
.config
.stores
)
33 sys
.stdout
= old_stdout
34 assert 'local' in os
.environ
['LOCAL'], os
.environ
['LOCAL']
36 def testAbsMain(self
):
37 p
= policy
.Policy(command_feed
, config
= self
.config
)
38 self
.config
.handler
.wait_for_blocker(p
.solve_with_downloads())
40 old_stdout
= sys
.stdout
42 sys
.stdout
= StringIO()
43 run
.execute_selections(p
.solver
.selections
, [], main
= '/runnable/runner', dry_run
= True, stores
= self
.config
.stores
)
45 sys
.stdout
= old_stdout
48 old_stdout
= sys
.stdout
50 sys
.stdout
= StringIO()
51 run
.execute_selections(p
.solver
.selections
, [], main
= '/runnable/not-there', dry_run
= True, stores
= self
.config
.stores
)
53 sys
.stdout
= old_stdout
54 except SafeException
as ex
:
55 assert 'not-there' in unicode(ex
)
58 p
= policy
.Policy(runnable
, config
= self
.config
)
59 self
.config
.handler
.wait_for_blocker(p
.solve_with_downloads())
60 old_stdout
= sys
.stdout
62 sys
.stdout
= StringIO()
63 run
.execute_selections(p
.solver
.selections
, [], dry_run
= True, stores
= self
.config
.stores
)
64 out
= sys
.stdout
.getvalue()
66 sys
.stdout
= old_stdout
67 assert 'runner-arg' in out
, out
69 def testWrapper(self
):
70 p
= policy
.Policy(runnable
, config
= self
.config
)
71 self
.config
.handler
.wait_for_blocker(p
.solve_with_downloads())
72 old_stdout
= sys
.stdout
74 sys
.stdout
= StringIO()
75 run
.execute_selections(p
.solver
.selections
, [], wrapper
= 'echo', dry_run
= True, stores
= self
.config
.stores
)
76 out
= sys
.stdout
.getvalue()
78 sys
.stdout
= old_stdout
79 assert '/bin/sh -c echo "$@"' in out
, out
80 assert 'runner-arg' in out
, out
81 assert 'script' in out
, out
83 def testRecursive(self
):
84 child
= subprocess
.Popen([local_0launch
, '--', recursive_runner
, 'user-arg'], stdout
= subprocess
.PIPE
)
85 stdout
, _
= child
.communicate()
86 assert 'Runner: script=A test script: args=command-arg -- arg-for-runnable recursive-arg -- user-arg' in stdout
, stdout
88 def testExecutable(self
):
89 child
= subprocess
.Popen([local_0launch
, '--', runexec
, 'user-arg-run'], stdout
= subprocess
.PIPE
)
90 stdout
, _
= child
.communicate()
91 assert 'Runner: script=A test script: args=foo-arg -- var user-arg-run' in stdout
, stdout
92 assert 'Runner: script=A test script: args=command-arg -- path user-arg-run' in stdout
, stdout
94 if __name__
== '__main__':