walkero
/
speedtest-cli
1
0
Fork 0
Code Issues Pull Requests Projects Releases 1 Wiki Activity

Create base class SocketTestBase to dedupe code

This commit is contained in:
Matt Martz 2018-03-23 09:37:38 -05:00
parent fedf42e838
commit ced2890261
1 changed files with 4 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
speedtest.py
View File

@ -758,7 +758,7 @@ class HTTPDownloader(threading.Thread):
pass
class SocketDownloader(threading.Thread):
class SocketTestBase(threading.Thread):
def __init__(self, i, address, size, start, timeout, shutdown_event=None,
source_address=None):
threading.Thread.__init__(self)
@ -780,6 +780,8 @@ class SocketDownloader(threading.Thread):
source_address=source_address
)
class SocketDownloader(SocketTestBase):
def run(self):
try:
if (timeit.default_timer() - self.starttime) <= self.timeout:
@ -908,28 +910,7 @@ class HTTPUploader(threading.Thread):
self.result = self.request.data.total
class SocketUploader(threading.Thread):
def __init__(self, i, address, size, start, timeout, shutdown_event=None,
source_address=None):
threading.Thread.__init__(self)
self.result = 0
self.starttime = start
self.timeout = timeout
self.i = i
self.size = size
self.remaining = self.size
if shutdown_event:
self._shutdown_event = shutdown_event
else:
self._shutdown_event = FakeShutdownEvent()
self.sock = connection_factory(
address,
timeout=timeout,
source_address=source_address
)
class SocketUploader(SocketTestBase):
def run(self):
try:
if (timeit.default_timer() - self.starttime) <= self.timeout: