test.py

###
# Copyright (c) 2013, Alec Leamas
# 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.

###

# Unused wildcard imports:
# pylint: disable=W0614,W0401
# Missing docstrings:
# pylint: disable=C0111
# supybot's typenames are irregular
# Too many public methods:
# pylint: disable=R0904


import os
import os.path
import socket
import subprocess

from supybot.test import *

from . import config
from . import plugin as irccat

CLIENT = os.path.join(os.path.dirname(__file__), 'irccat')

def clear_sections(testcase):
    if os.path.exists('test-sections.pickle'):
        os.unlink('test-sections.pickle')
    config.global_option('sectionspath').setValue('test-sections.pickle')
    config.global_option('port').setValue(23456)

def communicate(msg, sendonly):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.connect(('localhost', 23456))
        s.sendall(msg)

        if not sendonly:
            data = []
            try:
                data.append([s.recv(1024)])
                while data[-1]:
                    data.append(s.recv(1024))
            except socket.timeout:
                pass
            return b''.join(data)
    finally:
        s.close()


class IrccatTestList(PluginTestCase):
    plugins = ('Irccat', 'User')

    def setUp(self, nick='test'):      # pylint: disable=W0221
        clear_sections(self)
        PluginTestCase.setUp(self)
        self.assertNotError('reload Irccat')
        self.assertNotError('register suptest suptest')
        self.assertNotError('sectiondata ivar ivar #al-bot-test')

    def testList(self):
        self.assertResponse('sectionlist', 'ivar')


class IrccatTestCopy(ChannelPluginTestCase):
    plugins = ('Irccat', 'User')
    channel = '#test'

    def setUp(self, nick='test'):      # pylint: disable=W0221
        clear_sections(self)
        ChannelPluginTestCase.setUp(self)
        self.assertNotError('reload Irccat', private = True)
        self.assertNotError('register suptest suptest', private = True)
        self.assertNotError('sectiondata ivar ivarpw #test', private = True)

    def testCopy(self):
        communicate(b'ivar;ivarpw;ivar data\n', sendonly=True)
        result = self.getMsg(' ')
        self.assertIsNot(result, None)
        self.assertEqual(result.args[1], 'ivar data')

    def testBadFormat(self):
        communicate(b'ivar;ivarpw data\n', sendonly=True)
        self.assertRegexp(' ', 'Illegal format.*')

    def testBadPw(self):
        communicate(b'ivar;ivarpw22;ivar data\n', sendonly=True)
        self.assertRegexp(' ', 'Bad password.*')

    def testBadSection(self):
        communicate(b'ivaru22;ivarpw22;ivar data\n', sendonly=True)
        self.assertRegexp(' ', 'No such section.*')


class IrccatTestIrccat(ChannelPluginTestCase):
    plugins = ('Irccat', 'User')
    channel = '#test'
    cmd_tmpl = "echo '%s' | nc --send-only localhost 23456"

    def setUp(self, nick='test'):      # pylint: disable=W0221
        clear_sections(self)
        ChannelPluginTestCase.setUp(self)
        self.assertNotError('reload Irccat', private = True)
        self.assertNotError('register suptest suptest', private = True)
        self.assertNotError('sectiondata ivar ivarpw #test', private = True)

    def testIrccatEnvPw(self):
        cmd = 'IRCCAT_PASSWORD=ivarpw %s' \
              ' localhost 23456 ivar ivar data'
        subprocess.check_call(cmd % CLIENT, shell = True)
        self.assertResponse(' ', 'ivar data')

    def testIrccatStdinPw(self):
        cmd = '%s -s  localhost 23456 ivar ivar data'
        p = subprocess.Popen(cmd % CLIENT, shell = True, stdin = subprocess.PIPE)
        p.communicate(b'ivarpw\n')
        self.assertResponse(' ', 'ivar data')

    def testIrccatBadCmdline(self):
        cmd = 'IRCCAT_PASSWORD=ivarpw %s' \
              ' localhost 23456'
        with self.assertRaises(subprocess.CalledProcessError):
            subprocess.check_output(cmd % CLIENT, shell = True)

    def testIrccatBadPort(self):
        cmd = 'IRCCAT_PASSWORD=ivarpw %s' \
              ' localhost 23456xx ivar ivar data'
        with self.assertRaises(subprocess.CalledProcessError):
            subprocess.check_output(cmd % CLIENT, shell = True)


class IrccatTestData(PluginTestCase):
    plugins = ('Irccat', 'User')

    def setUp(self, nick='test'):      # pylint: disable=W0221
        clear_sections(self)
        PluginTestCase.setUp(self)
        self.assertNotError('reload Irccat')
        self.assertNotError('sectiondata ivar ivar #al-bot-test')
        self.assertNotError('sectiondata yngve yngve #al-bot-test')

    def testList(self):
        m = self._feedMsg('sectionlist')
        self.assertIn(m.args[1], ('yngve ivar', 'ivar yngve'))

    def testReload(self):
        self.assertResponse('reload Irccat', 'The operation succeeded.')

    def testShow(self):
        self.assertRegexp('sectionshow yngve', '.*#al-bot-test$')

    def testKill(self):
        self.assertNotError('sectionkill yngve')
        self.assertResponse('sectionlist', 'ivar')

    def testKillBadSection(self):
        self.assertResponse('sectionkill tore', 'Error: no such section')


class BlacklistTest(SupyTestCase):

    def setUp(self):
        SupyTestCase.setUp(self)
        self.blacklist = None

    def testBlock(self):
        self.blacklist = irccat._Blacklist()    # pylint: disable=W0212
        self.blacklist.FailMax = 5
        self.blacklist.BlockTime = 0.2

        host = '132.132.132.132'
        self.assertFalse(self.blacklist.onList(host))
        for i in [1, 2, 3, 4]:                 # pylint: disable=W0612
            self.blacklist.register(host, False)
        self.assertFalse(self.blacklist.onList(host))
        self.blacklist.register(host, False)
        self.assertTrue(self.blacklist.onList(host))
        time.sleep(0.25)
        self.assertFalse(self.blacklist.onList(host))
        for i in [1, 2, 3, 4, 5]:
            self.blacklist.register(host, False)
        self.assertTrue(self.blacklist.onList(host))

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