Simple mock server for testing using Flask

README.md

Mock server for testing using flask

License

MIT

Prepare

Install flask and requests to run this example.

mockserver.py

import requests

from flask import Flask, jsonify
from threading import Thread


class MockServer(Thread):
    def __init__(self, port=5000):
        super().__init__()
        self.port = port
        self.app = Flask(__name__)
        self.url = "http://localhost:%s" % self.port

        self.app.add_url_rule("/shutdown", view_func=self._shutdown_server)

    def _shutdown_server(self):
        from flask import request
        if not 'werkzeug.server.shutdown' in request.environ:
            raise RuntimeError('Not running the development server')
        request.environ['werkzeug.server.shutdown']()
        return 'Server shutting down...'

    def shutdown_server(self):
        requests.get("http://localhost:%s/shutdown" % self.port)
        self.join()

    def add_callback_response(self, url, callback, methods=('GET',)):
        callback.__name__ = str(uuid4())  # change name of method to mitigate flask exception
        self.app.add_url_rule(url, view_func=callback, methods=methods)

    def add_json_response(self, url, serializable, methods=('GET',)):
        def callback():
            return jsonify(serializable)
        
        self.add_callback_response(url, callback, methods=methods)

    def run(self):
        self.app.run(port=self.port)

requirements.txt

flask
requests

tests.py

import unittest
import requests

from mockserver import MockServer


class TestMockServer(unittest.TestCase):
    def setUp(self):
        self.server = MockServer(port=1234)
        self.server.start()

    def test_mock_with_json_serializable(self):
        self.server.add_json_response("/json", dict(hello="welt"))

        response = requests.get(self.server.url + "/json")

        self.assertEqual(200, response.status_code)
        self.assertIn('hello', response.json())
        self.assertEqual('welt', response.json()['hello'])

    def test_mock_with_callback(self):
        self.called = False

        def callback():
            self.called = True
            return 'Hallo'

        self.server.add_callback_response("/callback", callback)

        response = requests.get(self.server.url + "/callback")

        self.assertEqual(200, response.status_code)
        self.assertEqual('Hallo', response.text)
        self.assertTrue(self.called)

    def tearDown(self):
        self.server.shutdown_server()


if __name__ == '__main__':
    unittest.main()

@eruvanos, this is great! Thanks for sharing it! I am using this as a starting point for my mock server.

I added an /alive endpoint and a liveliness check to make sure that the mock server initializes before we use it in any tests. My code has diverged from this gist, but here is the relevant section:

        server_is_alive = False
        liveliness_attempts = 0
        while not server_is_alive:
            if liveliness_attempts >= 50:
                raise Exception('Failed to start and connect to mock server. '
                                f'Is port {self.port} in use by another application?')
            liveliness_attempts += 1
            try:
                requests.get(self.url + '/alive', timeout=0.2)
                server_is_alive = True
            except (requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout):
                time.sleep(0.1)

I also noticed that request.environ['werkzeug.server.shutdown'] has been deprecated. I updated my mock server to not use it:

class MockServer:
    def __init__(self, port=5050):
        self.port = port
        self.app = Flask(__name__)
        self.server = make_server('localhost', self.port, self.app)
        self.url = "http://localhost:%s" % self.port
        self.thread = None

        @self.app.route('/alive', methods=['GET'])
        def alive():
            return "True"

    def start(self):
        self.thread = Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()

        # Ensure server is alive before we continue running tests
        server_is_alive = False
        liveliness_attempts = 0
        while not server_is_alive:
            if liveliness_attempts >= 50:
                raise Exception('Failed to start and connect to mock server. '
                                f'Is port {self.port} in use by another application?')
            liveliness_attempts += 1
            try:
                requests.get(self.url + '/alive', timeout=0.2)
                server_is_alive = True
            except (requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout):
                time.sleep(0.1)

    def stop(self):
        self.server.shutdown()
        self.thread.join()

My fixture for it looks like this:

@pytest.fixture(scope='session')
def mock_server():
    server = MockServer()
    server.start()
    yield server
    server.stop()

Instead of using add_json_response or add_callback_response, I am using flask blueprints. Here is an example (which assumes you have created a blueprint called image_blueprint.

@pytest.fixture(scope='session')
def image_server_url(mock_server):
    mock_server.app.register_blueprint(image_blueprint, url_prefix='/images')
    return mock_server.url + '/images'