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
|
#!/usr/bin/env python
from __future__ import print_function
from argparse import ArgumentParser
from contextlib import contextmanager
import logging
import sys
from deploy_stack import (
BootstrapManager,
)
from jujupy import (
get_machine_dns_name,
)
from utility import (
add_basic_testing_arguments,
configure_logging,
JujuAssertionError,
temp_dir,
)
log = logging.getLogger("assess_bootstrap")
INVALID_URL = 'example.com/invalid'
@contextmanager
def thin_booted_context(bs_manager, **kwargs):
client = bs_manager.client
with bs_manager.top_context() as machines:
with bs_manager.bootstrap_context(machines):
client.kill_controller()
client.bootstrap(**kwargs)
with bs_manager.runtime_context(machines):
yield client
def assess_base_bootstrap(bs_manager):
with thin_booted_context(bs_manager) as client:
client.get_status(1)
log.info('Environment successfully bootstrapped.')
def prepare_metadata(client, local_dir):
"""Fill the given directory with metadata using sync_tools."""
client.sync_tools(local_dir)
@contextmanager
def prepare_temp_metadata(client, source_dir=None):
"""Fill a temporary directory with metadata using sync_tools."""
if source_dir is not None:
yield source_dir
else:
with temp_dir() as md_dir:
prepare_metadata(client, md_dir)
yield md_dir
def assess_metadata(bs_manager, local_source):
client = bs_manager.client
# This disconnects from the metadata source, as INVALID_URL is different.
# agent-metadata-url | tools-metadata-url
client.env.update_config({'agent-metadata-url': INVALID_URL})
with prepare_temp_metadata(client, local_source) as metadata_dir:
log.info('Metadata written to: {}'.format(metadata_dir))
with thin_booted_context(bs_manager,
metadata_source=metadata_dir):
log.info('Metadata bootstrap successful.')
data = client.get_model_config()
if INVALID_URL != data['agent-metadata-url']['value']:
raise JujuAssertionError('Error, possible web metadata.')
def get_controller_address(client):
"""Get the address of the controller for this model."""
return get_machine_dns_name(client, "0")
def get_controller_hostname(client):
"""Get the hostname of the controller for this model."""
controller_client = client.get_controller_client()
name = controller_client.run(['hostname'], machines=['0'], use_json=False)
return name.strip()
def assess_to(bs_manager, to):
"""Assess bootstraping with the --to option."""
if to is None:
raise ValueError('--to not given when testing to')
with thin_booted_context(bs_manager, to=to) as client:
log.info('To {} bootstrap successful.'.format(to))
addr = get_controller_hostname(client)
if addr != to:
raise JujuAssertionError('Not bootstrapped to the correct address.')
def assess_bootstrap(args):
bs_manager = BootstrapManager.from_args(args)
if 'base' == args.part:
assess_base_bootstrap(bs_manager)
elif 'metadata' == args.part:
assess_metadata(bs_manager, args.local_metadata_source)
elif 'to' == args.part:
assess_to(bs_manager, args.to)
def parse_args(argv=None):
"""Parse all arguments.
In addition to the basic testing arguments this script also accepts:
part: The first argument, which is the name of test part to run.
--local-metadata-source: If given it should be a directory that contains
the agent to use in the test. This skips downloading them."""
parser = ArgumentParser(description='Test the bootstrap command.')
parser.add_argument('part', choices=['base', 'metadata', 'to'],
help='Which part of bootstrap to assess')
add_basic_testing_arguments(parser)
parser.add_argument('--local-metadata-source',
action='store', default=None,
help='Directory with pre-loaded metadata.')
parser.add_argument('--to', action='store', default=None,
help='bootstrap to (when part=to only)')
return parser.parse_args(argv)
def main(argv=None):
args = parse_args(argv)
configure_logging(args.verbose)
assess_bootstrap(args)
return 0
if __name__ == '__main__':
sys.exit(main())
|