whoops, pytest-flask already has a live-server, thanks pytest-flask

This commit is contained in:
dandds 2018-10-02 13:58:09 -04:00
parent 5195b2d32d
commit 586a1eee5d
3 changed files with 9 additions and 125 deletions

View File

@ -7,7 +7,6 @@ from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
from .live_server import LiveServer
from .browsers import BROWSERSTACK_CONFIG
@ -20,18 +19,12 @@ def session(db, request):
@pytest.fixture(scope="session")
def live_app(app):
def app(app):
handler = RotatingFileHandler("log/acceptance.log", maxBytes=10000, backupCount=1)
handler.setLevel(logging.INFO)
app.logger.addHandler(handler)
runnable = LiveServer(app, port=8943, timeout=10)
runnable.spawn_live_server()
app.server_url = runnable.server_url
yield app
runnable.terminate()
return app
class DriverCollection(Mapping):

View File

@ -1,109 +0,0 @@
import gc
import multiprocessing
import socket
import socketserver
import time
from urllib.parse import urlparse, urljoin
# This is adapted from flask-testing, https://github.com/jarus/flask-testing
# Inspired by https://docs.djangoproject.com/en/dev/topics/testing/#django.test.LiveServerTestCase
class LiveServer:
def __init__(self, app, port=5000, timeout=5):
self.app = app
self._configured_port = port
self._timeout = timeout
self._port_value = multiprocessing.Value("i", self._configured_port)
@property
def server_url(self):
return "http://localhost:%s" % self._port_value.value
def spawn_live_server(self):
self._process = None
port_value = self._port_value
def worker(app, port):
# Based on solution: http://stackoverflow.com/a/27598916
# Monkey-patch the server_bind so we can determine the port bound
# by Flask. This handles the case where the port specified is `0`,
# which means that the OS chooses the port. This is the only known
# way (currently) of getting the port out of Flask once we call
# `run`.
original_socket_bind = socketserver.TCPServer.server_bind
def socket_bind_wrapper(self):
ret = original_socket_bind(self)
# Get the port and save it into the port_value, so the parent
# process can read it.
(_, port) = self.socket.getsockname()
port_value.value = port
socketserver.TCPServer.server_bind = original_socket_bind
return ret
socketserver.TCPServer.server_bind = socket_bind_wrapper
app.run(port=port, use_reloader=False)
self._process = multiprocessing.Process(
target=worker, args=(self.app, self._configured_port)
)
self._process.start()
# We must wait for the server to start listening, but give up
# after a specified maximum timeout
start_time = time.time()
while True:
elapsed_time = time.time() - start_time
if elapsed_time > self._timeout:
raise RuntimeError(
"Failed to start the server after %d seconds. " % self._timeout
)
if self._can_ping_server():
break
def _can_ping_server(self):
host, port = self.address
if port == 0:
# Port specified by the user was 0, and the OS has not yet assigned
# the proper port.
return False
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sock.connect((host, port))
except socket.error as e:
success = False
else:
success = True
finally:
sock.close()
return success
@property
def address(self):
"""
Gets the server address used to test the connection with a socket.
Respects both the LIVESERVER_PORT config value and overriding server_url
"""
parts = urlparse(self.server_url)
host = parts.hostname
port = parts.port
if port is None:
if parts.scheme == "http":
port = 80
elif parts.scheme == "https":
port = 443
else:
raise RuntimeError("Unsupported server url scheme: %s" % parts.scheme)
return host, port
def terminate(self):
if self._process:
self._process.terminate()

View File

@ -15,9 +15,10 @@ USER_CERT = "ssl/client-certs/atat.mil.crt"
@pytest.mark.parametrize("browser_type", BROWSERSTACK_CONFIG.keys())
def test_can_get_title(browser_type, live_app, drivers):
@pytest.mark.usefixtures('live_server')
def test_can_get_title(browser_type, app, drivers):
driver = drivers[browser_type]
driver.get(live_app.server_url)
driver.get(url_for("atst.root", _external=True))
assert "JEDI" in driver.title
@ -57,13 +58,12 @@ def _valid_login(client, driver):
)
def test_login(live_app, drivers, client, valid_user_from_cert):
@pytest.mark.usefixtures('live_server')
def test_login(drivers, client, app, valid_user_from_cert):
driver = drivers["win10_chrome62"]
driver.get(live_app.server_url)
driver.get(url_for("atst.root", _external=True))
cookie = _valid_login(client, driver)
requests_page = urljoin(
live_app.server_url, url_for("requests.requests_form_new", screen=1)
)
requests_page = url_for("requests.requests_form_new", screen=1, _external=True)
driver.get(requests_page)
user = valid_user_from_cert
assert user.last_name in driver.page_source