LADI
/
limnoria-irccat
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add Python 3 support.

This commit is contained in:
Valentin Lorentz 2017-05-25 09:55:13 +02:00
parent 0d14a005fe
commit 21e34db417
3 changed files with 51 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
__init__.py.tmpl
View File

@ -34,6 +34,8 @@ Listen for data on a TCP port and forward to designated channel(s).
# pylint: disable=C0103
from imp import reload
import supybot
import supybot.world as world

14
plugin.py
View File

@ -49,6 +49,7 @@ import crypt
import multiprocessing
import pickle
import random
import sys
import time
from twisted.internet import reactor, protocol
@ -62,7 +63,7 @@ from supybot.commands import commalist
from supybot.commands import threading
from supybot.commands import wrap
import config
from . import config
_HELP_URL = "https://github.com/leamas/supybot-irccat"
@ -150,7 +151,7 @@ class _Config(object):
def _dump(self):
''' Update persistent data.'''
pickle.dump(self._data, open(self._path, 'w'))
pickle.dump(self._data, open(self._path, 'wb'))
def get(self, section_name):
''' Return (password, channels) tuple or raise KeyError. '''
@ -175,7 +176,7 @@ class _Config(object):
class IrccatProtocol(basic.LineOnlyReceiver):
''' Line protocol: parse line, forward to channel(s). '''
delimiter = '\n'
delimiter = b'\n'
def __init__(self, config_, blacklist, msg_conn):
self.config = config_
@ -204,6 +205,13 @@ class IrccatProtocol(basic.LineOnlyReceiver):
self.msg_conn.send((what, ['#test']))
self.blacklist.register(self.peer.host, False)
try:
if sys.version_info[0] >= 3:
text = text.decode()
except UnicodeDecodeError:
warning('Invalid encoding: ' + repr(text))
return
try:
section, cleartext_pw, data = text.split(';', 2)
except ValueError:

59
test.py
View File

@ -39,13 +39,15 @@
import os
import os.path
import socket
import subprocess
from supybot.test import *
import config
import plugin as irccat
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'):
@ -53,6 +55,24 @@ def clear_sections(testcase):
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')
@ -71,7 +91,6 @@ class IrccatTestList(PluginTestCase):
class IrccatTestCopy(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)
@ -81,24 +100,21 @@ class IrccatTestCopy(ChannelPluginTestCase):
self.assertNotError('sectiondata ivar ivarpw #test', private = True)
def testCopy(self):
cmd = self.cmd_tmpl % 'ivar;ivarpw;ivar data'
subprocess.check_call(cmd, shell = True)
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):
cmd = self.cmd_tmpl % 'ivar;ivarpw data'
subprocess.check_call(cmd, shell = True)
communicate(b'ivar;ivarpw data\n', sendonly=True)
self.assertRegexp(' ', 'Illegal format.*')
def testBadPw(self):
cmd = self.cmd_tmpl % 'ivar;ivarpw22;ivar data'
subprocess.check_call(cmd, shell = True)
communicate(b'ivar;ivarpw22;ivar data\n', sendonly=True)
self.assertRegexp(' ', 'Bad password.*')
def testBadSection(self):
cmd = self.cmd_tmpl % 'ivaru22;ivarpw22;ivar data'
subprocess.check_call(cmd, shell = True)
communicate(b'ivaru22;ivarpw22;ivar data\n', sendonly=True)
self.assertRegexp(' ', 'No such section.*')
@ -115,28 +131,28 @@ class IrccatTestIrccat(ChannelPluginTestCase):
self.assertNotError('sectiondata ivar ivarpw #test', private = True)
def testIrccatEnvPw(self):
cmd = 'IRCCAT_PASSWORD=ivarpw plugins/Irccat/irccat' \
cmd = 'IRCCAT_PASSWORD=ivarpw %s' \
' localhost 23456 ivar ivar data'
subprocess.check_call(cmd, shell = True)
subprocess.check_call(cmd % CLIENT, shell = True)
self.assertResponse(' ', 'ivar data')
def testIrccatStdinPw(self):
cmd = 'plugins/Irccat/irccat -s localhost 23456 ivar ivar data'
p = subprocess.Popen(cmd, shell = True, stdin = subprocess.PIPE)
p.communicate('ivarpw\n')
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 plugins/Irccat/irccat' \
cmd = 'IRCCAT_PASSWORD=ivarpw %s' \
' localhost 23456'
with self.assertRaises(subprocess.CalledProcessError):
subprocess.check_output(cmd, shell = True)
subprocess.check_output(cmd % CLIENT, shell = True)
def testIrccatBadPort(self):
cmd = 'IRCCAT_PASSWORD=ivarpw plugins/Irccat/irccat' \
cmd = 'IRCCAT_PASSWORD=ivarpw %s' \
' localhost 23456xx ivar ivar data'
with self.assertRaises(subprocess.CalledProcessError):
subprocess.check_output(cmd, shell = True)
subprocess.check_output(cmd % CLIENT, shell = True)
class IrccatTestData(PluginTestCase):
@ -150,7 +166,8 @@ class IrccatTestData(PluginTestCase):
self.assertNotError('sectiondata yngve yngve #al-bot-test')
def testList(self):
self.assertResponse('sectionlist', 'yngve ivar')
m = self._feedMsg('sectionlist')
self.assertIn(m.args[1], ('yngve ivar', 'ivar yngve'))
def testReload(self):
self.assertResponse('reload Irccat', 'The operation succeeded.')