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()

Nice little snippet. Thanks for sharing.

For anyone interested, you can also wrap this up into a little fixture for use with pytest:

@pytest.fixture
def mock_server(request):
    server = MockServer()
    server.start()
    yield server
    server.shutdown_server()

def test_something(mock_server):
    api_path = '/api/something/'
    mock_server.add_json_response(api_path, {'something': 'special'})

    url = urljoin(mock_server.url, api_path)

    # Test something ...

Broadening the fixture scope can keep it alive for longer usage as well if you want to run it across a few different tests.

Thanks for sharing this snippet!
@tkh Thanks for sharing fixture for pytest!

Besides, There is some problem for using mock_server for pytest session scope.
when if using multiple add_callback_response in the same session, an error occurred like that

AssertionError: View function mapping is overwriting an existing endpoint function: callback

when using pytest fixture like this

# conftest.py
import pytest

from test.mockserver import Mockserver

@pytest.fixture(scope='session')
def mock_server(request):
    server = Mockserver(port=3000)
    server.start()
    yield server
    server.shutdown_server()

and multiple add_callback_response in test

# test_something.py
def test_get_post(mock_server):
    api_get_path = 'api/get'
    api_post_path = 'api/post'

    mock_server.add_json_response(api_get_path, {'something': 'special'}, methods=('GET', ))
    mock_server.add_json_response(api_post_path, {'something': 'special'}, methods=('POST', ))

    # and some codes ...

Because that add_json_response method use same method name 'callback', an error occurred,

AssertionError: View function mapping is overwriting an existing endpoint function: callback

So fix add_callback_response, add_json_response methods like these,

def add_callback_response(self, url, endpoint, callback, methods=('GET', )):
    self.app.add_url_rule(url, endpoint=endpoint, view_func=callback, methods=methods)

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

and use add_json_response like,

mock_server.add_json_response(api_get_path, api_get_path, {'something': 'special'}, methods=('GET', ))
mock_server.add_json_response(api_post_path, api_post_path, {'something': 'special'}, methods=('POST', ))

it works!

That looks super useful!
What licence are the snippets on?

You can also use https://pypi.org/project/http-server-mock/

from http_server_mock import HttpServerMock
import requests
app = HttpServerMock(__name__)

@app.route("/", methods=["GET"])
def index():
    return "Hello world"

with app.run("localhost", 5000):
    r = requests.get("http://localhost:5000/")
    # r.status_code == 200
    # r.text == "Hello world"

@RafalSkolasinski MIT

What is the reason behind importing flask only in the __init__ and not at the top of the mockserver module?

What is the reason behind importing flask only in the __init__ and not at the top of the mockserver module?

Changed it, there was no reason.

any chance you know how to disable logging ?

@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'