""" pytest support functions. """ import os import threading import tempfile import urllib.request import urllib.parse from urllib.error import URLError import pytest import flask import werkzeug import sqlalchemy from flask_sqlalchemy import SQLAlchemy import alembic import alembic.config import asl_articles from asl_articles import app from asl_articles.utils import to_bool from asl_articles import tests as asl_articles_tests _FLASK_SERVER_URL = ( "localhost", 5001 ) # nb: for the test Flask server we spin up _pytest_options = None # --------------------------------------------------------------------- def pytest_addoption( parser ): """Configure pytest options.""" # NOTE: This file needs to be in the project root for this to work :-/ # https://docs.pytest.org/en/latest/reference.html#initialization-hooks # add test options parser.addoption( # NOTE: We assume that the React frontend server is already running. "--web-url", action="store", dest="web_url", default="http://localhost:3000", help="React server to test against." ) parser.addoption( "--flask-url", action="store", dest="flask_url", default=None, help="Flask server to test against." ) parser.addoption( "--webdriver", action="store", dest="webdriver", default="chrome", help="Webdriver to use (chrome|firefox)." ) # add test options parser.addoption( "--headless", action="store_true", dest="headless", default=False, help="Run the tests headless." ) parser.addoption( "--window", action="store", dest="window_size", default="1000x700", help="Browser window size." ) # add test options parser.addoption( "--dbconn", action="store", dest="dbconn", default=None, help="Database connection string." ) # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - def pytest_configure( config ): """Called after command-line options have been parsed.""" global _pytest_options _pytest_options = config.option # notify the test suite about the pytest options asl_articles_tests.pytest_options = _pytest_options # --------------------------------------------------------------------- @pytest.fixture( scope="session" ) def flask_app( request ): """Prepare the Flask server.""" # initialize flask_url = request.config.getoption( "--flask-url" ) #pylint: disable=no-member # initialize # WTF?! https://github.com/pallets/flask/issues/824 def make_flask_url( endpoint, **kwargs ): """Generate a URL for the Flask backend server.""" with app.test_request_context(): url = flask.url_for( endpoint, _external=True, **kwargs ) if flask_url: url = url.replace( "http://localhost", flask_url ) else: url = url.replace( "localhost/", "{}:{}/".format(*_FLASK_SERVER_URL) ) return url app.url_for = make_flask_url # check if we need to start a local Flask server if not flask_url: # FUDGE! If we're going to create our own Flask server, we want to stop it from checking # the *configured* database connection string (since it will fail to start if there's a problem). asl_articles._disable_db_startup = True #pylint: disable=protected-access # yup - make it so server = werkzeug.serving.make_server( _FLASK_SERVER_URL[0], _FLASK_SERVER_URL[1], app, threaded=True ) thread = threading.Thread( target = server.serve_forever, daemon=True ) thread.start() # wait for the server to start up def is_ready(): """Try to connect to the Flask server.""" try: url = app.url_for( "ping" ) with urllib.request.urlopen( url ) as resp: assert resp.read() == b"pong" return True except URLError: return False except Exception as ex: #pylint: disable=broad-except assert False, "Unexpected exception: {}".format( ex ) asl_articles_tests.utils.wait_for( 5, is_ready ) # return the server to the caller try: yield app finally: # shutdown the local Flask server if not flask_url: server.shutdown() thread.join() # --------------------------------------------------------------------- @pytest.fixture( scope="session" ) def webdriver( request ): """Prepare a webdriver that can be used to control a browser.""" # initialize driver = request.config.getoption( "--webdriver" ) headless = request.config.getoption( "--headless" ) from selenium import webdriver as wb #pylint: disable=import-outside-toplevel if driver == "firefox": options = wb.FirefoxOptions() if headless: options.add_argument( "--headless" ) #pylint: disable=no-member driver = wb.Firefox( options=options ) elif driver == "chrome": options = wb.ChromeOptions() if headless: options.add_argument( "--headless" ) #pylint: disable=no-member options.add_argument( "--disable-gpu" ) driver = wb.Chrome( options=options ) else: raise RuntimeError( "Unknown webdriver: {}".format( driver ) ) # set the browser size window_size = request.config.getoption( "--window" ) if window_size: words = window_size.split( "x" ) #pylint: disable=no-member driver.set_window_size( int(words[0]), int(words[1]) ) # figure out which Flask backend server the React frontend should talk to flask_url = request.config.getoption( "--flask-url" ) if not flask_url: # we're talking to our own test Flask server flask_url = "http://{}:{}".format( *_FLASK_SERVER_URL ) # initialize web_url = request.config.getoption( "--web-url" ) assert web_url def make_web_url( url, **kwargs ): """Generate a URL for the React frontend.""" url = "{}/{}".format( web_url, url ) kwargs[ "_flask"] = flask_url kwargs[ "store_msgs"] = 1 # stop notification messages from building up and obscuring clicks kwargs[ "fake_uploads"] = 1 # alternate mechanism for uploading files url += "&" if "?" in url else "?" url += urllib.parse.urlencode( kwargs ) return url driver.make_url = make_web_url # return the webdriver to the caller try: yield driver finally: driver.quit() # --------------------------------------------------------------------- @pytest.fixture( scope="function" ) def dbconn( request ): """Prepare a database connection.""" # initialize conn_string = request.config.getoption( "--dbconn" ) temp_fname = None if conn_string: if os.path.isfile( conn_string ): # a file was specified - we assume it's an SQLite database conn_string = "sqlite:///{}".format( conn_string ) else: # create a temp file and install our database schema into it with tempfile.NamedTemporaryFile( delete=False ) as temp_file: temp_fname = temp_file.name dname = os.path.join( os.path.split(__file__)[0], "alembic/" ) cfg = alembic.config.Config( os.path.join( dname, "alembic.ini" ) ) cfg.set_main_option( "script_location", dname ) conn_string = "sqlite:///{}".format( temp_fname ) cfg.set_main_option( "sqlalchemy.url", conn_string ) alembic.command.upgrade( cfg, "head" ) # connect to the database engine = sqlalchemy.create_engine( conn_string, echo = to_bool( app.config.get( "SQLALCHEMY_ECHO" ) ) ) # IMPORTANT! The test suite often loads the database with test data, and then runs searches to see what happens. # In the normal case, this works fine: # - we use either an existing database, or create a temp file as an sqlite database (see above) # - this database is then installed into the temp Flask server that the "flask_app" fixture spun up, thus ensuring # that both the backend server and test code are working with the same database. # However, this doesn't work when the test suite is talking to a remote Flask server (via the --flask-url argument), # since it has no way of ensuring that the remote Flask server is talking to the same database. In this case, # it's the developer's responsibility to make sure that this is the case (by configuring the database in site.cfg). prev_db_state = None try: flask_url = request.config.getoption( "--flask-url" ) #pylint: disable=no-member if flask_url: # we are talking to a remote Flask server, we assume it's already talking to the database we are pass else: # remember the database our temp Flask server is currently using prev_db_state = ( app.config["SQLALCHEMY_DATABASE_URI"], asl_articles.db ) # replace the database the temp Flask server with ours app.config[ "SQLALCHEMY_DATABASE_URI" ] = conn_string app.db = SQLAlchemy( app ) # return the database connection to the caller yield engine finally: # restore the original database into our temp Flask server if prev_db_state: app.config[ "SQLALCHEMY_DATABASE_URI" ] = prev_db_state[0] asl_articles.db = prev_db_state[1] # clean up if temp_fname: os.unlink( temp_fname )