Limnoria/src/test.py

###
# Copyright (c) 2002-2005, Jeremiah Fincher
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
#   * Redistributions of source code must retain the above copyright notice,
#     this list of conditions, and the following disclaimer.
#   * Redistributions in binary form must reproduce the above copyright notice,
#     this list of conditions, and the following disclaimer in the
#     documentation and/or other materials provided with the distribution.
#   * Neither the name of the author of this software nor the name of
#     contributors to this software may be used to endorse or promote products
#     derived from this software without specific prior written consent.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
###

import gc
import os
import re
import sys
import time
import shutil
import urllib
import httplib
import unittest
import threading
import StringIO

import supybot.log as log
import supybot.i18n as i18n
import supybot.conf as conf
import supybot.utils as utils
import supybot.ircdb as ircdb
import supybot.world as world
import supybot.irclib as irclib
import supybot.plugin as plugin
import supybot.drivers as drivers
import supybot.ircmsgs as ircmsgs
import supybot.registry as registry
import supybot.ircutils as ircutils
import supybot.callbacks as callbacks
import supybot.httpserver as httpserver

i18n.import_conf()
network = True

# This is the global list of suites that are to be run.
suites = []

originalCallbacksGetHelp = callbacks.getHelp
lastGetHelp = 'x' * 1000
def cachingGetHelp(method, name=None, doc=None):
    global lastGetHelp
    lastGetHelp = originalCallbacksGetHelp(method, name, doc)
    return lastGetHelp
callbacks.getHelp = cachingGetHelp

def getTestIrc():
    irc = irclib.Irc('test')
    # Gotta clear the connect messages (USER, NICK, etc.)
    while irc.takeMsg():
        pass
    return irc

class TimeoutError(AssertionError):
    def __str__(self):
        return '%r timed out' % self.args[0]

class TestPlugin(callbacks.Plugin):
    def eval(self, irc, msg, args):
        """<text>

        This is the help for eval.  Since Owner doesn't have an eval command
        anymore, we needed to add this so as not to invalidate any of the tests
        that depended on that eval command.
        """
        try:
            irc.reply(repr(eval(' '.join(args))))
        except callbacks.ArgumentError:
            raise
        except Exception, e:
            irc.reply(utils.exnToString(e))
# Since we know we don't now need the Irc object, we just give None.  This
# might break if callbacks.Privmsg ever *requires* the Irc object.
TestInstance = TestPlugin(None)
conf.registerPlugin('TestPlugin', True, public=False)

class SupyTestCase(unittest.TestCase):
    """This class exists simply for extra logging.  It's come in useful in the
    past."""
    def setUp(self):
        log.critical('Beginning test case %s', self.id())
        threads = [t.getName() for t in threading.enumerate()]
        log.critical('Threads: %L', threads)
        unittest.TestCase.setUp(self)

    def tearDown(self):
        for irc in world.ircs[:]:
            irc._reallyDie()


class PluginTestCase(SupyTestCase):
    """Subclass this to write a test case for a plugin.  See
    plugins/Plugin/test.py for an example.
    """
    timeout = 10
    plugins = None
    cleanConfDir = True
    cleanDataDir = True
    config = {}
    def __init__(self, methodName='runTest'):
        originalRunTest = getattr(self, methodName)
        def runTest(self):
            run = True
            if hasattr(self, 'irc') and self.irc:
                for cb in self.irc.callbacks:
                    cbModule = sys.modules[cb.__class__.__module__]
                    if hasattr(cbModule, 'deprecated') and cbModule.deprecated:
                        print
                        print 'Ignored, %s is deprecated.' % cb.name()
                        run = False
            if run:
                originalRunTest()
        runTest = utils.python.changeFunctionName(runTest, methodName)
        setattr(self.__class__, methodName, runTest)
        SupyTestCase.__init__(self, methodName=methodName)
        self.originals = {}

    def setUp(self, nick='test', forceSetup=False):
        if not forceSetup and \
                self.__class__ in (PluginTestCase, ChannelPluginTestCase):
            # Necessary because there's a test in here that shouldn\'t run.
            return
        SupyTestCase.setUp(self)
        # Just in case, let's do this.  Too many people forget to call their
        # super methods.
        for irc in world.ircs[:]:
            irc._reallyDie()
        # Set conf variables appropriately.
        conf.supybot.reply.whenAddressedBy.chars.setValue('@')
        conf.supybot.reply.error.detailed.setValue(True)
        conf.supybot.reply.whenNotCommand.setValue(True)
        self.myVerbose = world.myVerbose
        def rmFiles(dir):
            for filename in os.listdir(dir):
                file = os.path.join(dir, filename)
                if os.path.isfile(file):
                    os.remove(file)
                else:
                    shutil.rmtree(file)
        if self.cleanConfDir:
            rmFiles(conf.supybot.directories.conf())
        if self.cleanDataDir:
            rmFiles(conf.supybot.directories.data())
        ircdb.users.reload()
        ircdb.ignores.reload()
        ircdb.channels.reload()
        if self.plugins is None:
            raise ValueError, 'PluginTestCase must have a "plugins" attribute.'
        self.nick = nick
        self.prefix = ircutils.joinHostmask(nick, 'user', 'host.domain.tld')
        self.irc = getTestIrc()
        MiscModule = plugin.loadPluginModule('Misc')
        OwnerModule = plugin.loadPluginModule('Owner')
        ConfigModule = plugin.loadPluginModule('Config')
        _ = plugin.loadPluginClass(self.irc, MiscModule)
        _ = plugin.loadPluginClass(self.irc, OwnerModule)
        _ = plugin.loadPluginClass(self.irc, ConfigModule)
        if isinstance(self.plugins, str):
            self.plugins = [self.plugins]
        else:
            for name in self.plugins:
                if name not in ('Owner', 'Misc', 'Config'):
                    module = plugin.loadPluginModule(name,
                                                     ignoreDeprecation=True)
                    cb = plugin.loadPluginClass(self.irc, module)
        self.irc.addCallback(TestInstance)
        for (name, value) in self.config.iteritems():
            group = conf.supybot
            parts = registry.split(name)
            if parts[0] == 'supybot':
                parts.pop(0)
            for part in parts:
                group = group.get(part)
            self.originals[group] = group()
            group.setValue(value)

    def tearDown(self):
        if self.__class__ in (PluginTestCase, ChannelPluginTestCase):
            # Necessary because there's a test in here that shouldn\'t run.
            return
        for (group, original) in self.originals.iteritems():
            group.setValue(original)
        ircdb.users.close()
        ircdb.ignores.close()
        ircdb.channels.close()
        SupyTestCase.tearDown(self)
        self.irc = None
        gc.collect()

    def _feedMsg(self, query, timeout=None, to=None, frm=None,
                 usePrefixChar=True):
        if to is None:
            to = self.irc.nick
        if frm is None:
            frm = self.prefix
        if timeout is None:
            timeout = self.timeout
        if self.myVerbose:
            print # Extra newline, so it's pretty.
        prefixChars = conf.supybot.reply.whenAddressedBy.chars()
        if not usePrefixChar and query[0] in prefixChars:
            query = query[1:]
        msg = ircmsgs.privmsg(to, query, prefix=frm)
        if self.myVerbose:
            print 'Feeding: %r' % msg
        self.irc.feedMsg(msg)
        fed = time.time()
        response = self.irc.takeMsg()
        while response is None and time.time() - fed < timeout:
            time.sleep(0.1) # So it doesn't suck up 100% cpu.
            drivers.run()
            response = self.irc.takeMsg()
        if self.myVerbose:
            print 'Response: %r' % response
        return response

    def getMsg(self, query, **kwargs):
        return self._feedMsg(query, **kwargs)

    def feedMsg(self, query, to=None, frm=None):
        """Just feeds it a message, that's all."""
        if to is None:
            to = self.irc.nick
        if frm is None:
            frm = self.prefix
        self.irc.feedMsg(ircmsgs.privmsg(to, query, prefix=frm))

    # These assertError/assertNoError are somewhat fragile.  The proper way to
    # do them would be to use a proxy for the irc object and intercept .error.
    # But that would be hard, so I don't bother.  When this breaks, it'll get
    # fixed, but not until then.
    def assertError(self, query, **kwargs):
        m = self._feedMsg(query, **kwargs)
        if m is None:
            raise TimeoutError, query
        if lastGetHelp not in m.args[1]:
            self.failUnless(m.args[1].startswith('Error:'),
                            '%r did not error: %s' % (query, m.args[1]))
        return m

    def assertSnarfError(self, query, **kwargs):
        return self.assertError(query, usePrefixChar=False, **kwargs)

    def assertNotError(self, query, **kwargs):
        m = self._feedMsg(query, **kwargs)
        if m is None:
            raise TimeoutError, query
        self.failIf(m.args[1].startswith('Error:'),
                    '%r errored: %s' % (query, m.args[1]))
        self.failIf(lastGetHelp in m.args[1],
                    '%r returned the help string.' % query)
        return m

    def assertSnarfNotError(self, query, **kwargs):
        return self.assertNotError(query, usePrefixChar=False, **kwargs)

    def assertHelp(self, query, **kwargs):
        m = self._feedMsg(query, **kwargs)
        if m is None:
            raise TimeoutError, query
        self.failUnless(lastGetHelp in m.args[1],
                        '%s is not the help (%s)' % (m.args[1], lastGetHelp))
        return m

    def assertNoResponse(self, query, timeout=0, **kwargs):
        m = self._feedMsg(query, timeout=timeout, **kwargs)
        self.failIf(m, 'Unexpected response: %r' % m)
        return m

    def assertSnarfNoResponse(self, query, timeout=0, **kwargs):
        return self.assertNoResponse(query, timeout=timeout,
                                     usePrefixChar=False, **kwargs)

    def assertResponse(self, query, expectedResponse, **kwargs):
        m = self._feedMsg(query, **kwargs)
        if m is None:
            raise TimeoutError, query
        self.assertEqual(m.args[1], expectedResponse,
                         '%r != %r' % (expectedResponse, m.args[1]))
        return m

    def assertSnarfResponse(self, query, expectedResponse, **kwargs):
        return self.assertResponse(query, expectedResponse,
                                   usePrefixChar=False, **kwargs)

    def assertRegexp(self, query, regexp, flags=re.I, **kwargs):
        m = self._feedMsg(query, **kwargs)
        if m is None:
            raise TimeoutError, query
        self.failUnless(re.search(regexp, m.args[1], flags),
                        '%r does not match %r' % (m.args[1], regexp))
        return m

    def assertSnarfRegexp(self, query, regexp, flags=re.I, **kwargs):
        return self.assertRegexp(query, regexp, flags=re.I,
                                 usePrefixChar=False, **kwargs)

    def assertNotRegexp(self, query, regexp, flags=re.I, **kwargs):
        m = self._feedMsg(query, **kwargs)
        if m is None:
            raise TimeoutError, query
        self.failUnless(re.search(regexp, m.args[1], flags) is None,
                        '%r matched %r' % (m.args[1], regexp))
        return m

    def assertSnarfNotRegexp(self, query, regexp, flags=re.I, **kwargs):
        return self.assertNotRegexp(query, regexp, flags=re.I,
                                    usePrefixChar=False, **kwargs)

    def assertAction(self, query, expectedResponse=None, **kwargs):
        m = self._feedMsg(query, **kwargs)
        if m is None:
            raise TimeoutError, query
        self.failUnless(ircmsgs.isAction(m), '%r is not an action.' % m)
        if expectedResponse is not None:
            s = ircmsgs.unAction(m)
            self.assertEqual(s, expectedResponse,
                             '%r != %r' % (s, expectedResponse))
        return m

    def assertSnarfAction(self, query, expectedResponse=None, **kwargs):
        return self.assertAction(query, expectedResponse=None,
                                 usePrefixChar=False, **kwargs)

    def assertActionRegexp(self, query, regexp, flags=re.I, **kwargs):
        m = self._feedMsg(query, **kwargs)
        if m is None:
            raise TimeoutError, query
        self.failUnless(ircmsgs.isAction(m))
        s = ircmsgs.unAction(m)
        self.failUnless(re.search(regexp, s, flags),
                        '%r does not match %r' % (s, regexp))

    def assertSnarfActionRegexp(self, query, regexp, flags=re.I, **kwargs):
        return self.assertActionRegexp(query, regexp, flags=re.I,
                                       usePrefixChar=False, **kwargs)

    _noTestDoc = ('Admin', 'Channel', 'Config',
                  'Misc', 'Owner', 'User', 'TestPlugin')
    def TestDocumentation(self):
        if self.__class__ in (PluginTestCase, ChannelPluginTestCase):
            return
        for cb in self.irc.callbacks:
            name = cb.name()
            if ((name in self._noTestDoc) and \
               not name.lower() in self.__class__.__name__.lower()):
                continue
            self.failUnless(sys.modules[cb.__class__.__name__].__doc__,
                            '%s has no module documentation.' % name)
            if hasattr(cb, 'isCommandMethod'):
                for attr in dir(cb):
                    if cb.isCommandMethod(attr) and \
                       attr == callbacks.canonicalName(attr):
                        self.failUnless(getattr(cb, attr, None).__doc__,
                                        '%s.%s has no help.' % (name, attr))



class ChannelPluginTestCase(PluginTestCase):
    channel = '#test'
    def setUp(self, nick='test', forceSetup=False):
        if not forceSetup and \
                self.__class__ in (PluginTestCase, ChannelPluginTestCase):
            return
        PluginTestCase.setUp(self)
        self.irc.feedMsg(ircmsgs.join(self.channel, prefix=self.prefix))
        m = self.irc.takeMsg()
        self.failIf(m is None, 'No message back from joining channel.')
        self.assertEqual(m.command, 'MODE')
        m = self.irc.takeMsg()
        self.failIf(m is None, 'No message back from joining channel.')
        self.assertEqual(m.command, 'WHO')

    def _feedMsg(self, query, timeout=None, to=None, frm=None, private=False,
                 usePrefixChar=True):
        if to is None:
            if private:
                to = self.irc.nick
            else:
                to = self.channel
        if frm is None:
            frm = self.prefix
        if timeout is None:
            timeout = self.timeout
        if self.myVerbose:
            print # Newline, just like PluginTestCase.
        prefixChars = conf.supybot.reply.whenAddressedBy.chars()
        if query[0] not in prefixChars and usePrefixChar:
            query = prefixChars[0] + query
        msg = ircmsgs.privmsg(to, query, prefix=frm)
        if self.myVerbose:
            print 'Feeding: %r' % msg
        self.irc.feedMsg(msg)
        fed = time.time()
        response = self.irc.takeMsg()
        while response is None and time.time() - fed < timeout:
            time.sleep(0.1)
            drivers.run()
            response = self.irc.takeMsg()
        if response is not None:
            if response.command == 'PRIVMSG':
                args = list(response.args)
                # Strip off nick: at beginning of response.
                if args[1].startswith(self.nick) or \
                   args[1].startswith(ircutils.nickFromHostmask(self.prefix)):
                    try:
                        args[1] = args[1].split(' ', 1)[1]
                    except IndexError:
                        # Odd.  We'll skip this.
                        pass
                ret = ircmsgs.privmsg(*args)
            else:
                ret = response
        else:
            ret = None
        if self.myVerbose:
            print 'Returning: %r' % ret
        return ret

    def feedMsg(self, query, to=None, frm=None, private=False):
        """Just feeds it a message, that's all."""
        if to is None:
            if private:
                to = self.irc.nick
            else:
                to = self.channel
        if frm is None:
            frm = self.prefix
        self.irc.feedMsg(ircmsgs.privmsg(to, query, prefix=frm))

class TestSupyHTTPServer(httpserver.SupyHTTPServer):
    def __init__(self, *args, **kwargs):
        pass
    def serve_forever(self, *args, **kwargs):
        pass
    def shutdown(self, *args, **kwargs):
        pass

class TestRequestHandler(httpserver.SupyHTTPRequestHandler):
    def __init__(self, rfile, wfile, *args, **kwargs):
        self._headers_mode = True
        self.rfile = rfile
        self.wfile = wfile
        self.handle_one_request()

    def send_response(self, code):
        assert self._headers_mode
        self._response = code
    def send_headers(self, name, value):
        assert self._headers_mode
        self._headers[name] = value
    def end_headers(self):
        assert self._headers_mode
        self._headers_mode = False

    def do_X(self, *args, **kwargs):
        assert httpserver.httpServer is not None, \
                'The HTTP server is not started.'
        self.server = httpserver.httpServer
        httpserver.SupyHTTPRequestHandler.do_X(self, *args, **kwargs)

# Partially stolen from the standart Python library :)
def open_http(url, data=None):
    """Use HTTP protocol."""
    import httplib
    user_passwd = None
    proxy_passwd= None
    if isinstance(url, str):
        host, selector = urllib.splithost(url)
        if host:
            user_passwd, host = urllib.splituser(host)
            host = urllib.unquote(host)
        realhost = host
    else:
        host, selector = url
        # check whether the proxy contains authorization information
        proxy_passwd, host = urllib.splituser(host)
        # now we proceed with the url we want to obtain
        urltype, rest = splittype(selector)
        url = rest
        user_passwd = None
        if urltype.lower() != 'http':
            realhost = None
        else:
            realhost, rest = splithost(rest)
            if realhost:
                user_passwd, realhost = urllib.splituser(realhost)
            if user_passwd:
                selector = "%s://%s%s" % (urltype, realhost, rest)
            if proxy_bypass(realhost):
                host = realhost

        #print "proxy via http:", host, selector
    if not host: raise IOError, ('http error', 'no host given')

    if proxy_passwd:
        import base64
        proxy_auth = base64.b64encode(proxy_passwd).strip()
    else:
        proxy_auth = None

    if user_passwd:
        import base64
        auth = base64.b64encode(user_passwd).strip()
    else:
        auth = None
    h = HTTP(host)
    if data is not None:
        h.putrequest('POST', selector)
        h.putheader('Content-Type', 'application/x-www-form-urlencoded')
        h.putheader('Content-Length', '%d' % len(data))
    else:
        h.putrequest('GET', selector)
    if proxy_auth: h.putheader('Proxy-Authorization', 'Basic %s' % proxy_auth)
    if auth: h.putheader('Authorization', 'Basic %s' % auth)
    if realhost: h.putheader('Host', realhost)
    for args in urllib.URLopener().addheaders: h.putheader(*args)
    h.endheaders()
    return h

class FakeHTTPConnection(httplib.HTTPConnection):
    _data = ''
    _headers = {}
    def __init__(self, rfile, wfile):
        httplib.HTTPConnection.__init__(self, 'localhost')
        self.rfile = rfile
        self.wfile = wfile
        self.connect()
    def send(self, data):
        self.wfile.write(data)
    #def putheader(self, name, value):
    #    self._headers[name] = value
    #def connect(self, *args, **kwargs):
    #    self.sock = self.wfile
    #def getresponse(self, *args, **kwargs):
    #    pass

class HTTP(httplib.HTTP):
    _connection_class = FakeHTTPConnection

class HTTPPluginTestCase(PluginTestCase):
    def setUp(self):
        PluginTestCase.setUp(self, forceSetup=True)

    def request(self, url, method='GET', read=True, data={}):
        assert url.startswith('/')
        wfile = StringIO.StringIO()
        rfile = StringIO.StringIO()
        connection = FakeHTTPConnection(wfile, rfile)
        connection.putrequest(method, url)
        connection.endheaders()
        rfile.seek(0)
        wfile.seek(0)
        handler = TestRequestHandler(rfile, wfile)
        if read:
            return (handler._response, wfile.read())
        else:
            return handler._response

    def assertHTTPResponse(self, uri, expectedResponse, **kwargs):
        response = self.request(uri, read=False, **kwargs)
        self.assertEqual(response, expectedResponse)

    def assertNotHTTPResponse(self, irc, expectedResponse, **kwargs):
        response = self.request(uri, read=False, **kwargs)
        self.assertNotEqual(response, expectedResponse)

class ChannelHTTPPluginTestCase(ChannelPluginTestCase, HTTPPluginTestCase):
    def setUp(self):
        ChannelPluginTestCase.setUp(self, forceSetup=True)

# vim:set shiftwidth=4 softtabstop=4 expandtab textwidth=79: