How to test a running web server?

user_12016688 0
Viewed 163

I'm writing a tcp server. It has an infinite loop to receive and respond requests. Something like this:

class TCPServer:
    ...

    def start(self):
        try:
            while True:
                connection, addr = self.socket.accept()
                logging.info(f'Address {self.format_address(addr)} connected.')
                data = connection.recv(self.RECEIVING_SIZE)
                logging.info(data)
                response = self.handle_request(data)
                connection.sendall(data)
                connection.close()
        except KeyboardInterrupt:
            self.socket.close()

The whole code is here. My question is how can I test it? I've wroten a test which never ends until I press ctrl-C.

class TestTCPServer:

    @staticmethod
    def _start_server(server):
        thr = threading.Thread(target=server.start, args=(), kwargs={})
        thr.start()
        return thr
    
    def test_connecting_to_tcp_server(self, client):
        host = '127.0.0.1'
        port = 12345
        tcp_server = TCPServer(host=host, port=port)
        running_server = self._start_server(tcp_server)
        client.connect((host, port))
        running_server.join()

How can I write a test to terminate after testing the connection?

pytest testing sockets python
2 Answers

Not sure if I fully understood your question. But this code connects to the server, sends data and prints the data it receives. Than the program is finished.

import socket
s = socket.socket()
s.connect(("127.0.0.1", 8080))
s.send("test".encode())
data = s.recv(1024).decode()
print("The server sent " + data)
s.close()

With your server, this client outputs:

The server sent test

And the server outputs:

INFO:root:Starting litening at: 127.0.0.1:8080

INFO:root:Address 127.0.0.1:59208 connected.

INFO:root:b'test'

user_69414430

I found a way to fix this issue. I can set the deamon parameter to True. This cause all threads to terminate when the main thread exits.

class TestTCPServer:

    @staticmethod
    def _start_server(server):
        thr = threading.Thread(target=server.start, args=(), kwargs={})
        thr.start()
        return thr
    
    def test_connecting_to_tcp_server(self, client):
        host = '127.0.0.1'
        port = 12345
        tcp_server = TCPServer(host=host, port=port)
        running_server = self._start_server(tcp_server)
        client.connect((host, port))
        running_server.join()

The problem with this approach is that is does not close the socket. So the port will be in used after the tests passed for a while.

user_120166880
Related