mirror of
https://github.com/nikdoof/python-ts3.git
synced 2025-12-17 11:59:27 +00:00
Add some basic network side tests.
This commit is contained in:
53
ts3/test.py
53
ts3/test.py
@@ -1,5 +1,8 @@
|
|||||||
import unittest
|
import unittest
|
||||||
from ts3.protocol import TS3Proto
|
import socket
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
from ts3.protocol import TS3Proto, ConnectionError, NoConnectionError
|
||||||
|
|
||||||
|
|
||||||
class TS3ProtoTest(unittest.TestCase):
|
class TS3ProtoTest(unittest.TestCase):
|
||||||
@@ -69,6 +72,54 @@ class TS3ProtoTest(unittest.TestCase):
|
|||||||
self.assertEqual(self.ts3.parse_data(data), parsed)
|
self.assertEqual(self.ts3.parse_data(data), parsed)
|
||||||
|
|
||||||
|
|
||||||
|
def dummy_ts3(event, sock):
|
||||||
|
"""
|
||||||
|
A simple dummy TS3 server to run tests against
|
||||||
|
"""
|
||||||
|
sock.listen(5)
|
||||||
|
event.set()
|
||||||
|
try:
|
||||||
|
conn, addr = sock.accept()
|
||||||
|
conn.send('TS3\n\r')
|
||||||
|
except socket.timeout:
|
||||||
|
pass
|
||||||
|
finally:
|
||||||
|
sock.close()
|
||||||
|
event.set()
|
||||||
|
|
||||||
|
|
||||||
|
class TS3ProtoNetworkTests(unittest.TestCase):
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.evt = threading.Event()
|
||||||
|
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||||
|
self.sock.settimeout(3)
|
||||||
|
self.sock.bind(('127.0.0.1', 0))
|
||||||
|
self.port = self.sock.getsockname()[1]
|
||||||
|
threading.Thread(target=dummy_ts3, args=(self.evt, self.sock)).start()
|
||||||
|
self.evt.wait()
|
||||||
|
self.evt.clear()
|
||||||
|
time.sleep(.1)
|
||||||
|
self.ts3 = TS3Proto()
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
self.evt.wait()
|
||||||
|
if hasattr(self.ts3._telnet, 'sock'):
|
||||||
|
self.ts3._telnet.sock.close()
|
||||||
|
|
||||||
|
def testConnect(self):
|
||||||
|
self.assertTrue(self.ts3.connect('127.0.0.1', self.port))
|
||||||
|
self.assertTrue(self.ts3.is_connected())
|
||||||
|
self.assertIsNone(self.ts3.check_connection())
|
||||||
|
|
||||||
|
def testConnectFail(self):
|
||||||
|
self.assertRaises(ConnectionError, self.ts3.connect, '127.0.0.1', 9911)
|
||||||
|
|
||||||
|
def testNoConnection(self):
|
||||||
|
self.assertFalse(self.ts3.is_connected())
|
||||||
|
self.assertRaises(NoConnectionError, self.ts3.check_connection)
|
||||||
|
|
||||||
|
|
||||||
def suite():
|
def suite():
|
||||||
suite = unittest.TestSuite()
|
suite = unittest.TestSuite()
|
||||||
suite.addTest(unittest.makeSuite(TS3ProtoTest))
|
suite.addTest(unittest.makeSuite(TS3ProtoTest))
|
||||||
|
|||||||
Reference in New Issue
Block a user