vee1e/mitmproxy
1
0
Fork 0
mirror of https://github.com/vee1e/mitmproxy.git synced 2026-09-04 03:37:17 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

connections tests: fix leaking sockets

Browse source
This commit is contained in:
Thomas Kriechbaumer 2017-05-24 13:21:58 +02:00
parent 928085c597
commit 2faaa0b2a2
2 changed files with 9 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

5
test/mitmproxy/net/tservers.py
View file

@ -16,9 +16,6 @@ class _ServerThread(threading.Thread):
def run(self):
self.server.serve_forever()
def shutdown(self):
self.server.shutdown()
class _TServer(tcp.TCPServer):
@ -104,7 +101,7 @@ class ServerTestBase:
@classmethod
def teardown_class(cls):
cls.server.shutdown()
cls.server.server.shutdown()
def teardown(self):
self.server.server.wait_for_silence()

12
test/mitmproxy/test_connections.py
View file

@ -140,16 +140,17 @@ class TestServerConnection:
assert d.last_log()
c.finish()
c.close()
d.shutdown()
def test_terminate_error(self):
d = test.Daemon()
c = connections.ServerConnection((d.IFACE, d.port))
c.connect()
c.close()
c.connection = mock.Mock()
c.connection.recv = mock.Mock(return_value=False)
c.connection.flush = mock.Mock(side_effect=exceptions.TcpDisconnect)
c.finish()
d.shutdown()
def test_sni(self):
@ -194,22 +195,25 @@ class TestClientConnectionTLS:
s = socket.create_connection(address)
s = ctx.wrap_socket(s, server_hostname=sni)
s.send(b'foobar')
s.shutdown(socket.SHUT_RDWR)
s.close()
threading.Thread(target=client_run).start()
connection, client_address = sock.accept()
c = connections.ClientConnection(connection, client_address, None)
cert = tutils.test_data.path("mitmproxy/net/data/server.crt")
with open(tutils.test_data.path("mitmproxy/net/data/server.key")) as f:
raw_key = f.read()
key = OpenSSL.crypto.load_privatekey(
OpenSSL.crypto.FILETYPE_PEM,
open(tutils.test_data.path("mitmproxy/net/data/server.key"), "rb").read())
raw_key)
c.convert_to_ssl(cert, key)
assert c.connected()
assert c.sni == sni
assert c.tls_established
assert c.rfile.read(6) == b'foobar'
c.finish()
sock.close()
class TestServerConnectionTLS(tservers.ServerTestBase):
@ -222,7 +226,7 @@ class TestServerConnectionTLS(tservers.ServerTestBase):
@pytest.mark.parametrize("clientcert", [
None,
tutils.test_data.path("mitmproxy/data/clientcert"),
os.path.join(tutils.test_data.path("mitmproxy/data/clientcert"), "client.pem"),
tutils.test_data.path("mitmproxy/data/clientcert/client.pem"),
])
def test_tls(self, clientcert):
c = connections.ServerConnection(("127.0.0.1", self.port))