forked from juju/juju
-
Notifications
You must be signed in to change notification settings - Fork 0
/
assess_upgrade_series.py
executable file
·127 lines (99 loc) · 3.95 KB
/
assess_upgrade_series.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
125
126
127
#!/usr/bin/env python
"""Assess upgrading using the 'upgrade-series' commands."""
from __future__ import print_function
import argparse
import logging
import os
import subprocess
import sys
from deploy_stack import (
BootstrapManager,
)
from jujucharm import (
local_charm_path,
)
from utility import (
add_basic_testing_arguments,
configure_logging,
JujuAssertionError,
)
__metaclass__ = type
log = logging.getLogger("assess_upgrade_series")
DEFAULT_FROM_SERIES = 'xenial'
DEFAULT_TO_SERIES = 'bionic'
def assess_juju_upgrade_series(client, args):
target_machine = '0'
assert_correct_series(client, target_machine, args.from_series)
upgrade_series_prepare(client, target_machine, args.to_series, agree=True)
reboot_machine(client, target_machine)
do_release_upgrade(client, target_machine)
reboot_machine(client, target_machine)
upgrade_series_complete(client, target_machine)
assert_correct_series(client, target_machine, args.to_series)
def upgrade_series_prepare(client, machine, series, **flags):
args = (machine, 'prepare', series)
if flags['agree']:
args += ('-y',)
client.juju('upgrade-series', args)
def upgrade_series_complete(client, machine):
args = (machine, 'complete')
client.juju('upgrade-series', args)
def do_release_upgrade(client, machine):
try:
output = client.get_juju_output(
'ssh', machine, 'sudo do-release-upgrade -f '
'DistUpgradeViewNonInteractive', timeout=3600)
log.info("do-release-upgrade response: ".format(output))
except subprocess.CalledProcessError as e:
raise AssertionError(
"do-release-upgrade failed on {}: {}".format(machine, e))
def reboot_machine(client, machine):
log.info("Restarting: {}".format(machine))
client.juju('run', ('--machine', machine, 'sudo shutdown -r now'))
log.info("wait_for_started()")
client.wait_for_started()
def assert_correct_series(client, machine, expected):
"""Verify that juju knows the correct series for the machine"""
status = client.get_status()
machine_series = status.status['machines'][machine]['series']
if machine_series != expected:
raise JujuAssertionError(
"Machine {} series of {} doesn't match the expected series: {}"
.format(machine, machine_series, expected))
def setup(client, series):
dummy_sink = local_charm_path(
charm='charms/dummy-sink',
juju_ver=client.version,
series=series,
repository=os.environ['JUJU_REPOSITORY'])
dummy_subordinate = local_charm_path(
charm='charms/dummy-subordinate',
juju_ver=client.version,
series=series,
repository=os.environ['JUJU_REPOSITORY'])
_, complete_primary = client.deploy(dummy_sink, series=series)
_, complete_subordinate = client.deploy(dummy_subordinate, series=series)
client.juju('add-relation', ('dummy-sink', 'dummy-subordinate'))
client.set_config('dummy-subordinate', {'token': 'Canonical'})
client.wait_for(complete_primary)
client.wait_for(complete_subordinate)
def parse_args(argv):
parser = argparse.ArgumentParser(description="Test juju update series.")
add_basic_testing_arguments(parser)
parser.add_argument('--from-series', default=DEFAULT_FROM_SERIES,
dest='from_series',
help='Series to start machine and units with')
parser.add_argument('--to-series', default=DEFAULT_TO_SERIES,
dest='to_series',
help='Series to upgrade machine and units to')
return parser.parse_args(argv)
def main(argv=None):
args = parse_args(argv)
configure_logging(args.verbose)
bs_manager = BootstrapManager.from_args(args)
with bs_manager.booted_context(args.upload_tools):
setup(bs_manager.client, args.from_series)
assess_juju_upgrade_series(bs_manager.client, args)
return 0
if __name__ == '__main__':
sys.exit(main())