Skip to content

Commit

Permalink
Refactored flask live instance fixture
Browse files Browse the repository at this point in the history
  • Loading branch information
@DiogenesAnalytics
DiogenesAnalytics committed Mar 15, 2024
1 parent 1e385c4 commit 7b25861
Show file tree
Hide file tree
Showing 6 changed files with 453 additions and 281 deletions.
310 changes: 56 additions & 254 deletions tests/conftest.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,25 +3,22 @@
import base64
import json
import os
import random
import shutil
import threading
from pathlib import Path
from typing import Any
from typing import Dict
from typing import Generator
from typing import Set
from typing import Tuple

import pytest
from flask import Flask
from flask import render_template
from flask import request
from flask import send_from_directory
from PIL import Image
from selenium.webdriver.common.keys import Keys
from werkzeug.datastructures import FileStorage
from werkzeug.datastructures import ImmutableMultiDict

from tests.server import TEST_SERVER_INFO
from tests.server import build_flask_app
from tests.server import run_threaded_flask_app


def pytest_configure(config):
Expand All @@ -34,23 +31,24 @@ def pytest_configure(config):
config.addinivalue_line("markers", "website: custom marker for website tests.")


def generate_unique_random_ports(num_ports: int) -> Generator[int, None, None]:
"""Generator that only yield unique random ports."""
# create set of used ports
used_ports: Set[int] = set()
def get_server_info() -> Tuple[int, str]:
"""Convenience function to get test server port and submit route."""
return TEST_SERVER_INFO["port"], TEST_SERVER_INFO["submit_route"]

# loop over ports
while len(used_ports) < num_ports:
# get random port
port = random.randint(5001, 65535)

# check it is unique
if port not in used_ports:
# send it forward
yield port
def base_custom_config() -> Dict[str, Any]:
"""Defines the basic JSON config file attributes."""
# get port and submit route
port, submit = get_server_info()

# mark it as used
used_ports.add(port)
# build base config
return {
"subject": "Testing",
"title": "Testing",
"form_backend_url": f"http://localhost:{port}{submit}",
"email": "[email protected]",
"questions": [],
}


def create_temp_websrc_dir(src: Path, dst: Path, src_files: Tuple[str, ...]) -> Path:
Expand All @@ -76,151 +74,6 @@ def create_temp_websrc_dir(src: Path, dst: Path, src_files: Tuple[str, ...]) ->
return sub_dir


def get_html_tag_from_mimetype(file: FileStorage, encoded_data: str) -> str:
"""Generate an HTML tag based on the MIME type of the file."""
# create data URL for reuse below
data_url = f"data:{file.mimetype};base64,{encoded_data}"

# match the mimetype
match file.mimetype.split("/")[0]:
case "image":
tag = f"<img src={data_url!r}>"
case "video":
tag = (
f"<video controls>"
f" <source src={data_url!r} type={file.mimetype!r}>"
f" Your browser does not support the video tag."
f"</video>"
)
case "audio":
tag = (
f"<audio controls>"
f" <source src={data_url!r} type={file.mimetype!r}>"
f" Your browser does not support the audio tag."
f"</audio>"
)
case _:
tag = f"<a href={data_url!r}>Download {file.filename}</a>"

return tag


def process_form_data(form_data: ImmutableMultiDict) -> Dict[str, Any]:
"""Process form data to handle multi-values."""
# setup processed results
processed_data: Dict[str, Any] = {}

# check form key/values
for key, value in form_data.items(multi=True):
# check if key indicates file(s)
if key in request.files:
processed_data[key] = ""

# check to see if there are multiple values
elif key in processed_data:
processed_data[key] += f", {value}"

# handle normally
else:
processed_data[key] = value

return processed_data


def process_uploaded_files(processed_data: Dict[str, Any]) -> None:
"""Process uploaded files and generate HTML tags."""
# get list of tuples for key/files pairs
for key, files in request.files.lists():
# loop over each file
for file in files:
# make sure it exists
if file.filename:
# get data from file
file_data = file.read()

# convert to base64 for data URL creation later ...
encoded_data = base64.b64encode(file_data).decode("utf-8")

# create tag
tag = get_html_tag_from_mimetype(file, encoded_data)

# update current results
if key in processed_data:
processed_data[key] += "<br>" + tag
else:
processed_data[key] = tag


def build_flask_app(serve_directory: Path, port: int, submit_route: str) -> Flask:
"""Assembles Flask app to serve static site."""
# instantiate app
app = Flask(__name__)

# update the port
app.config["PORT"] = port

# define routes
@app.route("/")
def index():
"""Serve the index file in the project dir."""
return send_from_directory(serve_directory, "index.html")

@app.route("/<path:path>")
def other_root_files(path):
"""Serve any other files (e.g. config.json) from the project dir."""
return send_from_directory(serve_directory, path)

@app.route("/styles/<path:path>")
def serve_styles(path):
"""Send any CSS files from the temp dir."""
css_file = os.path.join("styles", path)
if os.path.exists(os.path.join(serve_directory, css_file)):
return send_from_directory(serve_directory, css_file)
else:
return "CSS file not found\n", 404

@app.route("/scripts/<path:path>")
def serve_scripts(path):
"""Send any JavaScript files from the temp dir."""
js_file = os.path.join("scripts", path)
if os.path.exists(os.path.join(serve_directory, js_file)):
return send_from_directory(serve_directory, js_file)
else:
return "JavaScript file not found\n", 404

@app.route(submit_route, methods=["POST"])
def submit_form():
"""Render HTML form data as a response form."""
# log data
print(f"Form data received: {request.form}")

# process data
processed_data = process_form_data(request.form)

# log processed data
print(f"Processed data: {processed_data}")

# process any files
process_uploaded_files(processed_data)

# log files added
print(f"Added uploaded files: {processed_data}")

# now render response
return render_template("form_response_template.html", form_data=processed_data)

# return configured and route decorated Flask app
return app


def run_threaded_flask_app(app: Flask, port: int) -> None:
"""Run a Flask app using threading."""
# launch Flask app for project dir in thread
thread = threading.Thread(target=app.run, kwargs={"port": port})
thread.daemon = True
thread.start()


def load_config_file(directory: Path) -> Dict[str, Any]:
"""Load the JSON config file at directory."""
# open the config file in the project dir
Expand All @@ -236,10 +89,13 @@ def write_config_file(config: Dict[str, Any], src_path: Path) -> None:
json.dump(config, json_file, indent=2)


def prepare_default_config(config: Dict[str, Any], src_path: Path, port: int) -> None:
def prepare_default_config(config: Dict[str, Any], src_path: Path) -> None:
"""Update the default config copy with values approprate for testing."""
# get port and submit route
port, submit = get_server_info()

# update form backend
config["form_backend_url"] = f"http://localhost:{port}/submit"
config["form_backend_url"] = f"http://localhost:{port}{submit}"

# update input[type=file] accept attr
for question in config["questions"]:
Expand Down Expand Up @@ -404,46 +260,22 @@ def session_websrc_tmp_dir(
shutil.rmtree(temp_dir)


@pytest.fixture(scope="function")
def function_websrc_tmp_dir(
project_directory: Path, tmp_path: Path, website_files: Tuple[str, ...]
) -> Generator[Path, None, None]:
"""Create a per-function copy of the website source code for editing."""
# create a temporary directory
temp_dir = create_temp_websrc_dir(project_directory, tmp_path, website_files)

# provide the temporary directory path to the test function
yield temp_dir

# remove the temporary directory and its contents
shutil.rmtree(temp_dir)


@pytest.fixture(scope="session")
def default_site_config(project_directory: Path) -> Dict[str, Any]:
"""Load the default config.json file."""
return load_config_file(project_directory)


@pytest.fixture(scope="session")
def submit_route() -> str:
"""Defines the route used for the form submission testing."""
return "/submit"


@pytest.fixture(scope="session")
def session_web_app(
default_site_config: Dict[str, Any], session_websrc_tmp_dir: Path, submit_route: str
default_site_config: Dict[str, Any], session_websrc_tmp_dir: Path
) -> Flask:
"""Create a session-scoped Flask app for testing with the website source."""
# set port
port = 5000

# now update config.json with new backend url
prepare_default_config(default_site_config, session_websrc_tmp_dir, port)
prepare_default_config(default_site_config, session_websrc_tmp_dir)

# create app
return build_flask_app(session_websrc_tmp_dir, port, submit_route)
return build_flask_app(session_websrc_tmp_dir)


@pytest.fixture(scope="session")
Expand All @@ -454,38 +286,7 @@ def live_session_web_app_url(session_web_app: Flask) -> str:
assert port is not None

# start threaded app
run_threaded_flask_app(session_web_app, port)

# get url
return f"http://localhost:{port}"


@pytest.fixture(scope="function")
def unique_random_port() -> int:
"""Generate a unique random port greater than 5000."""
# control the number of ports generated
num_ports = 100
return next(generate_unique_random_ports(num_ports))


@pytest.fixture(scope="function")
def function_web_app(
function_websrc_tmp_dir: Path, submit_route: str, unique_random_port: int
) -> Flask:
"""Create a function-scoped Flask app for testing with the website source."""
# create app
return build_flask_app(function_websrc_tmp_dir, unique_random_port, submit_route)


@pytest.fixture(scope="function")
def live_function_web_app_url(function_web_app: Flask) -> str:
"""Runs session-scoped Flask app in a thread."""
# get port
port = function_web_app.config.get("PORT")
assert port is not None

# start threaded app
run_threaded_flask_app(function_web_app, port)
run_threaded_flask_app(session_web_app)

# get url
return f"http://localhost:{port}"
Expand All @@ -494,30 +295,31 @@ def live_function_web_app_url(function_web_app: Flask) -> str:
@pytest.fixture(scope="function")
def multiple_select_options_config() -> Dict[str, Any]:
"""Custom config file fixture for testing multiple select options."""
return {
"subject": "Testing Multiple Select Options",
"title": "Testing Multi-Select Options",
"form_backend_url": None,
"email": "[email protected]",
"questions": [
{
"label": "Select your country",
"name": "country",
"type": "selectbox",
"required": True,
"options": [
{
"label": "--Select all that apply--",
"value": "",
"selected": True,
"disabled": True,
},
{"label": "USA", "value": "USA"},
{"label": "Canada", "value": "CAN"},
{"label": "United Kingdom", "value": "UK"},
{"label": "Australia", "value": "AUS"},
],
"custom": {"multiple": True},
}
],
}
# get base config
config = base_custom_config()

# update questions
config["questions"] = [
{
"label": "Select your country",
"name": "country",
"type": "selectbox",
"required": True,
"options": [
{
"label": "--Select all that apply--",
"value": "",
"selected": True,
"disabled": True,
},
{"label": "USA", "value": "USA"},
{"label": "Canada", "value": "CAN"},
{"label": "United Kingdom", "value": "UK"},
{"label": "Australia", "value": "AUS"},
],
"custom": {"multiple": True},
}
]

# updated
return config
Loading

0 comments on commit 7b25861

Please sign in to comment.