#!/usr/bin/python # Could be achieved via mocker, but this class is test-framework-agnostic + # Not strictly necessary, but certainly improves speed and isolation # from the host system state (which is irrelevant to this test). @staticmethod def gethostname(): - return 'TEST_HOST_NAME' + return "TEST_HOST_NAME" @staticmethod def gethostbyname(_): - return 'TEST_IP_ADDRESS' + return "TEST_IP_ADDRESS"

Line too long (96 > 79 characters) (E501)

#!/usr/bin/python + # Could be achieved via mocker, but this class is test-framework-agnostic class MockSocketObject: # Not strictly necessary, but certainly improves speed and isolation # from the host system state (which is irrelevant to this test)

Line too long (90 > 79 characters) (E501)

import sys import worker.service as service # listen on all interfaces -host = '' +host = "" def main(args): if len(args) != 1: raise TypeError("Incorrect number of arguments. Please specify a port number")

Line too long (96 > 79 characters) (E501)

port = int(args[0]) my_socket = service.listen_to(host, port) service.handle_requests(my_socket) -if __name__ == '__main__': +if __name__ == "__main__": main(sys.argv[1:])

Line too long (105 > 79 characters) (E501)

my_socket.close() def _process_request(connection_dependency, address): request = connection_dependency.recv(1024).decode() - print('Connection from {} - received {}'.format(address, request)) + print("Connection from {} - received {}".format(address, request)) return _prepare_response() def _prepare_response(): hostinfo = _get_host_info(socket) - response = 'HTTP/1.0 200 OK\n\nHello from {} ({})'.format(*hostinfo) + response = "HTTP/1.0 200 OK\n\nHello from {} ({})".format(*hostinfo) return response.encode() def _get_ip_address(hostname, get_host_by_name_dependency): try: return get_host_by_name_dependency(hostname) except socket.gaierror as e: print("Could not get IP address for {}:\n{}".format(hostname, e)) - return 'UNKNOWN_IP' + return "UNKNOWN_IP" @functools.cache def _get_host_info(socket_dependency): hostname = socket_dependency.gethostname()

Line too long (110 > 79 characters) (E501)

@pytest.mark.unit def test_main_service_integration(mocker): # Arrange - mocks = { - 'listen_to': mocker.DEFAULT, - 'handle_requests': mocker.DEFAULT - } - service_mock = mocker.patch.multiple('worker.service', **mocks) + mocks = {"listen_to": mocker.DEFAULT, "handle_requests": mocker.DEFAULT} + service_mock = mocker.patch.multiple("worker.service", **mocks) # Act _start_sandbox(ANY_PORT, 0) # Assert

Line too long (101 > 79 characters) (E501)

class TestPerformance: SANDBOX_WARMUP_DURATION = 1 / 1000 MAX_PERFORMANCE_TEST_DURATION = 2 - PERFORMANCE_TEST_TIMEOUT_THRESHOLD = SANDBOX_WARMUP_DURATION + MAX_PERFORMANCE_TEST_DURATION + PERFORMANCE_TEST_TIMEOUT_THRESHOLD = ( + SANDBOX_WARMUP_DURATION + MAX_PERFORMANCE_TEST_DURATION + ) PERFORMANCE_REQUEST_COUNT = 5000 hostinfo = None response_list = None @pytest.fixture(scope="class", autouse=True) def sandbox_lifecycle_warmstart(self, class_mocker): # Arrange - spy = class_mocker.spy(service, 'listen_to') + spy = class_mocker.spy(service, "listen_to") _start_sandbox(ANY_PORT, self.SANDBOX_WARMUP_DURATION) hostinfo = spy.spy_return.getsockname() self.__class__.hostinfo = hostinfo self.__class__.response_list = [] _fetch_response(hostinfo, "TEST_ANY_REQUEST")

Line too long (107 > 79 characters) (E501)

def test_main_response(self): # Act response = _fetch_response(self.hostinfo, "TEST_ANY_REQUEST") # Assert - assert "Hello from" in response, "Received incorrect response {}".format(response) + assert "Hello from" in response, "Received incorrect response {}".format( + response + ) @pytest.mark.system @pytest.mark.timeout(PERFORMANCE_TEST_TIMEOUT_THRESHOLD) def test_performance_main_throughput(self): # Act start_timestamp = time.monotonic() _request_repeatedly(self.hostinfo, self.PERFORMANCE_REQUEST_COUNT) elapsed = time.monotonic() - start_timestamp - print("Transactions per second = {} (i.e. {} request/response iterations in {} seconds)" - .format(self.PERFORMANCE_REQUEST_COUNT / elapsed, self.PERFORMANCE_REQUEST_COUNT, elapsed)) + print( + "Transactions per second = {} (i.e. {} request/response iterations in {} seconds)".format( + self.PERFORMANCE_REQUEST_COUNT / elapsed, + self.PERFORMANCE_REQUEST_COUNT, + elapsed, + ) + ) # Assert - assert elapsed < self.MAX_PERFORMANCE_TEST_DURATION, "Execution took {} seconds (max is {} seconds)" \ - .format(elapsed, self.MAX_PERFORMANCE_TEST_DURATION) + assert ( + elapsed < self.MAX_PERFORMANCE_TEST_DURATION + ), "Execution took {} seconds (max is {} seconds)".format( + elapsed, self.MAX_PERFORMANCE_TEST_DURATION + ) @pytest.mark.system @pytest.mark.timeout(PERFORMANCE_TEST_TIMEOUT_THRESHOLD) def test_performance_main_success_rate(self): # Act - _request_repeatedly(self.hostinfo, self.PERFORMANCE_REQUEST_COUNT, self.response_list.append) + _request_repeatedly( + self.hostinfo, self.PERFORMANCE_REQUEST_COUNT, self.response_list.append + ) # Assert response_count = len(self.response_list) - assert response_count == self.PERFORMANCE_REQUEST_COUNT, "Expected {} responses, but received {}" \ - .format(self.PERFORMANCE_REQUEST_COUNT, response_count) + assert ( + response_count == self.PERFORMANCE_REQUEST_COUNT + ), "Expected {} responses, but received {}".format( + self.PERFORMANCE_REQUEST_COUNT, response_count + ) def _request_repeatedly(hostinfo, request_count, on_response=lambda noop: None): for _ in range(request_count): response = _fetch_response(hostinfo, "TEST_PERFORMANCE_ANY_REQUEST")

Line too long (80 > 79 characters) (E501)

s.connect(hostinfo) return s def _start_sandbox(port, startup_time): - threading.Thread( - target=sandbox.main, - args=[port], - daemon=True - ).start() + threading.Thread(target=sandbox.main, args=[port], daemon=True).start() time.sleep(startup_time)

Line too long (92 > 79 characters) (E501)

from . import mock_socket_object from . import mock_socket_package import worker.service as service unsupported_platform = { - 'condition': platform.system() != 'Linux', - 'reason': "Test only produces meaningful results in a production-like Linux environment" + "condition": platform.system() != "Linux", + "reason": "Test only produces meaningful results in a production-like Linux environment", } @pytest.mark.skipif(**unsupported_platform) @pytest.mark.system

Line too long (80 > 79 characters) (E501)

# Act result = service._get_ip_address("TEST_HOSTNAME", receive_hostname) # Assert - assert result == 'UNKNOWN_IP' + assert result == "UNKNOWN_IP" # Tests the implementation instead of the result. Low ROI to create and maintain this :( @pytest.mark.integration def test_listen_to_port(mocker): # Arrange socket_object_mock = mock_socket_object.MockSocketObject() - socket_method_mock = mocker.patch('socket.socket', return_value=socket_object_mock) - bind_method_mock = mocker.patch.object(socket_object_mock, 'bind', return_value=None) - listen_method_mock = mocker.patch.object(socket_object_mock, 'listen', return_value=None) + socket_method_mock = mocker.patch("socket.socket", return_value=socket_object_mock) + bind_method_mock = mocker.patch.object( + socket_object_mock, "bind", return_value=None + ) + listen_method_mock = mocker.patch.object( + socket_object_mock, "listen", return_value=None + ) # Act service.listen_to("localhost", 0) # Assert

Line too long (88 > 79 characters) (E501)

Line too long (87 > 79 characters) (E501)

@pytest.mark.unit def test_listen_to_port_returns_object(mocker): # Arrange - socket_method_mock = mocker.patch('socket.socket') + socket_method_mock = mocker.patch("socket.socket") socket_object_mock = mock_socket_object.MockSocketObject() - mocker.patch.object(socket_object_mock, 'bind', return_value=None) - mocker.patch.object(socket_object_mock, 'listen', return_value=None) + mocker.patch.object(socket_object_mock, "bind", return_value=None) + mocker.patch.object(socket_object_mock, "listen", return_value=None) socket_method_mock.return_value = socket_object_mock # Act result_socket = service.listen_to("localhost", 0)

Line too long (89 > 79 characters) (E501)

@pytest.mark.unit def test_prepared_response_includes_required_info(mocker): # Arrange - required_info = ('TEST_IP_ADDRESS', 'TEST_HOST_NAME') - mocker.patch('worker.service._get_host_info', return_value=required_info) + required_info = ("TEST_IP_ADDRESS", "TEST_HOST_NAME") + mocker.patch("worker.service._get_host_info", return_value=required_info) # Act response = service._prepare_response() # Assert contains_required_info = all(x.encode() in response for x in required_info) - assert contains_required_info, "Could not find both '{}' and '{}' in the response '{}'"\ - .format(*required_info, response) + assert ( + contains_required_info + ), "Could not find both '{}' and '{}' in the response '{}'".format( + *required_info, response + ) @pytest.mark.unit def test_get_hostinfo_returns_tuple(): # Act

Line too long (93 > 79 characters) (E501)

Line too long (92 > 79 characters) (E501)

Line too long (88 > 79 characters) (E501)

Line too long (86 > 79 characters) (E501)