Modernize auth tests to use shared server manager

Also move auth tests to be new style filename, obj naming

Also allow test task module selector to see new-style test modules

1 files changed, 136 insertions, 0 deletions

diff --git a/tests/auth.py b/tests/auth.py
new file mode 100644
index 00000000..08de6148
--- /dev/null
+++ b/tests/auth.py
@@ -0,0 +1,136 @@
+# Copyright (C) 2008  Robey Pointer <robeypointer@gmail.com>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation; either version 2.1 of the License, or (at your option)
+# any later version.
+#
+# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
+
+"""
+Some unit tests for authenticating over a Transport.
+"""
+
+import unittest
+from pytest import raises
+
+from paramiko import (
+    DSSKey,
+    BadAuthenticationType,
+    AuthenticationException,
+)
+
+from ._util import _support, server, unicodey
+
+
+class AuthHandler_:
+    def bad_auth_type(self):
+        """
+        verify that we get the right exception when an unsupported auth
+        type is requested.
+        """
+        # Server won't allow password auth for this user, so should fail
+        # and return just publickey allowed types
+        with server(
+            connect=dict(username="unknown", password="error"),
+            catch_error=True,
+        ) as (_, _, err):
+            assert isinstance(err, BadAuthenticationType)
+            assert err.allowed_types == ["publickey"]
+
+    def bad_password(self):
+        """
+        verify that a bad password gets the right exception, and that a retry
+        with the right password works.
+        """
+        # NOTE: Transport.connect doesn't do any auth upfront if no userauth
+        # related kwargs given.
+        with server(defer=True) as (tc, ts):
+            # Auth once, badly
+            with raises(AuthenticationException):
+                tc.auth_password(username="slowdive", password="error")
+            # And again, correctly
+            tc.auth_password(username="slowdive", password="pygmalion")
+
+    def multipart_auth(self):
+        """
+        verify that multipart auth works.
+        """
+        with server(defer=True) as (tc, ts):
+            assert tc.auth_password(
+                username="paranoid", password="paranoid"
+            ) == ["publickey"]
+            key = DSSKey.from_private_key_file(_support("dss.key"))
+            assert tc.auth_publickey(username="paranoid", key=key) == []
+
+    def interactive_auth(self):
+        """
+        verify keyboard-interactive auth works.
+        """
+
+        def handler(title, instructions, prompts):
+            self.got_title = title
+            self.got_instructions = instructions
+            self.got_prompts = prompts
+            return ["cat"]
+
+        with server(defer=True) as (tc, ts):
+            assert tc.auth_interactive("commie", handler) == []
+            assert self.got_title == "password"
+            assert self.got_prompts == [("Password", False)]
+
+    def interactive_fallback(self):
+        """
+        verify that a password auth attempt will fallback to "interactive"
+        if password auth isn't supported but interactive is.
+        """
+        with server(defer=True) as (tc, ts):
+            # This username results in an allowed_auth of just kbd-int,
+            # and has a configured interactive->response on the server.
+            assert tc.auth_password("commie", "cat") == []
+
+    def utf8(self):
+        """
+        verify that utf-8 encoding happens in authentication.
+        """
+        with server(defer=True) as (tc, ts):
+            assert tc.auth_password("utf8", unicodey) == []
+
+    def non_utf8(self):
+        """
+        verify that non-utf-8 encoded passwords can be used for broken
+        servers.
+        """
+        with server(defer=True) as (tc, ts):
+            assert tc.auth_password("non-utf8", "\xff") == []
+
+    def auth_exception_when_disconnected(self):
+        """
+        verify that we catch a server disconnecting during auth, and report
+        it as an auth failure.
+        """
+        with server(defer=True, skip_verify=True) as (tc, ts), raises(
+            AuthenticationException
+        ):
+            tc.auth_password("bad-server", "hello")
+
+    def non_responsive_triggers_auth_exception(self):
+        """
+        verify that authentication times out if server takes to long to
+        respond (or never responds).
+        """
+        with server(defer=True, skip_verify=True) as (tc, ts), raises(
+            AuthenticationException
+        ) as info:
+            tc.auth_timeout = 1  # 1 second, to speed up test
+            tc.auth_password("unresponsive-server", "hello")
+            assert "Authentication timeout" in str(info.value)