def test_deprecation(self):
     """
     Accessing L{twisted.words.protocols.msn} emits a deprecation warning
     """
     requireModule('twisted.words.protocols').msn
     warningsShown = self.flushWarnings([self.test_deprecation])
     self.assertEqual(len(warningsShown), 1)
     self.assertIdentical(warningsShown[0]['category'], DeprecationWarning)
     self.assertEqual(
         warningsShown[0]['message'],
         'twisted.words.protocols.msn was deprecated in Twisted 15.1.0: ' +
         'MSN has shutdown.')

    def test_requireModuleDefaultNone(self):
        """
        When module import fails it returns L{None} by default.
        """
        result = reflect.requireModule('no.such.module')

        self.assertIsNone(result)

    def getHandleErrorCode(self):
        """
        Return the argument L{OpenSSL.SSL.Error} will be constructed with for
        this case. This is basically just a random OpenSSL implementation
        detail. It would be better if this test worked in a way which did not
        require this.
        """
        # Windows 2000 SP 4 and Windows XP SP 2 give back WSAENOTSOCK for
        # SSL.Connection.write for some reason.  The twisted.protocols.tls
        # implementation of IReactorSSL doesn't suffer from this imprecation,
        # though, since it is isolated from the Windows I/O layer (I suppose?).

        # If test_properlyCloseFiles waited for the SSL handshake to complete
        # and performed an orderly shutdown, then this would probably be a
        # little less weird: writing to a shutdown SSL connection has a more
        # well-defined failure mode (or at least it should).

        # So figure out if twisted.protocols.tls is in use.  If it can be
        # imported, it should be.
        if requireModule('twisted.protocols.tls') is None:
            # It isn't available, so we expect WSAENOTSOCK if we're on Windows.
            if platform.getType() == 'win32':
                return errno.WSAENOTSOCK

        # Otherwise, we expect an error about how we tried to write to a
        # shutdown connection.  This is terribly implementation-specific.
        return [('SSL routines', 'SSL_write', 'protocol is shutdown')]

 def can_connect(self):
     if requireModule('kinterbasdb') is None:
         return False
     try:
         self.startDB()
         self.stopDB()
         return True
     except:
         return False

    def test_requireModuleRequestedImport(self):
        """
        When module import succeed it returns the module and not the default
        value.
        """
        from twisted.python import monkey

        default = object()

        self.assertIs(reflect.requireModule("twisted.python.monkey", default=default), monkey)

 def test_linux(self):
     """
     L{_getInstallFunction} chooses the epoll reactor on Linux, or poll if
     epoll is unavailable.
     """
     install = _getInstallFunction(linux)
     if requireModule('twisted.internet.epollreactor') is None:
         self.assertIsPoll(install)
     else:
         self.assertEqual(
             install.__module__, 'twisted.internet.epollreactor')

    def test_requireModuleImportError(self):
        """
        When module import fails with ImportError it returns the specified
        default value.
        """
        for name in ['nosuchmtopodule', 'no.such.module']:
            default = object()

            result = reflect.requireModule(name, default=default)

            self.assertIs(result, default)

    def test_defaultAuths(self):
        """
        Make sure that if the C{--auth} command-line option is not passed,
        the default checkers are (for backwards compatibility): SSH, UNIX, and
        PAM if available
        """
        numCheckers = 2

        if requireModule('twisted.cred.pamauth'):
            self.assertIn(IPluggableAuthenticationModules,
                self.options['credInterfaces'],
                "PAM should be one of the modules")
            numCheckers += 1

        self.assertIn(ISSHPrivateKey, self.options['credInterfaces'],
            "SSH should be one of the default checkers")
        self.assertIn(IUsernamePassword, self.options['credInterfaces'],
            "UNIX should be one of the default checkers")
        self.assertEqual(numCheckers, len(self.options['credCheckers']),
            "There should be %d checkers by default" % (numCheckers,))

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.conch.client.knownhosts}.
"""

import os
from binascii import Error as BinasciiError, b2a_base64, a2b_base64

from twisted.python.reflect import requireModule

if requireModule('Crypto') and requireModule('pyasn1'):
    from twisted.conch.ssh.keys import Key, BadKeyError
    from twisted.conch.client.knownhosts import \
        PlainEntry, HashedEntry, KnownHostsFile, UnparsedEntry, ConsoleUI
    from twisted.conch.client import default
else:
    skip = "PyCrypto and PyASN1 required for twisted.conch.knownhosts."

from zope.interface.verify import verifyObject

from twisted.python.filepath import FilePath
from twisted.trial.unittest import TestCase
from twisted.internet.defer import Deferred
from twisted.conch.interfaces import IKnownHostEntry
from twisted.conch.error import HostKeyChanged, UserRejectedKey, InvalidEntry
from twisted.test.testutils import ComparisonTestsMixin


sampleEncodedKey = (

# See LICENSE for details.

"""
Tests for L{twisted.mail.tap}.
"""

from twisted.trial.unittest import TestCase

from twisted.python.usage import UsageError
from twisted.mail import protocols
from twisted.mail.tap import Options, makeService
from twisted.python.filepath import FilePath
from twisted.python.reflect import requireModule
from twisted.internet import endpoints, defer

if requireModule('OpenSSL') is None:
    sslSkip = 'Missing OpenSSL package.'
else:
    sslSkip = None


class OptionsTests(TestCase):
    """
    Tests for the command line option parser used for I{twistd mail}.
    """
    def setUp(self):
        self.aliasFilename = self.mktemp()
        aliasFile = file(self.aliasFilename, 'w')
        aliasFile.write('someuser:\tdifferentuser\n')
        aliasFile.close()

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for the inotify wrapper in L{twisted.internet.inotify}.
"""

from twisted.internet import defer, reactor
from twisted.python import filepath, runtime
from twisted.python.reflect import requireModule
from twisted.trial import unittest

if requireModule('twisted.python._inotify') is not None:
    from twisted.internet import inotify
else:
    inotify = None



class INotifyTests(unittest.TestCase):
    """
    Define all the tests for the basic functionality exposed by
    L{inotify.INotify}.
    """
    if not runtime.platform.supportsINotify():
        skip = "This platform doesn't support INotify."

    def setUp(self):
        self.dirname = filepath.FilePath(self.mktemp())
        self.dirname.createDirectory()
        self.inotify = inotify.INotify()

# -*- test-case-name: twisted.conch.test.test_unix -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

from __future__ import absolute_import

from zope.interface import implementer

from twisted.internet.interfaces import IReactorProcess
from twisted.python.reflect import requireModule
from twisted.trial import unittest

cryptography = requireModule("cryptography")
unix = requireModule('twisted.conch.unix')



@implementer(IReactorProcess)
class MockProcessSpawner(object):
    """
    An L{IReactorProcess} that logs calls to C{spawnProcess}.
    """

    def __init__(self):
        self._spawnProcessCalls = []


    def spawnProcess(self, processProtocol, executable, args=(), env={},
                     path=None, uid=None, gid=None, usePTY=0, childFDs=None):
        """
        Log a call to C{spawnProcess}. Do not actually spawn a process.

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.conch.openssh_compat}.
"""

import os

from twisted.trial.unittest import TestCase
from twisted.python.filepath import FilePath
from twisted.python.reflect import requireModule

if requireModule("Crypto.Cipher.DES3") and requireModule("pyasn1"):
    from twisted.conch.openssh_compat.factory import OpenSSHFactory
else:
    OpenSSHFactory = None

from twisted.conch.test import keydata
from twisted.test.test_process import MockOS


class OpenSSHFactoryTests(TestCase):
    """
    Tests for L{OpenSSHFactory}.
    """

    if getattr(os, "geteuid", None) is None:
        skip = "geteuid/seteuid not available"
    elif OpenSSHFactory is None:
        skip = "Cannot run without PyCrypto or PyASN1"

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.conch.openssh_compat}.
"""

import os

from twisted.trial.unittest import TestCase
from twisted.python.filepath import FilePath
from twisted.python.reflect import requireModule

if requireModule('cryptography') and requireModule('pyasn1'):
    from twisted.conch.openssh_compat.factory import OpenSSHFactory
else:
    OpenSSHFactory = None

from twisted.conch.ssh._kex import getDHGeneratorAndPrime
from twisted.conch.test import keydata
from twisted.test.test_process import MockOS


class OpenSSHFactoryTests(TestCase):
    """
    Tests for L{OpenSSHFactory}.
    """
    if getattr(os, "geteuid", None) is None:
        skip = "geteuid/seteuid not available"
    elif OpenSSHFactory is None:
        skip = "Cannot run without cryptography or PyASN1"

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for implementations of L{inetdtap}.
"""

from twisted.python.reflect import requireModule
from twisted.trial import unittest

inetdtap = requireModule('twisted.runner.inetdtap')
inetdtapSkip = None
if inetdtap is None:
    inetdtapSkip = 'inetdtap not available'



class RPCServerTests(unittest.TestCase):
    """
    Tests for L{inetdtap.RPCServer}
    """
    if inetdtapSkip:
        skip = inetdtapSkip


    def test_deprecation(self):
        """
        It is deprecated.
        """
        inetdtap.RPCServer(
            'some-versions', '/tmp/rpc.conf', 'tcp', 'some-service')

# from the Python Standard Library
import os, re, socket, subprocess, errno
from sys import platform

# from Twisted
from twisted.python.reflect import requireModule
from twisted.internet import defer, threads, reactor
from twisted.internet.protocol import DatagramProtocol
from twisted.internet.error import CannotListenError
from twisted.python.procutils import which
from twisted.python import log
from twisted.internet.endpoints import AdoptedStreamServerEndpoint
from twisted.internet.interfaces import IReactorSocket

fcntl = requireModule("fcntl")

from foolscap.util import allocate_tcp_port # re-exported

try:
    import resource
    def increase_rlimits():
        # We'd like to raise our soft resource.RLIMIT_NOFILE, since certain
        # systems (OS-X, probably solaris) start with a relatively low limit
        # (256), and some unit tests want to open up more sockets than this.
        # Most linux systems start with both hard and soft limits at 1024,
        # which is plenty.

        # unfortunately the values to pass to setrlimit() vary widely from
        # one system to another. OS-X reports (256, HUGE), but the real hard
        # limit is 10240, and accepts (-1,-1) to mean raise it to the
        # maximum. Cygwin reports (256, -1), then ignores a request of

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for the command-line interfaces to conch.
"""
from twisted.python.reflect import requireModule

if requireModule('pyasn1'):
    pyasn1Skip = None
else:
    pyasn1Skip =  "Cannot run without PyASN1"

if requireModule('Crypto'):
    cryptoSkip = None
else:
    cryptoSkip = "can't run w/o PyCrypto"

if requireModule('tty'):
    ttySkip = None
else:
    ttySkip = "can't run w/o tty"

try:
    import Tkinter
except ImportError:
    tkskip = "can't run w/o Tkinter"
else:
    try:
        Tkinter.Tk().destroy()
    except Tkinter.TclError, e:

from twisted.internet.error import ConnectionClosed, FileDescriptorOverrun
from twisted.internet.address import UNIXAddress
from twisted.internet.endpoints import UNIXServerEndpoint, UNIXClientEndpoint
from twisted.internet.defer import Deferred, fail
from twisted.internet.task import LoopingCall
from twisted.internet import interfaces
from twisted.internet.protocol import (
    ServerFactory, ClientFactory, DatagramProtocol)
from twisted.internet.test.test_core import ObjectModelIntegrationMixin
from twisted.internet.test.test_tcp import StreamTransportTestsMixin
from twisted.internet.test.connectionmixins import (
    EndpointCreator, ConnectableProtocol, runProtocolsWithReactor,
    ConnectionTestsMixin, StreamClientTestsMixin)
from twisted.internet.test.reactormixins import ReactorBuilder

if requireModule('twisted.python.sendmsg') is None:
    sendmsgSkip = (
        "sendmsg extension unavailable, extended UNIX features disabled")
else:
    sendmsgSkip = None



class UNIXFamilyMixin:
    """
    Test-helper defining mixin for things related to AF_UNIX sockets.
    """
    def _modeTest(self, methodName, path, factory):
        """
        Assert that the mode of the created unix socket is set to the mode
        specified to the reactor method.

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE file for details.

"""
Tests for L{twisted.conch.scripts.cftp}.
"""

import locale
import time, sys, os, operator, getpass, struct
from io import BytesIO

from twisted.python.filepath import FilePath
from twisted.python.reflect import requireModule
from zope.interface import implementer

pyasn1 = requireModule('pyasn1')
cryptography = requireModule('cryptography')
unix = requireModule('twisted.conch.unix')

_reason = None
if cryptography and pyasn1:
    try:
        from twisted.conch.scripts import cftp
        from twisted.conch.scripts.cftp import SSHSession
        from twisted.conch.ssh import filetransfer
        from twisted.conch.test.test_filetransfer import FileTransferForTestAvatar
        from twisted.conch.test import test_ssh, test_conch
        from twisted.conch.test.test_conch import FakeStdio
    except ImportError:
        pass

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.conch.scripts.ckeygen}.
"""

import __builtin__
import getpass
import sys
from StringIO import StringIO

from twisted.python.reflect import requireModule

if requireModule("cryptography") and requireModule("pyasn1"):
    from twisted.conch.ssh.keys import Key, BadKeyError
    from twisted.conch.scripts.ckeygen import changePassPhrase, displayPublicKey, printFingerprint, _saveKey
else:
    skip = "cryptography and pyasn1 required for twisted.conch.scripts.ckeygen"

from twisted.python.filepath import FilePath
from twisted.trial.unittest import TestCase
from twisted.conch.test.keydata import publicRSA_openssh, privateRSA_openssh, privateRSA_openssh_encrypted


def makeGetpass(*passphrases):
    """
    Return a callable to patch C{getpass.getpass}.  Yields a passphrase each
    time called. Use case is to provide an old, then new passphrase(s) as if
    requested interactively.