-
-
Save cknv/152d81ffcc4a74be491d to your computer and use it in GitHub Desktop.
pytest-flask streaming
This file contains 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
from flask import Flask | |
import streamer | |
def create_app(): | |
app = Flask(__name__) | |
app.register_blueprint(streamer.mod) | |
return app |
This file contains 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
from app import create_app | |
import pytest | |
@pytest.fixture(scope='session') | |
def app(): | |
return create_app() | |
@pytest.yield_fixture | |
def client(app): | |
with app.test_client() as client: | |
yield client |
This file contains 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
import json | |
from flask import Blueprint | |
from flask import Response | |
from flask import stream_with_context | |
mod = Blueprint('streamer', __name__) | |
@mod.route('/stream') | |
def stream(): | |
gimme_data = generate_data(10) | |
json_data = ( | |
json.dumps(each) + '\n' | |
for each in gimme_data | |
) | |
return Response( | |
stream_with_context(json_data), | |
mimetype='application/x-json-stream', | |
) | |
def generate_data(length): | |
for i in range(length): | |
yield { | |
'id': i, | |
} |
This file contains 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
import json | |
from flask import url_for | |
def test_something(client): | |
resp = client.get( | |
url_for('streamer.stream'), | |
) | |
assert resp.status_code == 200 | |
data = [ | |
json.loads(each) | |
for each in resp.data.decode().split('\n') | |
if each | |
] | |
assert len(data) == 10 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment