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
, handler
, model
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 recursive_runner
= os
.path
.join(mydir
, 'runnable', 'RecursiveRunner.xml')
16 command_feed
= os
.path
.join(mydir
, 'Command.xml')
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
24 def testCommandBindings(self
):
26 p
= policy
.Policy(command_feed
, handler
= h
)
27 h
.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)
33 sys
.stdout
= old_stdout
34 assert 'local' in os
.environ
['LOCAL'], os
.environ
['LOCAL']
36 def testAbsMain(self
):
38 p
= policy
.Policy(command_feed
, handler
= h
)
39 h
.wait_for_blocker(p
.solve_with_downloads())
41 old_stdout
= sys
.stdout
43 sys
.stdout
= StringIO()
44 run
.execute_selections(p
.solver
.selections
, [], main
= '/runnable/runner', dry_run
= True)
46 sys
.stdout
= old_stdout
49 old_stdout
= sys
.stdout
51 sys
.stdout
= StringIO()
52 run
.execute_selections(p
.solver
.selections
, [], main
= '/runnable/not-there', dry_run
= True)
54 sys
.stdout
= old_stdout
55 except SafeException
, ex
:
56 assert 'not-there' in unicode(ex
)
60 p
= policy
.Policy(runnable
, handler
= h
)
61 h
.wait_for_blocker(p
.solve_with_downloads())
62 old_stdout
= sys
.stdout
64 sys
.stdout
= StringIO()
65 run
.execute_selections(p
.solver
.selections
, [], dry_run
= True)
66 out
= sys
.stdout
.getvalue()
68 sys
.stdout
= old_stdout
69 assert 'runner-arg' in out
, out
71 def testWrapper(self
):
73 p
= policy
.Policy(runnable
, handler
= h
)
74 h
.wait_for_blocker(p
.solve_with_downloads())
75 old_stdout
= sys
.stdout
77 sys
.stdout
= StringIO()
78 run
.execute_selections(p
.solver
.selections
, [], wrapper
= 'echo', dry_run
= True)
79 out
= sys
.stdout
.getvalue()
81 sys
.stdout
= old_stdout
82 assert '/bin/sh -c echo "$@"' in out
, out
83 assert 'runner-arg' in out
, out
84 assert 'script' in out
, out
86 def testRecursive(self
):
87 child
= subprocess
.Popen([local_0launch
, '--', recursive_runner
, 'user-arg'], stdout
= subprocess
.PIPE
)
88 stdout
, _
= child
.communicate()
89 assert 'Runner: script=A test script: args=command-arg -- arg-for-runnable recursive-arg -- user-arg' in stdout
, stdout
91 if __name__
== '__main__':