summaryrefslogtreecommitdiff
path: root/tests/test_socket.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_socket.py')
-rw-r--r--tests/test_socket.py42
1 files changed, 15 insertions, 27 deletions
diff --git a/tests/test_socket.py b/tests/test_socket.py
index a8c8595..b801b00 100644
--- a/tests/test_socket.py
+++ b/tests/test_socket.py
@@ -19,7 +19,7 @@ PEXPECT LICENSE
'''
import pexpect
-from pexpect import fdpexpect
+from pexpect import socket_pexpect
import unittest
from . import PexpectTestCase
import multiprocessing
@@ -133,12 +133,16 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase):
pass
exit(0)
+ def spawn(self, socket, timeout=30, use_poll=False):
+ """override me with other ways of spawning on a socket"""
+ return socket_pexpect.SocketSpawn(socket, timeout=timeout, use_poll=use_poll)
+
def socket_fn(self, timed_out, all_read):
result = 0
try:
sock = socket.socket(self.af, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
- session = fdpexpect.fdspawn(sock, timeout=10)
+ session = self.spawn(sock, timeout=10)
# Get all data from server
session.read_nonblocking(size=4096)
all_read.set()
@@ -152,7 +156,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase):
def test_socket(self):
sock = socket.socket(self.af, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
- session = fdpexpect.fdspawn(sock.fileno(), timeout=10)
+ session = self.spawn(sock, timeout=10)
session.expect(self.prompt1)
self.assertEqual(session.before, self.motd)
session.send(self.enter)
@@ -166,7 +170,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase):
def test_socket_with_write(self):
sock = socket.socket(self.af, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
- session = fdpexpect.fdspawn(sock.fileno(), timeout=10)
+ session = self.spawn(sock, timeout=10)
session.expect(self.prompt1)
self.assertEqual(session.before, self.motd)
session.write(self.enter)
@@ -177,19 +181,11 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase):
session.expect(pexpect.EOF)
self.assertEqual(session.before, b'')
- def test_not_int(self):
- with self.assertRaises(pexpect.ExceptionPexpect):
- session = fdpexpect.fdspawn('bogus', timeout=10)
-
- def test_not_file_descriptor(self):
- with self.assertRaises(pexpect.ExceptionPexpect):
- session = fdpexpect.fdspawn(-1, timeout=10)
-
def test_timeout(self):
with self.assertRaises(pexpect.TIMEOUT):
sock = socket.socket(self.af, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
- session = fdpexpect.fdspawn(sock, timeout=10)
+ session = self.spawn(sock, timeout=10)
session.expect(b'Bogus response')
def test_interrupt(self):
@@ -223,7 +219,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase):
def test_maxread(self):
sock = socket.socket(self.af, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
- session = fdpexpect.fdspawn(sock.fileno(), timeout=10)
+ session = self.spawn(sock, timeout=10)
session.maxread = 1100
session.expect(self.prompt1)
self.assertEqual(session.before, self.motd)
@@ -238,7 +234,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase):
def test_fd_isalive(self):
sock = socket.socket(self.af, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
- session = fdpexpect.fdspawn(sock.fileno(), timeout=10)
+ session = self.spawn(sock, timeout=10)
assert session.isalive()
sock.close()
assert not session.isalive(), "Should not be alive after close()"
@@ -246,7 +242,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase):
def test_fd_isalive_poll(self):
sock = socket.socket(self.af, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
- session = fdpexpect.fdspawn(sock.fileno(), timeout=10, use_poll=True)
+ session = self.spawn(sock, timeout=10, use_poll=True)
assert session.isalive()
sock.close()
assert not session.isalive(), "Should not be alive after close()"
@@ -254,27 +250,19 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase):
def test_fd_isatty(self):
sock = socket.socket(self.af, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
- session = fdpexpect.fdspawn(sock.fileno(), timeout=10)
+ session = self.spawn(sock, timeout=10)
assert not session.isatty()
session.close()
def test_fd_isatty_poll(self):
sock = socket.socket(self.af, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
- session = fdpexpect.fdspawn(sock.fileno(), timeout=10, use_poll=True)
+ session = self.spawn(sock, timeout=10, use_poll=True)
assert not session.isatty()
session.close()
- def test_fileobj(self):
- sock = socket.socket(self.af, socket.SOCK_STREAM)
- sock.connect((self.host, self.port))
- session = fdpexpect.fdspawn(sock, timeout=10) # Should get the fileno from the socket
- session.expect(self.prompt1)
- session.close()
- assert not session.isalive()
- session.close() # Smoketest - should be able to call this again
if __name__ == '__main__':
unittest.main()
-suite = unittest.makeSuite(ExpectTestCase, 'test')
+suite = unittest.TestLoader().loadTestsFromTestCase(ExpectTestCase)