forked from pyinfra-dev/pyinfra
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_api_deploys.py
124 lines (95 loc) · 4.1 KB
/
test_api_deploys.py
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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
from pyinfra.api import Config, State, StringCommand
from pyinfra.api.connect import connect_all, disconnect_all
from pyinfra.api.deploy import add_deploy, deploy
from pyinfra.api.operations import run_ops
from pyinfra.operations import server
from ..paramiko_util import PatchSSHTestCase
from ..util import make_inventory
class TestDeploysApi(PatchSSHTestCase):
def test_deploy(self):
inventory = make_inventory()
somehost = inventory.get_host("somehost")
anotherhost = inventory.get_host("anotherhost")
state = State(inventory, Config())
# Enable printing on this test to catch any exceptions in the formatting
state.print_output = True
state.print_input = True
state.print_fact_info = True
state.print_noop_info = True
connect_all(state)
@deploy()
def test_deploy(state=None, host=None):
server.shell(commands=["echo first command"])
server.shell(commands=["echo second command"])
add_deploy(state, test_deploy)
op_order = state.get_op_order()
# Ensure we have an op
assert len(op_order) == 2
# Ensure run ops works
run_ops(state)
first_op_hash = op_order[0]
assert state.op_meta[first_op_hash].names == {"test_deploy | server.shell"}
assert state.ops[somehost][first_op_hash].operation_meta._commands == [
StringCommand("echo first command"),
]
assert state.ops[anotherhost][first_op_hash].operation_meta._commands == [
StringCommand("echo first command"),
]
second_op_hash = op_order[1]
assert state.op_meta[second_op_hash].names == {"test_deploy | server.shell"}
assert state.ops[somehost][second_op_hash].operation_meta._commands == [
StringCommand("echo second command"),
]
assert state.ops[anotherhost][second_op_hash].operation_meta._commands == [
StringCommand("echo second command"),
]
# Ensure ops completed OK
assert state.results[somehost].success_ops == 2
assert state.results[somehost].ops == 2
assert state.results[anotherhost].success_ops == 2
assert state.results[anotherhost].ops == 2
# And w/o errors
assert state.results[somehost].error_ops == 0
assert state.results[anotherhost].error_ops == 0
disconnect_all(state)
def test_nested_deploy(self):
inventory = make_inventory()
somehost = inventory.get_host("somehost")
state = State(inventory, Config())
# Enable printing on this test to catch any exceptions in the formatting
state.print_output = True
state.print_input = True
state.print_fact_info = True
state.print_noop_info = True
connect_all(state)
@deploy()
def test_nested_deploy():
server.shell(commands=["echo nested command"])
@deploy()
def test_deploy():
server.shell(commands=["echo first command"])
test_nested_deploy()
server.shell(commands=["echo second command"])
add_deploy(state, test_deploy)
op_order = state.get_op_order()
# Ensure we have an op
assert len(op_order) == 3
# Ensure run ops works
run_ops(state)
first_op_hash = op_order[0]
assert state.op_meta[first_op_hash].names == {"test_deploy | server.shell"}
assert state.ops[somehost][first_op_hash].operation_meta._commands == [
StringCommand("echo first command"),
]
second_op_hash = op_order[1]
assert state.op_meta[second_op_hash].names == {
"test_deploy | test_nested_deploy | server.shell",
}
assert state.ops[somehost][second_op_hash].operation_meta._commands == [
StringCommand("echo nested command"),
]
third_op_hash = op_order[2]
assert state.op_meta[third_op_hash].names == {"test_deploy | server.shell"}
assert state.ops[somehost][third_op_hash].operation_meta._commands == [
StringCommand("echo second command"),
]