1# Copyright (c) 2014 Hewlett-Packard Development Company, L.P. 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); you may 4# not use this file except in compliance with the License. You may obtain 5# a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 12# License for the specific language governing permissions and limitations 13# under the License. 14 15import argparse 16import copy 17import os 18from unittest import mock 19 20import fixtures 21import testtools 22import yaml 23 24from openstack import config 25from openstack.config import cloud_region 26from openstack.config import defaults 27from openstack import exceptions 28from openstack.tests.unit.config import base 29 30 31def prompt_for_password(prompt=None): 32 """Fake prompt function that just returns a constant string""" 33 return 'promptpass' 34 35 36class TestConfig(base.TestCase): 37 38 def test_get_all(self): 39 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 40 vendor_files=[self.vendor_yaml], 41 secure_files=[self.no_yaml]) 42 clouds = c.get_all() 43 # We add two by hand because the regions cloud is going to exist 44 # thrice since it has three regions in it 45 user_clouds = [ 46 cloud for cloud in base.USER_CONF['clouds'].keys() 47 ] + ['_test_cloud_regions', '_test_cloud_regions'] 48 configured_clouds = [cloud.name for cloud in clouds] 49 self.assertCountEqual(user_clouds, configured_clouds) 50 51 def test_get_all_clouds(self): 52 # Ensure the alias is in place 53 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 54 vendor_files=[self.vendor_yaml], 55 secure_files=[self.no_yaml]) 56 clouds = c.get_all_clouds() 57 # We add two by hand because the regions cloud is going to exist 58 # thrice since it has three regions in it 59 user_clouds = [ 60 cloud for cloud in base.USER_CONF['clouds'].keys() 61 ] + ['_test_cloud_regions', '_test_cloud_regions'] 62 configured_clouds = [cloud.name for cloud in clouds] 63 self.assertCountEqual(user_clouds, configured_clouds) 64 65 def test_get_one(self): 66 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 67 vendor_files=[self.vendor_yaml]) 68 cloud = c.get_one(validate=False) 69 self.assertIsInstance(cloud, cloud_region.CloudRegion) 70 self.assertEqual(cloud.name, '') 71 72 def test_get_one_cloud(self): 73 # Ensure the alias is in place 74 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 75 vendor_files=[self.vendor_yaml]) 76 cloud = c.get_one_cloud(validate=False) 77 self.assertIsInstance(cloud, cloud_region.CloudRegion) 78 self.assertEqual(cloud.name, '') 79 80 def test_get_one_default_cloud_from_file(self): 81 single_conf = base._write_yaml({ 82 'clouds': { 83 'single': { 84 'auth': { 85 'auth_url': 'http://example.com/v2', 86 'username': 'testuser', 87 'password': 'testpass', 88 'project_name': 'testproject', 89 }, 90 'region_name': 'test-region', 91 } 92 } 93 }) 94 c = config.OpenStackConfig(config_files=[single_conf], 95 secure_files=[], 96 vendor_files=[self.vendor_yaml]) 97 cc = c.get_one() 98 self.assertEqual(cc.name, 'single') 99 100 def test_remote_profile(self): 101 single_conf = base._write_yaml({ 102 'clouds': { 103 'remote': { 104 'profile': 'https://example.com', 105 'auth': { 106 'username': 'testuser', 107 'password': 'testpass', 108 'project_name': 'testproject', 109 }, 110 'region_name': 'test-region', 111 } 112 } 113 }) 114 self.register_uris([ 115 dict(method='GET', 116 uri='https://example.com/.well-known/openstack/api', 117 json={ 118 "name": "example", 119 "profile": { 120 "auth": { 121 "auth_url": "https://auth.example.com/v3", 122 } 123 } 124 }), 125 ]) 126 127 c = config.OpenStackConfig(config_files=[single_conf]) 128 cc = c.get_one(cloud='remote') 129 self.assertEqual(cc.name, 'remote') 130 self.assertEqual(cc.auth['auth_url'], 'https://auth.example.com/v3') 131 self.assertEqual(cc.auth['username'], 'testuser') 132 133 def test_get_one_auth_defaults(self): 134 c = config.OpenStackConfig(config_files=[self.cloud_yaml]) 135 cc = c.get_one(cloud='_test-cloud_', auth={'username': 'user'}) 136 self.assertEqual('user', cc.auth['username']) 137 self.assertEqual( 138 defaults._defaults['auth_type'], 139 cc.auth_type, 140 ) 141 142 def test_get_one_with_config_files(self): 143 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 144 vendor_files=[self.vendor_yaml], 145 secure_files=[self.secure_yaml]) 146 self.assertIsInstance(c.cloud_config, dict) 147 self.assertIn('cache', c.cloud_config) 148 self.assertIsInstance(c.cloud_config['cache'], dict) 149 self.assertIn('max_age', c.cloud_config['cache']) 150 self.assertIn('path', c.cloud_config['cache']) 151 cc = c.get_one('_test-cloud_') 152 self._assert_cloud_details(cc) 153 cc = c.get_one('_test_cloud_no_vendor') 154 self._assert_cloud_details(cc) 155 156 def test_get_one_with_int_project_id(self): 157 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 158 vendor_files=[self.vendor_yaml]) 159 cc = c.get_one('_test-cloud-int-project_') 160 self.assertEqual('12345', cc.auth['project_id']) 161 162 def test_get_one_with_domain_id(self): 163 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 164 vendor_files=[self.vendor_yaml]) 165 cc = c.get_one('_test-cloud-domain-id_') 166 self.assertEqual('6789', cc.auth['user_domain_id']) 167 self.assertEqual('123456789', cc.auth['project_domain_id']) 168 self.assertNotIn('domain_id', cc.auth) 169 self.assertNotIn('domain-id', cc.auth) 170 self.assertNotIn('domain_id', cc) 171 172 def test_get_one_unscoped_identity(self): 173 single_conf = base._write_yaml({ 174 'clouds': { 175 'unscoped': { 176 'auth': { 177 'auth_url': 'http://example.com/v2', 178 'username': 'testuser', 179 'password': 'testpass', 180 }, 181 } 182 } 183 }) 184 c = config.OpenStackConfig(config_files=[single_conf], 185 secure_files=[], 186 vendor_files=[self.vendor_yaml]) 187 cc = c.get_one() 188 self.assertEqual('http://example.com/v2', cc.get_endpoint('identity')) 189 190 def test_get_one_domain_scoped(self): 191 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 192 vendor_files=[self.vendor_yaml]) 193 cc = c.get_one('_test-cloud-domain-scoped_') 194 self.assertEqual('12345', cc.auth['domain_id']) 195 self.assertNotIn('user_domain_id', cc.auth) 196 self.assertNotIn('project_domain_id', cc.auth) 197 self.assertIsNone(cc.get_endpoint('identity')) 198 199 def test_get_one_infer_user_domain(self): 200 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 201 vendor_files=[self.vendor_yaml]) 202 cc = c.get_one('_test-cloud-int-project_') 203 self.assertEqual('awesome-domain', cc.auth['user_domain_id']) 204 self.assertEqual('awesome-domain', cc.auth['project_domain_id']) 205 self.assertNotIn('domain_id', cc.auth) 206 self.assertNotIn('domain_id', cc) 207 208 def test_get_one_infer_passcode(self): 209 single_conf = base._write_yaml({ 210 'clouds': { 211 'mfa': { 212 'auth_type': 'v3multifactor', 213 'auth_methods': ['v3password', 'v3totp'], 214 'auth': { 215 'auth_url': 'fake_url', 216 'username': 'testuser', 217 'password': 'testpass', 218 'project_name': 'testproject', 219 'project_domain_name': 'projectdomain', 220 'user_domain_name': 'udn' 221 }, 222 'region_name': 'test-region', 223 } 224 } 225 }) 226 227 c = config.OpenStackConfig(config_files=[single_conf]) 228 cc = c.get_one(cloud='mfa', passcode='123') 229 self.assertEqual('123', cc.auth['passcode']) 230 231 def test_get_one_with_hyphenated_project_id(self): 232 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 233 vendor_files=[self.vendor_yaml]) 234 cc = c.get_one('_test_cloud_hyphenated') 235 self.assertEqual('12345', cc.auth['project_id']) 236 237 def test_get_one_with_hyphenated_kwargs(self): 238 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 239 vendor_files=[self.vendor_yaml]) 240 args = { 241 'auth': { 242 'username': 'testuser', 243 'password': 'testpass', 244 'project-id': '12345', 245 'auth-url': 'http://example.com/v2', 246 }, 247 'region_name': 'test-region', 248 } 249 cc = c.get_one(**args) 250 self.assertEqual('http://example.com/v2', cc.auth['auth_url']) 251 252 def test_no_environ(self): 253 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 254 vendor_files=[self.vendor_yaml]) 255 self.assertRaises( 256 exceptions.ConfigException, c.get_one, 'envvars') 257 258 def test_fallthrough(self): 259 c = config.OpenStackConfig(config_files=[self.no_yaml], 260 vendor_files=[self.no_yaml], 261 secure_files=[self.no_yaml]) 262 for k in os.environ.keys(): 263 if k.startswith('OS_'): 264 self.useFixture(fixtures.EnvironmentVariable(k)) 265 c.get_one(cloud='defaults', validate=False) 266 267 def test_prefer_ipv6_true(self): 268 c = config.OpenStackConfig(config_files=[self.no_yaml], 269 vendor_files=[self.no_yaml], 270 secure_files=[self.no_yaml]) 271 cc = c.get_one(cloud='defaults', validate=False) 272 self.assertTrue(cc.prefer_ipv6) 273 274 def test_prefer_ipv6_false(self): 275 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 276 vendor_files=[self.vendor_yaml]) 277 cc = c.get_one(cloud='_test-cloud_') 278 self.assertFalse(cc.prefer_ipv6) 279 280 def test_force_ipv4_true(self): 281 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 282 vendor_files=[self.vendor_yaml]) 283 cc = c.get_one(cloud='_test-cloud_') 284 self.assertTrue(cc.force_ipv4) 285 286 def test_force_ipv4_false(self): 287 c = config.OpenStackConfig(config_files=[self.no_yaml], 288 vendor_files=[self.no_yaml], 289 secure_files=[self.no_yaml]) 290 cc = c.get_one(cloud='defaults', validate=False) 291 self.assertFalse(cc.force_ipv4) 292 293 def test_get_one_auth_merge(self): 294 c = config.OpenStackConfig(config_files=[self.cloud_yaml]) 295 cc = c.get_one(cloud='_test-cloud_', auth={'username': 'user'}) 296 self.assertEqual('user', cc.auth['username']) 297 self.assertEqual('testpass', cc.auth['password']) 298 299 def test_get_one_networks(self): 300 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 301 vendor_files=[self.vendor_yaml]) 302 cc = c.get_one('_test-cloud-networks_') 303 self.assertEqual( 304 ['a-public', 'another-public', 'split-default'], 305 cc.get_external_networks()) 306 self.assertEqual( 307 ['a-private', 'another-private', 'split-no-default'], 308 cc.get_internal_networks()) 309 self.assertEqual('a-public', cc.get_nat_source()) 310 self.assertEqual('another-private', cc.get_nat_destination()) 311 self.assertEqual('another-public', cc.get_default_network()) 312 self.assertEqual( 313 ['a-public', 'another-public', 'split-no-default'], 314 cc.get_external_ipv4_networks()) 315 self.assertEqual( 316 ['a-public', 'another-public', 'split-default'], 317 cc.get_external_ipv6_networks()) 318 319 def test_get_one_no_networks(self): 320 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 321 vendor_files=[self.vendor_yaml]) 322 cc = c.get_one('_test-cloud-domain-scoped_') 323 self.assertEqual([], cc.get_external_networks()) 324 self.assertEqual([], cc.get_internal_networks()) 325 self.assertIsNone(cc.get_nat_source()) 326 self.assertIsNone(cc.get_nat_destination()) 327 self.assertIsNone(cc.get_default_network()) 328 329 def test_only_secure_yaml(self): 330 c = config.OpenStackConfig(config_files=['nonexistent'], 331 vendor_files=['nonexistent'], 332 secure_files=[self.secure_yaml]) 333 cc = c.get_one(cloud='_test_cloud_no_vendor', validate=False) 334 self.assertEqual('testpass', cc.auth['password']) 335 336 def test_get_cloud_names(self): 337 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 338 secure_files=[self.no_yaml]) 339 self.assertCountEqual( 340 ['_test-cloud-domain-id_', 341 '_test-cloud-domain-scoped_', 342 '_test-cloud-int-project_', 343 '_test-cloud-networks_', 344 '_test-cloud_', 345 '_test-cloud_no_region', 346 '_test_cloud_hyphenated', 347 '_test_cloud_no_vendor', 348 '_test_cloud_regions', 349 '_test-cloud-override-metrics', 350 ], 351 c.get_cloud_names()) 352 c = config.OpenStackConfig(config_files=[self.no_yaml], 353 vendor_files=[self.no_yaml], 354 secure_files=[self.no_yaml]) 355 for k in os.environ.keys(): 356 if k.startswith('OS_'): 357 self.useFixture(fixtures.EnvironmentVariable(k)) 358 c.get_one(cloud='defaults', validate=False) 359 self.assertEqual(['defaults'], sorted(c.get_cloud_names())) 360 361 def test_set_one_cloud_creates_file(self): 362 config_dir = fixtures.TempDir() 363 self.useFixture(config_dir) 364 config_path = os.path.join(config_dir.path, 'clouds.yaml') 365 config.OpenStackConfig.set_one_cloud(config_path, '_test_cloud_') 366 self.assertTrue(os.path.isfile(config_path)) 367 with open(config_path) as fh: 368 self.assertEqual({'clouds': {'_test_cloud_': {}}}, 369 yaml.safe_load(fh)) 370 371 def test_set_one_cloud_updates_cloud(self): 372 new_config = { 373 'cloud': 'new_cloud', 374 'auth': { 375 'password': 'newpass' 376 } 377 } 378 379 resulting_cloud_config = { 380 'auth': { 381 'password': 'newpass', 382 'username': 'testuser', 383 'auth_url': 'http://example.com/v2', 384 }, 385 'cloud': 'new_cloud', 386 'profile': '_test_cloud_in_our_cloud', 387 'region_name': 'test-region' 388 } 389 resulting_config = copy.deepcopy(base.USER_CONF) 390 resulting_config['clouds']['_test-cloud_'] = resulting_cloud_config 391 config.OpenStackConfig.set_one_cloud(self.cloud_yaml, '_test-cloud_', 392 new_config) 393 with open(self.cloud_yaml) as fh: 394 written_config = yaml.safe_load(fh) 395 # We write a cache config for testing 396 written_config['cache'].pop('path', None) 397 self.assertEqual(written_config, resulting_config) 398 399 def test_get_region_no_region_default(self): 400 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 401 vendor_files=[self.vendor_yaml], 402 secure_files=[self.no_yaml]) 403 region = c._get_region(cloud='_test-cloud_no_region') 404 self.assertEqual(region, {'name': '', 'values': {}}) 405 406 def test_get_region_no_region(self): 407 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 408 vendor_files=[self.vendor_yaml], 409 secure_files=[self.no_yaml]) 410 region = c._get_region(cloud='_test-cloud_no_region', 411 region_name='override-region') 412 self.assertEqual(region, {'name': 'override-region', 'values': {}}) 413 414 def test_get_region_region_is_none(self): 415 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 416 vendor_files=[self.vendor_yaml], 417 secure_files=[self.no_yaml]) 418 region = c._get_region(cloud='_test-cloud_no_region', region_name=None) 419 self.assertEqual(region, {'name': '', 'values': {}}) 420 421 def test_get_region_region_set(self): 422 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 423 vendor_files=[self.vendor_yaml], 424 secure_files=[self.no_yaml]) 425 region = c._get_region(cloud='_test-cloud_', region_name='test-region') 426 self.assertEqual(region, {'name': 'test-region', 'values': {}}) 427 428 def test_get_region_many_regions_default(self): 429 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 430 vendor_files=[self.vendor_yaml], 431 secure_files=[self.no_yaml]) 432 region = c._get_region(cloud='_test_cloud_regions', 433 region_name='') 434 self.assertEqual(region, {'name': 'region1', 'values': 435 {'external_network': 'region1-network'}}) 436 437 def test_get_region_many_regions(self): 438 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 439 vendor_files=[self.vendor_yaml], 440 secure_files=[self.no_yaml]) 441 region = c._get_region(cloud='_test_cloud_regions', 442 region_name='region2') 443 self.assertEqual(region, {'name': 'region2', 'values': 444 {'external_network': 'my-network'}}) 445 446 def test_get_region_by_name_no_value(self): 447 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 448 vendor_files=[self.vendor_yaml]) 449 region = c._get_region(cloud='_test_cloud_regions', 450 region_name='region-no-value') 451 self.assertEqual(region, {'name': 'region-no-value', 'values': {}}) 452 453 def test_get_region_invalid_region(self): 454 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 455 vendor_files=[self.vendor_yaml], 456 secure_files=[self.no_yaml]) 457 self.assertRaises( 458 exceptions.ConfigException, c._get_region, 459 cloud='_test_cloud_regions', region_name='invalid-region') 460 461 def test_get_region_no_cloud(self): 462 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 463 vendor_files=[self.vendor_yaml], 464 secure_files=[self.no_yaml]) 465 region = c._get_region(region_name='no-cloud-region') 466 self.assertEqual(region, {'name': 'no-cloud-region', 'values': {}}) 467 468 def test_get_region_invalid_keys(self): 469 invalid_conf = base._write_yaml({ 470 'clouds': { 471 '_test_cloud': { 472 'profile': '_test_cloud_in_our_cloud', 473 'auth': { 474 'auth_url': 'http://example.com/v2', 475 'username': 'testuser', 476 'password': 'testpass', 477 }, 478 'regions': [ 479 { 480 'name': 'region1', 481 'foo': 'bar' 482 }, 483 ] 484 } 485 } 486 }) 487 c = config.OpenStackConfig(config_files=[invalid_conf], 488 vendor_files=[self.vendor_yaml]) 489 self.assertRaises( 490 exceptions.ConfigException, c._get_region, 491 cloud='_test_cloud', region_name='region1') 492 493 @mock.patch('openstack.config.cloud_region.keyring') 494 @mock.patch( 495 'keystoneauth1.identity.base.BaseIdentityPlugin.set_auth_state') 496 def test_load_auth_cache_not_found(self, ks_mock, kr_mock): 497 c = config.OpenStackConfig( 498 config_files=[self.cloud_yaml], secure_files=[]) 499 c._cache_auth = True 500 501 kr_mock.get_password = mock.Mock(side_effect=[RuntimeError]) 502 503 region = c.get_one('_test-cloud_') 504 kr_mock.get_password.assert_called_with( 505 'openstacksdk', region._auth.get_cache_id()) 506 ks_mock.assert_not_called() 507 508 @mock.patch('openstack.config.cloud_region.keyring') 509 @mock.patch( 510 'keystoneauth1.identity.base.BaseIdentityPlugin.set_auth_state') 511 def test_load_auth_cache_found(self, ks_mock, kr_mock): 512 c = config.OpenStackConfig( 513 config_files=[self.cloud_yaml], secure_files=[]) 514 c._cache_auth = True 515 fake_auth = {'a': 'b'} 516 517 kr_mock.get_password = mock.Mock(return_value=fake_auth) 518 519 region = c.get_one('_test-cloud_') 520 kr_mock.get_password.assert_called_with( 521 'openstacksdk', region._auth.get_cache_id()) 522 ks_mock.assert_called_with(fake_auth) 523 524 @mock.patch('openstack.config.cloud_region.keyring') 525 def test_set_auth_cache(self, kr_mock): 526 c = config.OpenStackConfig( 527 config_files=[self.cloud_yaml], secure_files=[]) 528 c._cache_auth = True 529 530 kr_mock.get_password = mock.Mock(side_effect=[RuntimeError]) 531 kr_mock.set_password = mock.Mock() 532 533 region = c.get_one('_test-cloud_') 534 535 region.set_auth_cache() 536 kr_mock.set_password.assert_called_with( 537 'openstacksdk', region._auth.get_cache_id(), 538 region._auth.get_auth_state()) 539 540 def test_metrics_global(self): 541 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 542 vendor_files=[self.vendor_yaml], 543 secure_files=[self.secure_yaml]) 544 self.assertIsInstance(c.cloud_config, dict) 545 cc = c.get_one('_test-cloud_') 546 statsd = { 547 'host': '127.0.0.1', 548 'port': '1234', 549 } 550 # NOTE(ianw) we don't test/call get_<stat>_client() because we 551 # don't want to instantiate the client, which tries to 552 # connect / do hostname lookups. 553 self.assertEqual(statsd['host'], cc._statsd_host) 554 self.assertEqual(statsd['port'], cc._statsd_port) 555 self.assertEqual('openstack.api', cc.get_statsd_prefix()) 556 influxdb = { 557 'use_udp': True, 558 'host': '127.0.0.1', 559 'port': '1234', 560 'username': 'username', 561 'password': 'password', 562 'database': 'database', 563 'measurement': 'measurement.name', 564 'timeout': 10 565 } 566 self.assertEqual(influxdb, cc._influxdb_config) 567 568 def test_metrics_override(self): 569 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 570 vendor_files=[self.vendor_yaml], 571 secure_files=[self.secure_yaml]) 572 self.assertIsInstance(c.cloud_config, dict) 573 cc = c.get_one('_test-cloud-override-metrics') 574 statsd = { 575 'host': '127.0.0.1', 576 'port': '4321', 577 'prefix': 'statsd.override.prefix' 578 } 579 self.assertEqual(statsd['host'], cc._statsd_host) 580 self.assertEqual(statsd['port'], cc._statsd_port) 581 self.assertEqual(statsd['prefix'], cc.get_statsd_prefix()) 582 influxdb = { 583 'use_udp': True, 584 'host': '127.0.0.1', 585 'port': '1234', 586 'username': 'override-username', 587 'password': 'override-password', 588 'database': 'override-database', 589 'measurement': 'measurement.name', 590 'timeout': 10 591 } 592 self.assertEqual(influxdb, cc._influxdb_config) 593 594 595class TestExcludedFormattedConfigValue(base.TestCase): 596 # verify https://storyboard.openstack.org/#!/story/1635696 597 # 598 # get_one_cloud() and get_one_cloud_osc() iterate over config 599 # values and try to expand any variables in those values by 600 # calling value.format(), however some config values 601 # (e.g. password) should never have format() applied to them, not 602 # only might that change the password but it will also cause the 603 # format() function to raise an exception if it can not parse the 604 # format string. Examples would be single brace (e.g. 'foo{') 605 # which raises an ValueError because it's looking for a matching 606 # end brace or a brace pair with a key value that cannot be found 607 # (e.g. 'foo{bar}') which raises a KeyError. 608 609 def setUp(self): 610 super(TestExcludedFormattedConfigValue, self).setUp() 611 612 self.args = dict( 613 auth_url='http://example.com/v2', 614 username='user', 615 project_name='project', 616 region_name='region2', 617 snack_type='cookie', 618 os_auth_token='no-good-things', 619 ) 620 621 self.options = argparse.Namespace(**self.args) 622 623 def test_get_one_cloud_password_brace(self): 624 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 625 vendor_files=[self.vendor_yaml]) 626 627 password = 'foo{' # Would raise ValueError, single brace 628 self.options.password = password 629 cc = c.get_one_cloud( 630 cloud='_test_cloud_regions', argparse=self.options, validate=False) 631 self.assertEqual(cc.password, password) 632 633 password = 'foo{bar}' # Would raise KeyError, 'bar' not found 634 self.options.password = password 635 cc = c.get_one_cloud( 636 cloud='_test_cloud_regions', argparse=self.options, validate=False) 637 self.assertEqual(cc.password, password) 638 639 def test_get_one_cloud_osc_password_brace(self): 640 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 641 vendor_files=[self.vendor_yaml]) 642 password = 'foo{' # Would raise ValueError, single brace 643 self.options.password = password 644 cc = c.get_one_cloud_osc( 645 cloud='_test_cloud_regions', argparse=self.options, validate=False) 646 self.assertEqual(cc.password, password) 647 648 password = 'foo{bar}' # Would raise KeyError, 'bar' not found 649 self.options.password = password 650 cc = c.get_one_cloud_osc( 651 cloud='_test_cloud_regions', argparse=self.options, validate=False) 652 self.assertEqual(cc.password, password) 653 654 655class TestConfigArgparse(base.TestCase): 656 657 def setUp(self): 658 super(TestConfigArgparse, self).setUp() 659 660 self.args = dict( 661 auth_url='http://example.com/v2', 662 username='user', 663 password='password', 664 project_name='project', 665 region_name='region2', 666 snack_type='cookie', 667 os_auth_token='no-good-things', 668 ) 669 670 self.options = argparse.Namespace(**self.args) 671 672 def test_get_one_bad_region_argparse(self): 673 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 674 vendor_files=[self.vendor_yaml]) 675 676 self.assertRaises( 677 exceptions.ConfigException, c.get_one, 678 cloud='_test-cloud_', argparse=self.options) 679 680 def test_get_one_argparse(self): 681 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 682 vendor_files=[self.vendor_yaml]) 683 684 cc = c.get_one( 685 cloud='_test_cloud_regions', argparse=self.options, validate=False) 686 self.assertEqual(cc.region_name, 'region2') 687 self.assertEqual(cc.snack_type, 'cookie') 688 689 def test_get_one_precedence(self): 690 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 691 vendor_files=[self.vendor_yaml]) 692 693 kwargs = { 694 'auth': { 695 'username': 'testuser', 696 'password': 'authpass', 697 'project-id': 'testproject', 698 'auth_url': 'http://example.com/v2', 699 }, 700 'region_name': 'kwarg_region', 701 'password': 'ansible_password', 702 'arbitrary': 'value', 703 } 704 705 args = dict( 706 auth_url='http://example.com/v2', 707 username='user', 708 password='argpass', 709 project_name='project', 710 region_name='region2', 711 snack_type='cookie', 712 ) 713 714 options = argparse.Namespace(**args) 715 cc = c.get_one( 716 argparse=options, **kwargs) 717 self.assertEqual(cc.region_name, 'region2') 718 self.assertEqual(cc.auth['password'], 'authpass') 719 self.assertEqual(cc.snack_type, 'cookie') 720 721 def test_get_one_cloud_precedence_osc(self): 722 c = config.OpenStackConfig( 723 config_files=[self.cloud_yaml], 724 vendor_files=[self.vendor_yaml], 725 ) 726 727 kwargs = { 728 'auth': { 729 'username': 'testuser', 730 'password': 'authpass', 731 'project-id': 'testproject', 732 'auth_url': 'http://example.com/v2', 733 }, 734 'region_name': 'kwarg_region', 735 'password': 'ansible_password', 736 'arbitrary': 'value', 737 } 738 739 args = dict( 740 auth_url='http://example.com/v2', 741 username='user', 742 password='argpass', 743 project_name='project', 744 region_name='region2', 745 snack_type='cookie', 746 ) 747 748 options = argparse.Namespace(**args) 749 cc = c.get_one_cloud_osc( 750 argparse=options, 751 **kwargs 752 ) 753 self.assertEqual(cc.region_name, 'region2') 754 self.assertEqual(cc.auth['password'], 'argpass') 755 self.assertEqual(cc.snack_type, 'cookie') 756 757 def test_get_one_precedence_no_argparse(self): 758 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 759 vendor_files=[self.vendor_yaml]) 760 761 kwargs = { 762 'auth': { 763 'username': 'testuser', 764 'password': 'authpass', 765 'project-id': 'testproject', 766 'auth_url': 'http://example.com/v2', 767 }, 768 'region_name': 'kwarg_region', 769 'password': 'ansible_password', 770 'arbitrary': 'value', 771 } 772 773 cc = c.get_one(**kwargs) 774 self.assertEqual(cc.region_name, 'kwarg_region') 775 self.assertEqual(cc.auth['password'], 'authpass') 776 self.assertIsNone(cc.password) 777 778 def test_get_one_just_argparse(self): 779 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 780 vendor_files=[self.vendor_yaml]) 781 782 cc = c.get_one(argparse=self.options, validate=False) 783 self.assertIsNone(cc.cloud) 784 self.assertEqual(cc.region_name, 'region2') 785 self.assertEqual(cc.snack_type, 'cookie') 786 787 def test_get_one_just_kwargs(self): 788 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 789 vendor_files=[self.vendor_yaml]) 790 791 cc = c.get_one(validate=False, **self.args) 792 self.assertIsNone(cc.cloud) 793 self.assertEqual(cc.region_name, 'region2') 794 self.assertEqual(cc.snack_type, 'cookie') 795 796 def test_get_one_dash_kwargs(self): 797 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 798 vendor_files=[self.vendor_yaml]) 799 800 args = { 801 'auth-url': 'http://example.com/v2', 802 'username': 'user', 803 'password': 'password', 804 'project_name': 'project', 805 'region_name': 'other-test-region', 806 'snack_type': 'cookie', 807 } 808 cc = c.get_one(**args) 809 self.assertIsNone(cc.cloud) 810 self.assertEqual(cc.region_name, 'other-test-region') 811 self.assertEqual(cc.snack_type, 'cookie') 812 813 def test_get_one_no_argparse(self): 814 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 815 vendor_files=[self.vendor_yaml]) 816 817 cc = c.get_one(cloud='_test-cloud_', argparse=None) 818 self._assert_cloud_details(cc) 819 self.assertEqual(cc.region_name, 'test-region') 820 self.assertIsNone(cc.snack_type) 821 822 def test_get_one_no_argparse_regions(self): 823 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 824 vendor_files=[self.vendor_yaml]) 825 826 cc = c.get_one(cloud='_test_cloud_regions', argparse=None) 827 self._assert_cloud_details(cc) 828 self.assertEqual(cc.region_name, 'region1') 829 self.assertIsNone(cc.snack_type) 830 831 def test_get_one_bad_region(self): 832 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 833 vendor_files=[self.vendor_yaml]) 834 835 self.assertRaises( 836 exceptions.ConfigException, 837 c.get_one, 838 cloud='_test_cloud_regions', region_name='bad') 839 840 def test_get_one_bad_region_no_regions(self): 841 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 842 vendor_files=[self.vendor_yaml]) 843 self.assertRaises( 844 exceptions.ConfigException, 845 c.get_one, 846 cloud='_test-cloud_', region_name='bad_region') 847 848 def test_get_one_no_argparse_region2(self): 849 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 850 vendor_files=[self.vendor_yaml]) 851 852 cc = c.get_one( 853 cloud='_test_cloud_regions', region_name='region2', argparse=None) 854 self._assert_cloud_details(cc) 855 self.assertEqual(cc.region_name, 'region2') 856 self.assertIsNone(cc.snack_type) 857 858 def test_get_one_network(self): 859 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 860 vendor_files=[self.vendor_yaml]) 861 862 cc = c.get_one( 863 cloud='_test_cloud_regions', region_name='region1', argparse=None) 864 self._assert_cloud_details(cc) 865 self.assertEqual(cc.region_name, 'region1') 866 self.assertEqual('region1-network', cc.config['external_network']) 867 868 def test_get_one_per_region_network(self): 869 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 870 vendor_files=[self.vendor_yaml]) 871 872 cc = c.get_one( 873 cloud='_test_cloud_regions', region_name='region2', argparse=None) 874 self._assert_cloud_details(cc) 875 self.assertEqual(cc.region_name, 'region2') 876 self.assertEqual('my-network', cc.config['external_network']) 877 878 def test_get_one_no_yaml_no_cloud(self): 879 c = config.OpenStackConfig(load_yaml_config=False) 880 881 self.assertRaises( 882 exceptions.ConfigException, 883 c.get_one, 884 cloud='_test_cloud_regions', region_name='region2', argparse=None) 885 886 def test_get_one_no_yaml(self): 887 c = config.OpenStackConfig(load_yaml_config=False) 888 889 cc = c.get_one( 890 region_name='region2', argparse=None, 891 **base.USER_CONF['clouds']['_test_cloud_regions']) 892 # Not using assert_cloud_details because of cache settings which 893 # are not present without the file 894 self.assertIsInstance(cc, cloud_region.CloudRegion) 895 self.assertTrue(hasattr(cc, 'auth')) 896 self.assertIsInstance(cc.auth, dict) 897 self.assertIsNone(cc.cloud) 898 self.assertIn('username', cc.auth) 899 self.assertEqual('testuser', cc.auth['username']) 900 self.assertEqual('testpass', cc.auth['password']) 901 self.assertFalse(cc.config['image_api_use_tasks']) 902 self.assertTrue('project_name' in cc.auth or 'project_id' in cc.auth) 903 if 'project_name' in cc.auth: 904 self.assertEqual('testproject', cc.auth['project_name']) 905 elif 'project_id' in cc.auth: 906 self.assertEqual('testproject', cc.auth['project_id']) 907 self.assertEqual(cc.region_name, 'region2') 908 909 def test_fix_env_args(self): 910 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 911 vendor_files=[self.vendor_yaml]) 912 913 env_args = {'os-compute-api-version': 1} 914 fixed_args = c._fix_args(env_args) 915 916 self.assertDictEqual({'compute_api_version': 1}, fixed_args) 917 918 def test_extra_config(self): 919 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 920 vendor_files=[self.vendor_yaml]) 921 922 defaults = {'use_hostnames': False, 'other-value': 'something'} 923 ansible_options = c.get_extra_config('ansible', defaults) 924 925 # This should show that the default for use_hostnames above is 926 # overridden by the value in the config file defined in base.py 927 # It should also show that other-value key is normalized and passed 928 # through even though there is no corresponding value in the config 929 # file, and that expand-hostvars key is normalized and the value 930 # from the config comes through even though there is no default. 931 self.assertDictEqual( 932 { 933 'expand_hostvars': False, 934 'use_hostnames': True, 935 'other_value': 'something', 936 }, 937 ansible_options) 938 939 def test_get_client_config(self): 940 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 941 vendor_files=[self.vendor_yaml]) 942 943 cc = c.get_one( 944 cloud='_test_cloud_regions') 945 946 defaults = { 947 'use_hostnames': False, 948 'other-value': 'something', 949 'force_ipv4': False, 950 } 951 ansible_options = cc.get_client_config('ansible', defaults) 952 953 # This should show that the default for use_hostnames and force_ipv4 954 # above is overridden by the value in the config file defined in 955 # base.py 956 # It should also show that other-value key is normalized and passed 957 # through even though there is no corresponding value in the config 958 # file, and that expand-hostvars key is normalized and the value 959 # from the config comes through even though there is no default. 960 self.assertDictEqual( 961 { 962 'expand_hostvars': False, 963 'use_hostnames': True, 964 'other_value': 'something', 965 'force_ipv4': True, 966 }, 967 ansible_options) 968 969 def test_register_argparse_cloud(self): 970 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 971 vendor_files=[self.vendor_yaml]) 972 parser = argparse.ArgumentParser() 973 c.register_argparse_arguments(parser, []) 974 opts, _remain = parser.parse_known_args(['--os-cloud', 'foo']) 975 self.assertEqual(opts.os_cloud, 'foo') 976 977 def test_env_argparse_precedence(self): 978 self.useFixture(fixtures.EnvironmentVariable( 979 'OS_TENANT_NAME', 'tenants-are-bad')) 980 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 981 vendor_files=[self.vendor_yaml]) 982 983 cc = c.get_one( 984 cloud='envvars', argparse=self.options, validate=False) 985 self.assertEqual(cc.auth['project_name'], 'project') 986 987 def test_argparse_default_no_token(self): 988 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 989 vendor_files=[self.vendor_yaml]) 990 991 parser = argparse.ArgumentParser() 992 c.register_argparse_arguments(parser, []) 993 # novaclient will add this 994 parser.add_argument('--os-auth-token') 995 opts, _remain = parser.parse_known_args() 996 cc = c.get_one( 997 cloud='_test_cloud_regions', argparse=opts) 998 self.assertEqual(cc.config['auth_type'], 'password') 999 self.assertNotIn('token', cc.config['auth']) 1000 1001 def test_argparse_token(self): 1002 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1003 vendor_files=[self.vendor_yaml]) 1004 1005 parser = argparse.ArgumentParser() 1006 c.register_argparse_arguments(parser, []) 1007 # novaclient will add this 1008 parser.add_argument('--os-auth-token') 1009 opts, _remain = parser.parse_known_args( 1010 ['--os-auth-token', 'very-bad-things', 1011 '--os-auth-type', 'token']) 1012 cc = c.get_one(argparse=opts, validate=False) 1013 self.assertEqual(cc.config['auth_type'], 'token') 1014 self.assertEqual(cc.config['auth']['token'], 'very-bad-things') 1015 1016 def test_argparse_underscores(self): 1017 c = config.OpenStackConfig(config_files=[self.no_yaml], 1018 vendor_files=[self.no_yaml], 1019 secure_files=[self.no_yaml]) 1020 parser = argparse.ArgumentParser() 1021 parser.add_argument('--os_username') 1022 argv = [ 1023 '--os_username', 'user', '--os_password', 'pass', 1024 '--os-auth-url', 'auth-url', '--os-project-name', 'project'] 1025 c.register_argparse_arguments(parser, argv=argv) 1026 opts, _remain = parser.parse_known_args(argv) 1027 cc = c.get_one(argparse=opts) 1028 self.assertEqual(cc.config['auth']['username'], 'user') 1029 self.assertEqual(cc.config['auth']['password'], 'pass') 1030 self.assertEqual(cc.config['auth']['auth_url'], 'auth-url') 1031 1032 def test_argparse_action_append_no_underscore(self): 1033 c = config.OpenStackConfig(config_files=[self.no_yaml], 1034 vendor_files=[self.no_yaml], 1035 secure_files=[self.no_yaml]) 1036 parser = argparse.ArgumentParser() 1037 parser.add_argument('--foo', action='append') 1038 argv = ['--foo', '1', '--foo', '2'] 1039 c.register_argparse_arguments(parser, argv=argv) 1040 opts, _remain = parser.parse_known_args(argv) 1041 self.assertEqual(opts.foo, ['1', '2']) 1042 1043 def test_argparse_underscores_duplicate(self): 1044 c = config.OpenStackConfig(config_files=[self.no_yaml], 1045 vendor_files=[self.no_yaml], 1046 secure_files=[self.no_yaml]) 1047 parser = argparse.ArgumentParser() 1048 parser.add_argument('--os_username') 1049 argv = [ 1050 '--os_username', 'user', '--os_password', 'pass', 1051 '--os-username', 'user1', '--os-password', 'pass1', 1052 '--os-auth-url', 'auth-url', '--os-project-name', 'project'] 1053 self.assertRaises( 1054 exceptions.ConfigException, 1055 c.register_argparse_arguments, 1056 parser=parser, argv=argv) 1057 1058 def test_register_argparse_bad_plugin(self): 1059 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1060 vendor_files=[self.vendor_yaml]) 1061 parser = argparse.ArgumentParser() 1062 self.assertRaises( 1063 exceptions.ConfigException, 1064 c.register_argparse_arguments, 1065 parser, ['--os-auth-type', 'foo']) 1066 1067 def test_register_argparse_not_password(self): 1068 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1069 vendor_files=[self.vendor_yaml]) 1070 parser = argparse.ArgumentParser() 1071 args = [ 1072 '--os-auth-type', 'v3token', 1073 '--os-token', 'some-secret', 1074 ] 1075 c.register_argparse_arguments(parser, args) 1076 opts, _remain = parser.parse_known_args(args) 1077 self.assertEqual(opts.os_token, 'some-secret') 1078 1079 def test_register_argparse_password(self): 1080 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1081 vendor_files=[self.vendor_yaml]) 1082 parser = argparse.ArgumentParser() 1083 args = [ 1084 '--os-password', 'some-secret', 1085 ] 1086 c.register_argparse_arguments(parser, args) 1087 opts, _remain = parser.parse_known_args(args) 1088 self.assertEqual(opts.os_password, 'some-secret') 1089 with testtools.ExpectedException(AttributeError): 1090 opts.os_token 1091 1092 def test_register_argparse_service_type(self): 1093 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1094 vendor_files=[self.vendor_yaml]) 1095 parser = argparse.ArgumentParser() 1096 args = [ 1097 '--os-service-type', 'network', 1098 '--os-endpoint-type', 'admin', 1099 '--http-timeout', '20', 1100 ] 1101 c.register_argparse_arguments(parser, args) 1102 opts, _remain = parser.parse_known_args(args) 1103 self.assertEqual(opts.os_service_type, 'network') 1104 self.assertEqual(opts.os_endpoint_type, 'admin') 1105 self.assertEqual(opts.http_timeout, '20') 1106 with testtools.ExpectedException(AttributeError): 1107 opts.os_network_service_type 1108 cloud = c.get_one(argparse=opts, validate=False) 1109 self.assertEqual(cloud.config['service_type'], 'network') 1110 self.assertEqual(cloud.config['interface'], 'admin') 1111 self.assertEqual(cloud.config['api_timeout'], '20') 1112 self.assertNotIn('http_timeout', cloud.config) 1113 1114 def test_register_argparse_network_service_type(self): 1115 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1116 vendor_files=[self.vendor_yaml]) 1117 parser = argparse.ArgumentParser() 1118 args = [ 1119 '--os-endpoint-type', 'admin', 1120 '--network-api-version', '4', 1121 ] 1122 c.register_argparse_arguments(parser, args, ['network']) 1123 opts, _remain = parser.parse_known_args(args) 1124 self.assertEqual(opts.os_service_type, 'network') 1125 self.assertEqual(opts.os_endpoint_type, 'admin') 1126 self.assertIsNone(opts.os_network_service_type) 1127 self.assertIsNone(opts.os_network_api_version) 1128 self.assertEqual(opts.network_api_version, '4') 1129 cloud = c.get_one(argparse=opts, validate=False) 1130 self.assertEqual(cloud.config['service_type'], 'network') 1131 self.assertEqual(cloud.config['interface'], 'admin') 1132 self.assertEqual(cloud.config['network_api_version'], '4') 1133 self.assertNotIn('http_timeout', cloud.config) 1134 1135 def test_register_argparse_network_service_types(self): 1136 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1137 vendor_files=[self.vendor_yaml]) 1138 parser = argparse.ArgumentParser() 1139 args = [ 1140 '--os-compute-service-name', 'cloudServers', 1141 '--os-network-service-type', 'badtype', 1142 '--os-endpoint-type', 'admin', 1143 '--network-api-version', '4', 1144 ] 1145 c.register_argparse_arguments( 1146 parser, args, ['compute', 'network', 'volume']) 1147 opts, _remain = parser.parse_known_args(args) 1148 self.assertEqual(opts.os_network_service_type, 'badtype') 1149 self.assertIsNone(opts.os_compute_service_type) 1150 self.assertIsNone(opts.os_volume_service_type) 1151 self.assertEqual(opts.os_service_type, 'compute') 1152 self.assertEqual(opts.os_compute_service_name, 'cloudServers') 1153 self.assertEqual(opts.os_endpoint_type, 'admin') 1154 self.assertIsNone(opts.os_network_api_version) 1155 self.assertEqual(opts.network_api_version, '4') 1156 cloud = c.get_one(argparse=opts, validate=False) 1157 self.assertEqual(cloud.config['service_type'], 'compute') 1158 self.assertEqual(cloud.config['network_service_type'], 'badtype') 1159 self.assertEqual(cloud.config['interface'], 'admin') 1160 self.assertEqual(cloud.config['network_api_version'], '4') 1161 self.assertNotIn('volume_service_type', cloud.config) 1162 self.assertNotIn('http_timeout', cloud.config) 1163 1164 1165class TestConfigPrompt(base.TestCase): 1166 1167 def setUp(self): 1168 super(TestConfigPrompt, self).setUp() 1169 1170 self.args = dict( 1171 auth_url='http://example.com/v2', 1172 username='user', 1173 project_name='project', 1174 # region_name='region2', 1175 auth_type='password', 1176 ) 1177 1178 self.options = argparse.Namespace(**self.args) 1179 1180 def test_get_one_prompt(self): 1181 c = config.OpenStackConfig( 1182 config_files=[self.cloud_yaml], 1183 vendor_files=[self.vendor_yaml], 1184 pw_func=prompt_for_password, 1185 ) 1186 1187 # This needs a cloud definition without a password. 1188 # If this starts failing unexpectedly check that the cloud_yaml 1189 # and/or vendor_yaml do not have a password in the selected cloud. 1190 cc = c.get_one( 1191 cloud='_test_cloud_no_vendor', 1192 argparse=self.options, 1193 ) 1194 self.assertEqual('promptpass', cc.auth['password']) 1195 1196 1197class TestConfigDefault(base.TestCase): 1198 1199 def setUp(self): 1200 super(TestConfigDefault, self).setUp() 1201 1202 # Reset defaults after each test so that other tests are 1203 # not affected by any changes. 1204 self.addCleanup(self._reset_defaults) 1205 1206 def _reset_defaults(self): 1207 defaults._defaults = None 1208 1209 def test_set_no_default(self): 1210 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1211 vendor_files=[self.vendor_yaml]) 1212 cc = c.get_one(cloud='_test-cloud_', argparse=None) 1213 self._assert_cloud_details(cc) 1214 self.assertEqual('password', cc.auth_type) 1215 1216 1217class TestBackwardsCompatibility(base.TestCase): 1218 1219 def test_set_no_default(self): 1220 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1221 vendor_files=[self.vendor_yaml]) 1222 cloud = { 1223 'identity_endpoint_type': 'admin', 1224 'compute_endpoint_type': 'private', 1225 'endpoint_type': 'public', 1226 'auth_type': 'v3password', 1227 } 1228 result = c._fix_backwards_interface(cloud) 1229 expected = { 1230 'identity_interface': 'admin', 1231 'compute_interface': 'private', 1232 'interface': 'public', 1233 'auth_type': 'v3password', 1234 } 1235 self.assertDictEqual(expected, result) 1236 1237 def test_project_v2password(self): 1238 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1239 vendor_files=[self.vendor_yaml]) 1240 cloud = { 1241 'auth_type': 'v2password', 1242 'auth': { 1243 'project-name': 'my_project_name', 1244 'project-id': 'my_project_id' 1245 } 1246 } 1247 result = c._fix_backwards_project(cloud) 1248 expected = { 1249 'auth_type': 'v2password', 1250 'auth': { 1251 'tenant_name': 'my_project_name', 1252 'tenant_id': 'my_project_id' 1253 } 1254 } 1255 self.assertEqual(expected, result) 1256 1257 def test_project_password(self): 1258 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1259 vendor_files=[self.vendor_yaml]) 1260 cloud = { 1261 'auth_type': 'password', 1262 'auth': { 1263 'project-name': 'my_project_name', 1264 'project-id': 'my_project_id' 1265 } 1266 } 1267 result = c._fix_backwards_project(cloud) 1268 expected = { 1269 'auth_type': 'password', 1270 'auth': { 1271 'project_name': 'my_project_name', 1272 'project_id': 'my_project_id' 1273 } 1274 } 1275 self.assertEqual(expected, result) 1276 1277 def test_backwards_network_fail(self): 1278 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1279 vendor_files=[self.vendor_yaml]) 1280 cloud = { 1281 'external_network': 'public', 1282 'networks': [ 1283 {'name': 'private', 'routes_externally': False}, 1284 ] 1285 } 1286 self.assertRaises( 1287 exceptions.ConfigException, 1288 c._fix_backwards_networks, cloud) 1289 1290 def test_backwards_network(self): 1291 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1292 vendor_files=[self.vendor_yaml]) 1293 cloud = { 1294 'external_network': 'public', 1295 'internal_network': 'private', 1296 } 1297 result = c._fix_backwards_networks(cloud) 1298 expected = { 1299 'external_network': 'public', 1300 'internal_network': 'private', 1301 'networks': [ 1302 {'name': 'public', 'routes_externally': True, 1303 'nat_destination': False, 'default_interface': True}, 1304 {'name': 'private', 'routes_externally': False, 1305 'nat_destination': True, 'default_interface': False}, 1306 ] 1307 } 1308 self.assertEqual(expected, result) 1309 1310 def test_normalize_network(self): 1311 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1312 vendor_files=[self.vendor_yaml]) 1313 cloud = { 1314 'networks': [ 1315 {'name': 'private'} 1316 ] 1317 } 1318 result = c._fix_backwards_networks(cloud) 1319 expected = { 1320 'networks': [ 1321 {'name': 'private', 'routes_externally': False, 1322 'nat_destination': False, 'default_interface': False, 1323 'nat_source': False, 1324 'routes_ipv4_externally': False, 1325 'routes_ipv6_externally': False}, 1326 ] 1327 } 1328 self.assertEqual(expected, result) 1329 1330 def test_single_default_interface(self): 1331 c = config.OpenStackConfig(config_files=[self.cloud_yaml], 1332 vendor_files=[self.vendor_yaml]) 1333 cloud = { 1334 'networks': [ 1335 {'name': 'blue', 'default_interface': True}, 1336 {'name': 'purple', 'default_interface': True}, 1337 ] 1338 } 1339 self.assertRaises( 1340 exceptions.ConfigException, 1341 c._fix_backwards_networks, cloud) 1342