Home > front end >  Flask make_server always raises "OSError: [Errno 98] Address already in use"
Flask make_server always raises "OSError: [Errno 98] Address already in use"

Time:12-20

I am trying to write unit tests for some Python code I am working on, and some of this code contacts an API when it is done. I am trying to have a simple Flask API running to mock this API and check that the code sends the correct info. Here is the code:

import unittest
import time
from threading import Thread
from flask import Flask
from flask_restx import Api, Resource
from werkzeug.serving import make_server

mock_app = Flask(__name__)
mock_api = Api(mock_app)


# Mock API
data_in = []
data_out = ""
result_code = 200

@mock_api.route('/jobs')
class MockAPI(Resource):
    def post(self):
        global data_in, data_out, result_code
        data_in.append(mock_api.payload)
        return data_out, result_code


# Unit test class
class TestClass(unittest.TestCase):
    
    def __init__(self, arg):
        super().__init__(arg)
        # Some needed fields
        # ...

        # Mock API Server
        self.mock_server = make_server('localhost', 6000, mock_wfm)
        self.mock_server_thread = Thread(target = self.mock_wfm.serve_forever)

The line with the call to make_server is the one causing the exception. Commands like lsof -i :6000 don't return anything, changing the address or port doesn't fix anything either.

Edit:

After adding a simple print before the faulty line, I discovered that the code was in fact called twice, causing the error. I don't know why yet.

CodePudding user response:

The problem was that the __init__ method is apparently called for every test, so it was failing at the second one every time.

The solution is to put the creation of the server and the Thread in the setUp method, and close them in the tearDown.

    def setUp(self):
        self.mock_server = make_server('localhost', 6000, mock_wfm)
        self.mock_server_thread = Thread(target = self.mock_server.serve_forever)
        self.mock_server_thread.start()
        time.sleep(1)

    def tearDown(self):
        self.mock_server.shutdown()
        self.mock_server_thread.join()
  • Related