Skip to content

Commit aebd8b0

Browse files
lukpuehlukpueh
authored
Merge pull request #1708 from MVrachev/prepare-for-lint
Lint additional four files in the tests folder
2 parents 2db0cd6 + d748bd3 commit aebd8b0

File tree

4 files changed

+400
-374
lines changed

4 files changed

+400
-374
lines changed

‎tests/aggregate_tests.py

+13-10
Original file line numberDiff line numberDiff line change
@@ -29,13 +29,16 @@
2929
import sys
3030
import unittest
3131

32-
if __name__ == '__main__':
33-
suite = unittest.TestLoader().discover(".")
34-
all_tests_passed = unittest.TextTestRunner(
35-
verbosity=1, buffer=True).run(suite).wasSuccessful()
36-
37-
if not all_tests_passed:
38-
sys.exit(1)
39-
40-
else:
41-
sys.exit(0)
32+
if __name__ == "__main__":
33+
suite = unittest.TestLoader().discover(".")
34+
all_tests_passed = (
35+
unittest.TextTestRunner(verbosity=1, buffer=True)
36+
.run(suite)
37+
.wasSuccessful()
38+
)
39+
40+
if not all_tests_passed:
41+
sys.exit(1)
42+
43+
else:
44+
sys.exit(0)

‎tests/simple_server.py

+17-15
Original file line numberDiff line numberDiff line change
@@ -25,40 +25,42 @@
2525
http://docs.python.org/library/simplehttpserver.html#module-SimpleHTTPServer
2626
"""
2727

28-
import sys
29-
import random
3028
import socketserver
29+
import sys
3130
from http.server import SimpleHTTPRequestHandler
31+
from typing import Type, Union
3232

3333

3434
class QuietHTTPRequestHandler(SimpleHTTPRequestHandler):
35-
"""A SimpleHTTPRequestHandler that does not write incoming requests to
36-
stderr. """
37-
def log_request(self, code='-', size='-'):
38-
pass
35+
"""A SimpleHTTPRequestHandler that does not write incoming requests to
36+
stderr."""
37+
38+
def log_request(
39+
self, code: Union[int, str] = "-", size: Union[int, str] = "-"
40+
) -> None:
41+
pass
42+
3943

4044
# NOTE: On Windows/Python2 tests that use this simple_server.py in a
4145
# subprocesses hang after a certain amount of requests (~68), if a PIPE is
4246
# passed as Popen's stderr argument. This problem doesn't emerge if
4347
# we silence the HTTP messages.
4448
# If you decide to receive the HTTP messages, then this bug
4549
# could reappear.
46-
use_quiet_http_request_handler = True
4750

48-
if len(sys.argv) > 2:
49-
use_quiet_http_request_handler = sys.argv[2]
51+
# pylint: disable=invalid-name
52+
handler: Type[Union[SimpleHTTPRequestHandler, QuietHTTPRequestHandler]]
5053

51-
if use_quiet_http_request_handler:
52-
handler = QuietHTTPRequestHandler
54+
if len(sys.argv) > 2 and sys.argv[2]:
55+
handler = QuietHTTPRequestHandler
5356
else:
54-
handler = SimpleHTTPRequestHandler
57+
handler = SimpleHTTPRequestHandler
5558

5659
# Allow re-use so you can re-run tests as often as you want even if the
5760
# tests re-use ports. Otherwise TCP TIME-WAIT prevents reuse for ~1 minute
5861
socketserver.TCPServer.allow_reuse_address = True
5962

60-
httpd = socketserver.TCPServer(('localhost', 0), handler)
61-
port_message = 'bind succeeded, server port is: ' \
62-
+ str(httpd.server_address[1])
63+
httpd = socketserver.TCPServer(("localhost", 0), handler)
64+
port_message = "bind succeeded, server port is: " + str(httpd.server_address[1])
6365
print(port_message)
6466
httpd.serve_forever()

‎tests/test_utils.py

+92-88
Original file line numberDiff line numberDiff line change
@@ -20,104 +20,108 @@
2020
Provide tests for some of the functions in utils.py module.
2121
"""
2222

23-
import os
2423
import logging
25-
import unittest
24+
import os
2625
import socket
2726
import sys
28-
29-
import tuf.unittest_toolbox as unittest_toolbox
27+
import unittest
3028

3129
from tests import utils
30+
from tuf import unittest_toolbox
3231

3332
logger = logging.getLogger(__name__)
3433

35-
class TestServerProcess(unittest_toolbox.Modified_TestCase):
36-
37-
def tearDown(self):
38-
# Make sure we are calling clean on existing attribute.
39-
if hasattr(self, 'server_process_handler'):
40-
self.server_process_handler.clean()
41-
4234

43-
def can_connect(self):
35+
def can_connect(port: int) -> bool:
36+
"""Check if a socket can connect on the given port"""
4437
try:
45-
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
46-
sock.connect(('localhost', self.server_process_handler.port))
47-
return True
48-
except:
49-
return False
38+
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
39+
sock.connect(("localhost", port))
40+
return True
41+
# pylint: disable=broad-except
42+
except Exception:
43+
return False
5044
finally:
51-
# The process will always enter in finally even we return.
52-
if sock:
53-
sock.close()
54-
55-
56-
def test_simple_server_startup(self):
57-
# Test normal case
58-
self.server_process_handler = utils.TestServerProcess(log=logger)
59-
60-
# Make sure we can connect to the server
61-
self.assertTrue(self.can_connect())
62-
63-
64-
def test_simple_https_server_startup(self):
65-
# Test normal case
66-
good_cert_path = os.path.join('ssl_certs', 'ssl_cert.crt')
67-
self.server_process_handler = utils.TestServerProcess(log=logger,
68-
server='simple_https_server.py', extra_cmd_args=[good_cert_path])
69-
70-
# Make sure we can connect to the server
71-
self.assertTrue(self.can_connect())
72-
self.server_process_handler.clean()
73-
74-
# Test when no cert file is provided
75-
self.server_process_handler = utils.TestServerProcess(log=logger,
76-
server='simple_https_server.py')
45+
# The process will always enter in finally even after return.
46+
if sock:
47+
sock.close()
7748

78-
# Make sure we can connect to the server
79-
self.assertTrue(self.can_connect())
80-
self.server_process_handler.clean()
8149

82-
# Test with a non existing cert file.
83-
non_existing_cert_path = os.path.join('ssl_certs', 'non_existing.crt')
84-
self.server_process_handler = utils.TestServerProcess(log=logger,
85-
server='simple_https_server.py',
86-
extra_cmd_args=[non_existing_cert_path])
87-
88-
# Make sure we can connect to the server
89-
self.assertTrue(self.can_connect())
90-
91-
92-
def test_slow_retrieval_server_startup(self):
93-
# Test normal case
94-
self.server_process_handler = utils.TestServerProcess(log=logger,
95-
server='slow_retrieval_server.py')
96-
97-
# Make sure we can connect to the server
98-
self.assertTrue(self.can_connect())
99-
100-
101-
def test_cleanup(self):
102-
# Test normal case
103-
self.server_process_handler = utils.TestServerProcess(log=logger,
104-
server='simple_server.py')
105-
106-
self.server_process_handler.clean()
107-
108-
# Check if the process has successfully been killed.
109-
self.assertFalse(self.server_process_handler.is_process_running())
110-
111-
112-
def test_server_exit_before_timeout(self):
113-
with self.assertRaises(utils.TestServerProcessError):
114-
utils.TestServerProcess(logger, server='non_existing_server.py')
115-
116-
# Test starting a server which immediately exits."
117-
with self.assertRaises(utils.TestServerProcessError):
118-
utils.TestServerProcess(logger, server='fast_server_exit.py')
119-
120-
121-
if __name__ == '__main__':
122-
utils.configure_test_logging(sys.argv)
123-
unittest.main()
50+
class TestServerProcess(unittest_toolbox.Modified_TestCase):
51+
"""Test functionality provided in TestServerProcess from tests/utils.py."""
52+
53+
def test_simple_server_startup(self) -> None:
54+
# Test normal case
55+
server_process_handler = utils.TestServerProcess(log=logger)
56+
57+
# Make sure we can connect to the server
58+
self.assertTrue(can_connect(server_process_handler.port))
59+
server_process_handler.clean()
60+
61+
def test_simple_https_server_startup(self) -> None:
62+
# Test normal case
63+
good_cert_path = os.path.join("ssl_certs", "ssl_cert.crt")
64+
server_process_handler = utils.TestServerProcess(
65+
log=logger,
66+
server="simple_https_server.py",
67+
extra_cmd_args=[good_cert_path],
68+
)
69+
70+
# Make sure we can connect to the server
71+
self.assertTrue(can_connect(server_process_handler.port))
72+
server_process_handler.clean()
73+
74+
# Test when no cert file is provided
75+
server_process_handler = utils.TestServerProcess(
76+
log=logger, server="simple_https_server.py"
77+
)
78+
79+
# Make sure we can connect to the server
80+
self.assertTrue(can_connect(server_process_handler.port))
81+
server_process_handler.clean()
82+
83+
# Test with a non existing cert file.
84+
non_existing_cert_path = os.path.join("ssl_certs", "non_existing.crt")
85+
server_process_handler = utils.TestServerProcess(
86+
log=logger,
87+
server="simple_https_server.py",
88+
extra_cmd_args=[non_existing_cert_path],
89+
)
90+
91+
# Make sure we can connect to the server
92+
self.assertTrue(can_connect(server_process_handler.port))
93+
server_process_handler.clean()
94+
95+
def test_slow_retrieval_server_startup(self) -> None:
96+
# Test normal case
97+
server_process_handler = utils.TestServerProcess(
98+
log=logger, server="slow_retrieval_server.py"
99+
)
100+
101+
# Make sure we can connect to the server
102+
self.assertTrue(can_connect(server_process_handler.port))
103+
server_process_handler.clean()
104+
105+
def test_cleanup(self) -> None:
106+
# Test normal case
107+
server_process_handler = utils.TestServerProcess(
108+
log=logger, server="simple_server.py"
109+
)
110+
111+
server_process_handler.clean()
112+
113+
# Check if the process has successfully been killed.
114+
self.assertFalse(server_process_handler.is_process_running())
115+
116+
def test_server_exit_before_timeout(self) -> None:
117+
with self.assertRaises(utils.TestServerProcessError):
118+
utils.TestServerProcess(logger, server="non_existing_server.py")
119+
120+
# Test starting a server which immediately exits."
121+
with self.assertRaises(utils.TestServerProcessError):
122+
utils.TestServerProcess(logger, server="fast_server_exit.py")
123+
124+
125+
if __name__ == "__main__":
126+
utils.configure_test_logging(sys.argv)
127+
unittest.main()

0 commit comments

Comments
 (0)