Last active
October 19, 2016 20:06
-
-
Save dillonhicks/4f772a15c487b7354a72bed2d6488c13 to your computer and use it in GitHub Desktop.
Flask Live Server Fixture Running on Ephemeral Port
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # -*- coding: utf-8 -*- | |
| from __future__ import absolute_import | |
| import socket | |
| import multiprocessing | |
| import time | |
| import flask | |
| import pytest | |
| class FlaskTestAppContextMixin(object): | |
| @pytest.fixture | |
| def app(self): | |
| app = flask.Flask(type(self).__name__) | |
| app.config['TESTING'] = True | |
| return app | |
| @pytest.fixture | |
| def ephemeral_port(self): | |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
| sock.bind(('localhost', 0)) | |
| port = sock.getsockname()[1] | |
| sock.close() | |
| return port | |
| @pytest.fixture | |
| def server_running(self, app, ephemeral_port): | |
| def worker(app, port): | |
| app.run(port=port, use_reloader=False) | |
| server_process = multiprocessing.Process(target=worker, args=(app, ephemeral_port)) | |
| server_process.start() | |
| # We must wait for the server to start listening, but give up | |
| # after a specified maximum timeout | |
| timeout = 5 | |
| start_time = time.time() | |
| while True: | |
| elapsed_time = (time.time() - start_time) | |
| if elapsed_time > timeout: | |
| raise RuntimeError("Failed to start the server after %d seconds. " % timeout) | |
| if self._can_ping_server(ephemeral_port): | |
| break | |
| try: | |
| with app.test_request_context(): | |
| yield | |
| finally: | |
| server_process.terminate() | |
| def _can_ping_server(self, port): | |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
| try: | |
| sock.connect(('localhost', port)) | |
| except socket.error as e: | |
| success = False | |
| else: | |
| success = True | |
| finally: | |
| sock.close() | |
| return success | |
| class TestMyService(FlaskTestAppContextMixin): | |
| @pytest.fixture | |
| def my_service_blueprint(self, app): | |
| blueprint = # Init blueprint here | |
| app.register_blueprint(blueprint) | |
| return zoo_views | |
| def test_get_my_route(self, my_service_blueprint, running_server, app): | |
| with app.test_client() as client: | |
| rv = client.get('/my/route') | |
| assert rv.status_code == 200 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment