Add prompting for KSA options1.21.0

Teach OpenStackConfig to prompt the user for KSA plugin options
that have no value but have a prompt string defined.

* Add pw_func argument to __init__() to be used as the callback
  for prompting the user.  The default is None which skips the
  prompt step.
* Add option_prompt() method to perform the checks for prompting,
  call the callback and save the returned value. This is public
  to handle cases where simply passing in a callback is insufficient
  for the prompt mechanism.

Related-Bug: #1617384
Change-Id: I5faa86e94d6f71282ac270e2acfbd3016638c780

2 files changed, 59 insertions, 1 deletions

diff --git a/os_client_config/config.py b/os_client_config/config.py
index 586f0a7..54048aa 100644
--- a/os_client_config/config.py
+++ b/os_client_config/config.py
@@ -173,7 +173,8 @@ class OpenStackConfig(object):
 
     def __init__(self, config_files=None, vendor_files=None,
                  override_defaults=None, force_ipv4=None,
-                 envvar_prefix=None, secure_files=None):
+                 envvar_prefix=None, secure_files=None,
+                 pw_func=None):
         self.log = _log.setup_logging(__name__)
 
         self._config_files = config_files or CONFIG_FILES
@@ -288,6 +289,10 @@ class OpenStackConfig(object):
         # Flag location to hold the peeked value of an argparse timeout value
         self._argv_timeout = False
 
+        # Save the password callback
+        # password = self._pw_callback(prompt="Password: ")
+        self._pw_callback = pw_func
+
     def get_extra_config(self, key, defaults=None):
         """Fetch an arbitrary extra chunk of config, laying in defaults.
 
@@ -924,6 +929,9 @@ class OpenStackConfig(object):
                 winning_value,
             )
 
+            # See if this needs a prompting
+            config = self.option_prompt(config, p_opt)
+
         return config
 
     def _validate_auth_correctly(self, config, loader):
@@ -952,6 +960,19 @@ class OpenStackConfig(object):
                 winning_value,
             )
 
+            # See if this needs a prompting
+            config = self.option_prompt(config, p_opt)
+
+        return config
+
+    def option_prompt(self, config, p_opt):
+        """Prompt user for option that requires a value"""
+        if (
+                p_opt.prompt is not None and
+                p_opt.dest not in config['auth'] and
+                self._pw_callback is not None
+        ):
+            config['auth'][p_opt.dest] = self._pw_callback(p_opt.prompt)
         return config
 
     def _clean_up_after_ourselves(self, config, p_opt, winning_value):
diff --git a/os_client_config/tests/test_config.py b/os_client_config/tests/test_config.py
index 5bcf766..5cded54 100644
--- a/os_client_config/tests/test_config.py
+++ b/os_client_config/tests/test_config.py
@@ -27,6 +27,11 @@ from os_client_config import exceptions
 from os_client_config.tests import base
 
 
+def prompt_for_password(prompt=None):
+    """Fake prompt function that just returns a constant string"""
+    return 'promptpass'
+
+
 class TestConfig(base.TestCase):
 
     def test_get_all_clouds(self):
@@ -787,6 +792,38 @@ class TestConfigArgparse(base.TestCase):
         self.assertNotIn('http_timeout', cloud.config)
 
 
+class TestConfigPrompt(base.TestCase):
+
+    def setUp(self):
+        super(TestConfigPrompt, self).setUp()
+
+        self.args = dict(
+            auth_url='http://example.com/v2',
+            username='user',
+            project_name='project',
+            # region_name='region2',
+            auth_type='password',
+        )
+
+        self.options = argparse.Namespace(**self.args)
+
+    def test_get_one_cloud_prompt(self):
+        c = config.OpenStackConfig(
+            config_files=[self.cloud_yaml],
+            vendor_files=[self.vendor_yaml],
+            pw_func=prompt_for_password,
+        )
+
+        # This needs a cloud definition without a password.
+        # If this starts failing unexpectedly check that the cloud_yaml
+        # and/or vendor_yaml do not have a password in the selected cloud.
+        cc = c.get_one_cloud(
+            cloud='_test_cloud_no_vendor',
+            argparse=self.options,
+        )
+        self.assertEqual('promptpass', cc.auth['password'])
+
+
 class TestConfigDefault(base.TestCase):
 
     def setUp(self):