Skip to content

Commit 580fd4d

Browse files
author
Martin Vrachev
committed
Apply linters on tests/test_utils.py
Apply all 4 linters on tests/test_utils.py, so we can lint it in the future and not rename and exclude it. As we are going to use part or most of tests/utils.py after we remove the test files testing the old code, then we would need to keep the test file testing utils.py - test_utils.py Black and isort changes where automatically made. The only manual changes are: - pylint disable once in can_connect - remove TestServerProcess.tearDown and instead clean it manually in the tests, so we don't have to define server_process_handler as a class variable and assert it's type in every test - move can_connect outside TestServerProcess as it doesn't use self anymore - add type annotations - function docstrings Signed-off-by: Martin Vrachev <mvrachev@vmware.com>
1 parent d203834 commit 580fd4d

1 file changed

Lines changed: 92 additions & 88 deletions

File tree

‎tests/test_utils.py‎

Lines changed: 92 additions & 88 deletions
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)