File: test_parallel.py

package info (click to toggle)
fabric 1.14.0-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, buster, sid
  • size: 1,240 kB
  • sloc: python: 7,363; makefile: 10
file content (93 lines) | stat: -rw-r--r-- 2,606 bytes parent folder | download | duplicates (4)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
from __future__ import with_statement

from fabric.api import run, parallel, env, hide, execute, settings

from utils import FabricTest, eq_, aborts, mock_streams
from server import server, RESPONSES, USER, HOST, PORT

# TODO: move this into test_tasks? meh.

class OhNoesException(Exception): pass


class TestParallel(FabricTest):
    @server()
    @parallel
    def test_parallel(self):
        """
        Want to do a simple call and respond
        """
        env.pool_size = 10
        cmd = "ls /simple"
        with hide('everything'):
            eq_(run(cmd), RESPONSES[cmd])

    @server(port=2200)
    @server(port=2201)
    def test_env_host_no_user_or_port(self):
        """
        Ensure env.host doesn't get user/port parts when parallel
        """
        @parallel
        def _task():
            run("ls /simple")
            assert USER not in env.host
            assert str(PORT) not in env.host

        host_string = '%s@%s:%%s' % (USER, HOST)
        with hide('everything'):
            execute(_task, hosts=[host_string % 2200, host_string % 2201])

    @server(port=2200)
    @server(port=2201)
    @aborts
    def test_parallel_failures_abort(self):
        with hide('everything'):
            host1 = '127.0.0.1:2200'
            host2 = '127.0.0.1:2201'

            @parallel
            def mytask():
                run("ls /")
                if env.host_string == host2:
                    raise OhNoesException
            
            execute(mytask, hosts=[host1, host2])

    @server(port=2200)
    @server(port=2201)
    @mock_streams('stderr') # To hide the traceback for now
    def test_parallel_failures_honor_warn_only(self):
        with hide('everything'):
            host1 = '127.0.0.1:2200'
            host2 = '127.0.0.1:2201'

            @parallel
            def mytask():
                run("ls /")
                if env.host_string == host2:
                    raise OhNoesException

            with settings(warn_only=True):
                result = execute(mytask, hosts=[host1, host2])
            eq_(result[host1], None)
            assert isinstance(result[host2], OhNoesException)


    @server(port=2200)
    @server(port=2201)
    def test_parallel_implies_linewise(self):
        host1 = '127.0.0.1:2200'
        host2 = '127.0.0.1:2201'

        assert not env.linewise

        @parallel
        def mytask():
            run("ls /")
            return env.linewise

        with hide('everything'):
            result = execute(mytask, hosts=[host1, host2])
        eq_(result[host1], True)
        eq_(result[host2], True)