Repository navigation
Expand file tree
/
Copy pathshell.py
More file actions
222 lines (190 loc) · 8.26 KB
/
Copy pathshell.py
File metadata and controls
222 lines (190 loc) · 8.26 KB
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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
# Copyright 2012-2013 OpenStack Foundation
# Copyright 2015 Dean Troyer
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
"""Command-line interface to the OpenStack APIs"""
import argparse
import sys
from typing import Any
import warnings
from osc_lib.api import auth
from osc_lib.command import commandmanager
from osc_lib import shell
import openstackclient
from openstackclient.common import clientmanager
DEFAULT_DOMAIN = 'default'
# list of modules that were originally out-of-tree and are now in
# core OSC
IGNORED_MODULES = (
'neutron_taas.taas_client.osc',
'neutronclient.osc.v2.fwaas',
'neutronclient.osc.v2.taas',
'neutronclient.osc.v2.networking_bgpvpn',
'neutronclient.osc.v2.dynamic_routing',
'neutronclient.osc.v2.subnet_onboard',
'neutronclient.osc.v2.vpnaas',
'cyborgclient.osc.v2',
# FIXME(stephenfin): this is ignored temporarily while we complete
# implementation. This should be dropped once completed.
'openstackclient.share.v2',
)
class OpenStackShell(shell.OpenStackShell):
client_manager: clientmanager.ClientManager
def __init__(self) -> None:
command_manager = commandmanager.CommandManager(
'openstack.cli', ignored_modules=IGNORED_MODULES
)
super().__init__(
description=__doc__.strip(),
version=openstackclient.__version__,
command_manager=command_manager,
deferred_help=True,
)
self.api_version = {}
# Assume TLS host certificate verification is enabled
self.verify = True
# ignore warnings from openstacksdk since our users can't do anything
# about them
warnings.filterwarnings('ignore', module='openstack')
def build_option_parser(
self,
description: str | None,
version: str | None,
argparse_kwargs: dict[str, Any] | None = None,
) -> argparse.ArgumentParser:
parser = super().build_option_parser(
description, version, argparse_kwargs
)
parser = clientmanager.build_plugin_option_parser(parser)
parser = auth.build_auth_plugins_option_parser(parser)
return parser
def _final_defaults(self) -> None:
super()._final_defaults()
# Set the default plugin to admin_token if endpoint and token are given
if self.options.endpoint and self.options.token:
# Use token authentication
self._auth_type = 'admin_token'
else:
self._auth_type = 'password'
def _load_plugins(self) -> None:
"""Load plugins via stevedore."""
# Loop through extensions to get API versions
for mod in clientmanager.PLUGIN_MODULES:
default_version = getattr(mod, 'DEFAULT_API_VERSION', None)
# Only replace the first instance of "os", some service names will
# have "os" in their name, like: "antiddos"
option = mod.API_VERSION_OPTION.replace('os_', '', 1)
version_opt = str(self.cloud.config.get(option, default_version))
if version_opt:
api = mod.API_NAME
self.api_version[api] = version_opt
# Add a plugin interface to let the module validate the version
# requested by the user
skip_old_check = False
mod_check_api_version = getattr(mod, 'check_api_version', None)
if mod_check_api_version:
# this throws an exception if invalid
skip_old_check = mod_check_api_version(version_opt)
# NOTE(stephenfin): API_VERSIONS has traditionally been a
# dictionary but the values are only used internally and are
# ignored for the modules using SDK. So we now support tuples
# instead.
mod_versions = getattr(mod, 'API_VERSIONS', None)
if mod_versions is not None and not isinstance(
mod_versions, dict | tuple
):
raise TypeError(
f'Plugin {mod} has incompatible API_VERSIONS. '
f'Expected: tuple, dict. Got: {type(mod_versions)}. '
f'Please report this to your package maintainer.'
)
if mod_versions and not skip_old_check:
if version_opt not in mod_versions:
sorted_versions = sorted(
list(mod.API_VERSIONS),
key=lambda s: list(map(int, s.split('.'))),
)
self.log.warning(
"%(name)s API version %(version)s is not in "
"supported versions: %(supported)s",
{
'name': api,
'version': version_opt,
'supported': ', '.join(sorted_versions),
},
)
# Build our command group which we expect to look like:
#
# openstack.<service>.vN
#
# Note that command groups deal only with major versions
cmd_group = '.'.join(
[
'openstack',
api.replace('-', '_'),
'v' + version_opt.replace('.', '_').split('_')[0],
]
)
self.command_manager.add_command_group(cmd_group)
self.log.debug(
'%(name)s API version %(version)s, cmd group %(group)s',
{'name': api, 'version': version_opt, 'group': cmd_group},
)
mod_extensions = getattr(mod, 'API_EXTENSIONS', None)
if not mod_extensions:
continue
for extension in mod_extensions:
extension_cmd_group = '.'.join([cmd_group, extension])
self.command_manager.add_command_group(extension_cmd_group)
self.log.debug(
'%(name)s API version %(version)s '
'(%(extension)s extension), cmd group %(group)s',
{
'name': api,
'version': version_opt,
'extension': extension,
'group': cmd_group,
},
)
def _load_commands(self) -> None:
"""Load commands via cliff/stevedore
osc-lib has no opinion on what commands should be loaded
"""
# Commands that span multiple APIs
self.command_manager.add_command_group('openstack.common')
# This is the naive extension implementation referred to in
# blueprint 'client-extensions'
# Extension modules can register their commands in an
# 'openstack.extension' entry point group:
# entry_points={
# 'openstack.extension': [
# 'list_repo=qaz.github.repo:ListRepo',
# 'show_repo=qaz.github.repo:ShowRepo',
# ],
# }
self.command_manager.add_command_group('openstack.extension')
def initialize_app(self, argv: list[str]) -> None:
super().initialize_app(argv)
# Re-create the client_manager with our subclass
self.client_manager = clientmanager.ClientManager(
cli_options=self.cloud,
api_version=self.api_version,
pw_func=shell.prompt_for_password,
)
def main(argv: list[str] | None = None) -> int:
if argv is None:
argv = sys.argv[1:]
return OpenStackShell().run(argv)
if __name__ == "__main__":
sys.exit(main())