python 2/3 compatibility: use input() from six

(bandit will flag the use of input() in python 2 as a high severity
security issue otherwise)

Also adds a wrapper as a workaround for
pytest-dev/pytest#1598

securedrop/manage.py

@@ -14,6 +14,7 @@
 import time
 import traceback
 
+from six.moves import input
 from contextlib import contextmanager
 from flask import current_app
 from sqlalchemy import create_engine
@@ -32,6 +33,12 @@
 log = logging.getLogger(__name__)
 
 
+def obtain_input(text):
+    """Wrapper for testability as suggested in
+    https://github.com/pytest-dev/pytest/issues/1598#issuecomment-224761877"""
+    return input(text)
+
+
 def reset(args):
     """Clears the SecureDrop development applications' state, restoring them to
     the way they were immediately after running `setup_dev.sh`. This command:
@@ -109,7 +116,7 @@ def add_journalist(args):
 
 def _get_username():
     while True:
-        username = input('Username: ')
+        username = obtain_input('Username: ')
         try:
             Journalist.check_username_acceptable(username)
         except InvalidUsernameException as e:
@@ -121,8 +128,8 @@ def _get_username():
 def _get_yubikey_usage():
     '''Function used to allow for test suite mocking'''
     while True:
-        answer = input('Will this user be using a YubiKey [HOTP]? '
-                       '(y/N): ').lower().strip()
+        answer = obtain_input('Will this user be using a YubiKey [HOTP]? '
+                              '(y/N): ').lower().strip()
         if answer in ('y', 'yes'):
             return True
         elif answer in ('', 'n', 'no'):
@@ -153,7 +160,7 @@ def _add_user(is_admin=False):
         otp_secret = None
         if is_hotp:
             while True:
-                otp_secret = input(
+                otp_secret = obtain_input(
                     "Please configure this user's YubiKey and enter the "
                     "secret: ")
                 if otp_secret:
@@ -203,12 +210,12 @@ def _add_user(is_admin=False):
 
 
 def _get_username_to_delete():
-    return input('Username to delete: ')
+    return obtain_input('Username to delete: ')
 
 
 def _get_delete_confirmation(user):
-    confirmation = input('Are you sure you want to delete user '
-                         '"{}" (y/n)?'.format(user))
+    confirmation = obtain_input('Are you sure you want to delete user '
+                                '"{}" (y/n)?'.format(user))
     if confirmation.lower() != 'y':
         print(('Confirmation not received: user "{}" was NOT '
               'deleted'.format(user)))

securedrop/tests/test_manage.py

@@ -7,11 +7,8 @@
 import os
 import manage
 import mock
-import sys
 import time
 
-from io import StringIO
-
 os.environ['SECUREDROP_ENV'] = 'test'  # noqa
 
 from models import Journalist, db
@@ -42,35 +39,34 @@ def test_verbose(caplog):
 
 
 def test_get_username_success():
-    with mock.patch("builtins.input", return_value='jen'):
+    with mock.patch("manage.obtain_input", return_value='jen'):
         assert manage._get_username() == 'jen'
 
 
 def test_get_username_fail():
     bad_username = 'a' * (Journalist.MIN_USERNAME_LEN - 1)
-    with mock.patch("builtins.input",
+    with mock.patch("manage.obtain_input",
                     side_effect=[bad_username, 'jen']):
         assert manage._get_username() == 'jen'
 
 
 def test_get_yubikey_usage_yes():
-    with mock.patch("builtins.input", return_value='y'):
+    with mock.patch("manage.obtain_input", return_value='y'):
         assert manage._get_yubikey_usage()
 
 
 def test_get_yubikey_usage_no():
-    with mock.patch("builtins.input", return_value='n'):
+    with mock.patch("manage.obtain_input", return_value='n'):
         assert not manage._get_yubikey_usage()
 
 
 # Note: we use the `journalist_app` fixture because it creates the DB
-def test_handle_invalid_secret(journalist_app, config, mocker):
+def test_handle_invalid_secret(journalist_app, config, mocker, capsys):
     """Regression test for bad secret logic in manage.py"""
 
     mocker.patch("manage._get_username", return_value='ntoll'),
     mocker.patch("manage._get_yubikey_usage", return_value=True),
-    mocker.patch("builtins.input", side_effect=YUBIKEY_HOTP),
-    mocker.patch("sys.stdout", new_callable=StringIO),
+    mocker.patch("manage.obtain_input", side_effect=YUBIKEY_HOTP),
 
     original_config = manage.config
 
@@ -80,23 +76,23 @@ def test_handle_invalid_secret(journalist_app, config, mocker):
 
         # We will try to provide one invalid and one valid secret
         return_value = manage._add_user()
+        out, err = capsys.readouterr()
 
         assert return_value == 0
-        assert 'Try again.' in sys.stdout.getvalue()
-        assert 'successfully added' in sys.stdout.getvalue()
+        assert 'Try again.' in out
+        assert 'successfully added' in out
     finally:
         manage.config = original_config
 
 
 # Note: we use the `journalist_app` fixture because it creates the DB
 def test_exception_handling_when_duplicate_username(journalist_app,
                                                     config,
-                                                    mocker):
+                                                    mocker, capsys):
     """Regression test for duplicate username logic in manage.py"""
 
     mocker.patch("manage._get_username", return_value='foo-bar-baz')
     mocker.patch("manage._get_yubikey_usage", return_value=False)
-    mocker.patch("sys.stdout", new_callable=StringIO)
 
     original_config = manage.config
 
@@ -106,14 +102,16 @@ def test_exception_handling_when_duplicate_username(journalist_app,
 
         # Inserting the user for the first time should succeed
         return_value = manage._add_user()
+        out, err = capsys.readouterr()
+
         assert return_value == 0
-        assert 'successfully added' in sys.stdout.getvalue()
+        assert 'successfully added' in out
 
         # Inserting the user for a second time should fail
         return_value = manage._add_user()
+        out, err = capsys.readouterr()
         assert return_value == 1
-        assert ('ERROR: That username is already taken!' in
-                sys.stdout.getvalue())
+        assert 'ERROR: That username is already taken!' in out
     finally:
         manage.config = original_config
 
@@ -142,26 +140,26 @@ def test_delete_user(journalist_app, config, mocker):
 
 
 # Note: we use the `journalist_app` fixture because it creates the DB
-def test_delete_non_existent_user(journalist_app, config, mocker):
+def test_delete_non_existent_user(journalist_app, config, mocker, capsys):
     mocker.patch("manage._get_username_to_delete",
                  return_value='does-not-exist')
     mocker.patch('manage._get_delete_confirmation', return_value=True)
-    mocker.patch("sys.stdout", new_callable=StringIO)
 
     original_config = manage.config
 
     try:
         # We need to override the config to point at the per-test DB
         manage.config = config
         return_value = manage.delete_user(args=None)
+        out, err = capsys.readouterr()
         assert return_value == 0
-        assert 'ERROR: That user was not found!' in sys.stdout.getvalue()
+        assert 'ERROR: That user was not found!' in out
     finally:
         manage.config = original_config
 
 
 def test_get_username_to_delete(mocker):
-    mocker.patch("builtins.input", return_value='test-user-12345')
+    mocker.patch("manage.obtain_input", return_value='test-user-12345')
     return_value = manage._get_username_to_delete()
     assert return_value == 'test-user-12345'
 
@@ -192,7 +190,7 @@ def test_reset(journalist_app, test_journo, alembic_config, config):
 
 
 def test_get_username(mocker):
-    mocker.patch("builtins.input", return_value='foo-bar-baz')
+    mocker.patch("manage.obtain_input", return_value='foo-bar-baz')
     assert manage._get_username() == 'foo-bar-baz'