gajim-plural/test/unit/test_socks5.py

175 lines
4.7 KiB
Python
Raw Normal View History

2011-08-02 05:13:56 +02:00
'''
Tests for dispatcher_nb.py
'''
import unittest
import lib
lib.setup_env()
from mock import Mock
import sys
2011-08-15 18:16:17 +02:00
import socket
2011-08-02 05:13:56 +02:00
2017-08-12 02:39:55 +02:00
from gajim.common.socks5 import *
from gajim.common import jingle_xtls
2011-08-02 05:13:56 +02:00
class fake_sock(Mock):
def __init__(self, sockobj):
Mock.__init__(self)
2013-01-01 19:36:56 +01:00
self.sockobj = sockobj
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
def setup_stream(self):
sha1 = self.sockobj._get_sha1_auth()
self.incoming = []
self.incoming.append(self.sockobj._get_auth_response())
self.incoming.append(self.sockobj._get_request_buff(sha1, 0x00))
self.outgoing = []
self.outgoing.append(self.sockobj._get_auth_buff())
self.outgoing.append(self.sockobj._get_request_buff(sha1))
2011-08-02 05:13:56 +02:00
def switch_stream(self):
# Roles are reversed, client will be expecting server stream
# and server will be expecting client stream
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
temp = self.incoming
self.incoming = self.outgoing
self.outgoing = temp
def _recv(self, foo):
return self.incoming.pop(0)
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
def _send(self, data):
2013-01-01 19:36:56 +01:00
# This method is surrounded by a try block,
2011-08-02 05:13:56 +02:00
# we can't use assert here
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
if data != self.outgoing[0]:
2013-01-01 19:36:56 +01:00
print('FAILED SENDING TEST')
2011-08-02 05:13:56 +02:00
self.outgoing.pop(0)
class fake_idlequeue(Mock):
def __init__(self):
2013-01-01 19:36:56 +01:00
Mock.__init__(self)
2011-08-02 05:13:56 +02:00
def plug_idle(self, obj, writable=True, readable=True):
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
if readable:
obj.pollin()
if writable:
obj.pollout()
class TestSocks5(unittest.TestCase):
'''
Test class for Socks5
'''
def setUp(self):
streamhost = { 'host': None,
'port': 1,
'initiator' : None,
'target' : None}
2011-08-15 18:16:17 +02:00
queue = Mock()
queue.file_props = {}
2011-08-02 05:13:56 +02:00
#self.sockobj = Socks5Receiver(fake_idlequeue(), streamhost, None)
2011-08-15 18:16:17 +02:00
self.sockobj = Socks5Sender(fake_idlequeue(), None, 'server', Mock() ,
2013-01-01 19:36:56 +01:00
None, None, True, file_props={})
2011-08-02 05:13:56 +02:00
sock = fake_sock(self.sockobj)
self.sockobj._sock = sock
self.sockobj._recv = sock._recv
self.sockobj._send = sock._send
self.sockobj.state = 1
self.sockobj.connected = True
self.sockobj.pollend = self._pollend
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
# Something that the receiver needs
#self.sockobj.file_props['type'] = 'r'
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
# Something that the sender needs
self.sockobj.file_props = {}
self.sockobj.file_props['type'] = 'r'
2011-08-15 18:16:17 +02:00
self.sockobj.file_props['paused'] = ''
2011-08-02 05:13:56 +02:00
self.sockobj.queue = Mock()
self.sockobj.queue.process_result = self._pollend
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
def _pollend(self, foo = None, duu = None):
# This is a disconnect function
sys.exit("end of the road")
def _check_inout(self):
# Check if there isn't anything else to receive or send
2013-01-01 19:36:56 +01:00
sock = self.sockobj._sock
2011-08-02 05:13:56 +02:00
assert(sock.incoming == [])
assert(sock.outgoing == [])
2013-01-01 19:36:56 +01:00
2011-08-15 18:16:17 +02:00
def test_connection_server(self):
return
mocksock = self.sockobj._sock
mocksock.setup_stream()
2013-01-01 19:36:56 +01:00
#self.sockobj._sock.switch_stream()
2011-08-15 18:16:17 +02:00
s = socket.socket(2, 1, 6)
server = ('127.0.0.1', 28000)
s.connect(server)
2013-01-01 19:36:56 +01:00
2011-08-15 18:16:17 +02:00
s.send(mocksock.outgoing.pop(0))
2017-09-10 12:21:10 +02:00
self.assertEqual(s.recv(64), mocksock.incoming.pop(0))
2013-01-01 19:36:56 +01:00
2011-08-15 18:16:17 +02:00
s.send(mocksock.outgoing.pop(0))
2017-09-10 12:21:10 +02:00
self.assertEqual(s.recv(64), mocksock.incoming.pop(0))
2013-01-01 19:36:56 +01:00
2011-08-15 18:16:17 +02:00
def test_connection_client(self):
2013-01-01 19:36:56 +01:00
2011-08-15 18:16:17 +02:00
mocksock = self.sockobj._sock
mocksock.setup_stream()
2013-01-01 19:36:56 +01:00
mocksock.switch_stream()
2011-08-15 18:16:17 +02:00
s = socket.socket(10, 1, 6)
2013-01-01 19:36:56 +01:00
2011-08-15 18:16:17 +02:00
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
2013-01-01 19:36:56 +01:00
2011-08-15 18:16:17 +02:00
netadd = ('::', 28000, 0, 0)
s.bind(netadd)
s.listen(socket.SOMAXCONN)
(s, address) = s.accept()
2013-01-01 19:36:56 +01:00
2017-09-10 12:21:10 +02:00
self.assertEqual(s.recv(64), mocksock.incoming.pop(0))
2011-08-15 18:16:17 +02:00
s.send(mocksock.outgoing.pop(0))
2013-01-01 19:36:56 +01:00
2011-08-15 18:16:17 +02:00
buff = s.recv(64)
inco = mocksock.incoming.pop(0)
2017-09-10 12:21:10 +02:00
#self.assertEqual(s.recv(64), mocksock.incoming.pop(0))
2011-08-15 18:16:17 +02:00
s.send(mocksock.outgoing.pop(0))
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
def test_client_negoc(self):
return
self.sockobj._sock.setup_stream()
try:
self.sockobj.pollout()
except SystemExit:
pass
2013-01-01 19:36:56 +01:00
self._check_inout()
2011-08-02 05:13:56 +02:00
def test_server_negoc(self):
2011-08-15 18:16:17 +02:00
return
2011-08-02 05:13:56 +02:00
self.sockobj._sock.setup_stream()
2013-01-01 19:36:56 +01:00
self.sockobj._sock.switch_stream()
2011-08-02 05:13:56 +02:00
try:
self.sockobj.idlequeue.plug_idle(self.sockobj, False, True)
except SystemExit:
pass
self._check_inout()
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
if __name__ == '__main__':
2013-01-01 19:36:56 +01:00
2011-08-02 05:13:56 +02:00
unittest.main()