test_service.py 文件源码

def test_server_down_auth(self):
        """
        Verify an ldap.SERVER_DOWN error will retry 2 more times and that
        the connection is closed if all attempts fail.
        """
        service = self.service()

        testStats = {}
        # Verify that without a SERVER_DOWN we don't need to retry, and we
        # still have a connection in the pool
        service._authenticateUsernamePassword_inThread(
            u"uid=wsanchez,cn=user,{0}".format(self.baseDN),
            u"zehcnasw",
            testStats=testStats
        )
        self.assertEquals(testStats["retryNumber"], 0)
        self.assertEquals(len(service.connectionPools["auth"].connections), 1)

        testStats["raise"] = ldap.SERVER_DOWN

        # Now try auth again
        try:
            service._authenticateUsernamePassword_inThread(
                u"uid=wsanchez,cn=user,{0}".format(self.baseDN),
                u"zehcnasw",
                testStats=testStats
            )
        except LDAPQueryError:
            # Verify the number of times we retried
            self.assertEquals(testStats["retryNumber"], 2)
        except:
            self.fail("Should have raised LDAPQueryError")