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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
#!/usr/bin/python
from __future__ import print_function
from argparse import ArgumentParser
from status import StatusTester
from deploy_stack import (
BootstrapManager,
)
from utility import (
add_basic_testing_arguments,
local_charm_path,
)
__metaclass__ = type
def run_complete_status(client, status):
"""run the complete set of tests possible for any StatusParser.
:param client: python juju client.
:type client: jujupy.EnvJujuClient
:param status: a BaseStatusParser
:type status: BaseStatusParser
"""
status.s.assert_machines_len(2)
status.s.assert_machines_ids(("0", "1"))
juju_status = client.get_status()
for name, machine in juju_status.iter_machines(False):
status.s.assert_machine_agent_state(name, "started")
status.s.assert_machine_agent_version(name,
machine.get("agent-version"))
status.s.assert_machine_dns_name(name, machine.get("dns-name"))
status.s.assert_machine_instance_id(name,
machine.get("instance-id"))
status.s.assert_machine_series(name, machine.get("series"))
status.s.assert_machine_hardware(name, machine.get("hardware"))
state_server = client.get_controller_member_status(machine)
if state_server:
status.s.assert_machine_member_status(name, "has-vote")
status.s.assert_service_charm("statusstresser",
"local:trusty/statusstresser-1")
status.s.assert_service_exposed("statusstresser", False)
status.s.assert_service_service_status("statusstresser",
{"current": "active",
"message": "called in "
"config-changed hook"})
status.s.assert_unit_workload_status("statusstresser/0",
{"current": "active",
"message": "called in "
"config-changed hook"})
status.s.assert_unit_agent_status("statusstresser/0",
{"current": "idle", "message": ""})
status.s.assert_unit_agent_state("statusstresser/0", "started")
agent_versions = juju_status.get_agent_versions()
for version in agent_versions:
for item in agent_versions[version]:
if not item.isdigit():
status.s.assert_unit_agent_version(item, version)
status.s.assert_unit_machine("statusstresser/0", "1")
def run_reduced_status(client, status):
"""run a subset of the status asserts.
run a reduced set of tests for a StatusParser, this is useful for
status outputs such as Tabular that hold less information.
:param client: python juju client.
:type client: jujupy.EnvJujuClient
:param status: a BaseStatusParser
:type status: BaseStatusParser
"""
status.s.assert_machines_len(2)
status.s.assert_machines_ids(("0", "1"))
juju_status = client.get_status()
for name, machine in juju_status.iter_machines(False):
status.s.assert_machine_agent_state(name, "started")
status.s.assert_machine_agent_version(name,
machine.get("agent-version"))
status.s.assert_machine_dns_name(name, machine.get("dns-name"))
status.s.assert_machine_instance_id(name, machine.get("instance-id"))
status.s.assert_machine_series(name, machine.get("series"))
status.s.assert_machine_hardware(name, machine.get("hardware"))
status.s.assert_service_charm("statusstresser",
"local:trusty/statusstresser-1")
status.s.assert_service_exposed("statusstresser", False)
status.s.assert_service_service_status("statusstresser",
{"current": "active",
"message": ""})
status.s.assert_unit_workload_status("statusstresser/0",
{"current": "active",
"message": "called in "
"config-changed hook"})
status.s.assert_unit_agent_status("statusstresser/0",
{"current": "idle", "message": ""})
status.s.assert_unit_machine("statusstresser/0", "1")
def test_status_set_on_install(client):
"""Test the status after install.
Test that status set is proplerly called during install and
that all formats are returning proper information.
:param client: python juju client.
:type client: jujupy.EnvJujuClient
"""
status = StatusTester.from_text(client.get_status(60, True, False,
"--format=yaml"),
"yaml")
run_complete_status(client, status)
status = StatusTester.from_text(client.get_status(60, True, False,
"--format=json"),
"json")
run_complete_status(client, status)
status = StatusTester.from_text(client.get_status(60, True, False,
"--format=tabular"),
"tabular")
run_reduced_status(client, status)
def parse_args():
"""Parse all arguments."""
parser = ArgumentParser('Test status outputs')
add_basic_testing_arguments(parser)
return parser.parse_args()
def main():
args = parse_args()
series = args.series
if series is None:
args.series = 'precise'
bs_manager = BootstrapManager.from_args(args)
client = bs_manager.client
with bs_manager.booted_context(args.upload_tools):
client.get_status(60)
charm = local_charm_path(
charm='statusstresser', juju_ver=client.version, series='trusty')
client.deploy(charm)
client.wait_for_started()
test_status_set_on_install(client)
if __name__ == '__main__':
main()
|