diff --git a/tests/conftest.py b/tests/conftest.py index d1b39ae..9e21071 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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): @@ -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": "foo@bar.com", + "questions": [], + } def create_temp_websrc_dir(src: Path, dst: Path, src_files: Tuple[str, ...]) -> Path: @@ -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"" - case "video": - tag = ( - f"" - ) - case "audio": - tag = ( - f"" - ) - case _: - tag = f"Download {file.filename}" - - 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] += "
" + 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("/") - 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/") - 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/") - 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 @@ -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"]: @@ -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") @@ -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}" @@ -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": "foo@bar.com", - "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 diff --git a/tests/data_structures.py b/tests/data_structures.py new file mode 100644 index 0000000..ee2b4fd --- /dev/null +++ b/tests/data_structures.py @@ -0,0 +1,17 @@ +"""Custom classes used in testing.""" + + +class ImmutableDict(dict): + """Implements an immutable dictionary for safely using dictionary fixtures.""" + + def __init__(self, *args, **kwargs): + """Pass through to parent constructor nothing fancy here.""" + super().__init__(*args, **kwargs) + + def __setitem__(self, key, value): + """Remove item assignment.""" + raise TypeError("ImmutableDict does not support item assignment") + + def __delitem__(self, key): + """Remove item deletion.""" + raise TypeError("ImmutableDict does not support item deletion") diff --git a/tests/requirements.txt b/tests/requirements.txt index 1bcade8..68d463c 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -68,6 +68,7 @@ soupsieve==2.5 tabcompleter==1.3.0 trio==0.24.0 trio-websocket==0.11.1 +types-requests==2.31.0.20240311 typing_extensions==4.10.0 urllib3==2.2.1 Werkzeug==3.0.1 diff --git a/tests/server.py b/tests/server.py new file mode 100644 index 0000000..df38e92 --- /dev/null +++ b/tests/server.py @@ -0,0 +1,270 @@ +"""Defines functions related to the custom Flask testing server.""" + +import base64 +import os +import random +import secrets +import threading +from pathlib import Path +from typing import Any +from typing import Dict +from typing import Generator +from typing import Set + +from flask import Blueprint +from flask import Flask +from flask import jsonify +from flask import render_template +from flask import request +from flask import send_from_directory +from flask import session +from werkzeug.datastructures import FileStorage +from werkzeug.datastructures import ImmutableMultiDict + +from tests.data_structures import ImmutableDict + + +TEST_SERVER_INFO = ImmutableDict( + { + "port": 5000, + "secret_key": secrets.token_hex(16), + "submit_route": "/submit", + } +) + +CONFIG_DATA_MAP: Dict[str, Any] = {} + + +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() + + # 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 + + # mark it as used + used_ports.add(port) + + +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"" + case "video": + tag = ( + f"" + ) + case "audio": + tag = ( + f"" + ) + case _: + tag = f"Download {file.filename}" + + 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] += "
" + tag + else: + processed_data[key] = tag + + +def create_main_blueprint( + serve_directory: Path, config_data_map: Dict[str, Any] +) -> Blueprint: + """Builds a Flask Blueprint for all main routes.""" + main_bp = Blueprint("main", __name__) + + @main_bp.route("/") + def index(): + """Serve the index file in the project dir.""" + # get token from query parameters + token = request.args.get("token") + + # check if token exists + if token: + # get config data + config_data = config_data_map.get(token) + + # update session data + session["config_data"] = config_data + + # notify + print(f"Received token: {token}") + print(f"Website will be configured using: {config_data}") + + return send_from_directory(serve_directory, "index.html") + + @main_bp.route("/") + def other_root_files(path): + """Serve any other files (e.g. config.json) from the project dir.""" + if "config.json" in path and (config_data := session.get("config_data")): + print(f"Serving updated config.json data: {config_data}") + return jsonify(config_data) + else: + return send_from_directory(serve_directory, path) + + @main_bp.route("/styles/") + 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 + + @main_bp.route("/scripts/") + 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 + + return main_bp + + +def create_config_blueprint(config_data_map: Dict[str, Any]) -> Blueprint: + """Builds a Flask Blueprint for all config updating routes.""" + config_bp = Blueprint("config", __name__) + + @config_bp.route("/update_config", methods=["POST"]) + def update_config(): + """Update session with new JSON data.""" + if request.is_json: + config_data = request.json + token = secrets.token_urlsafe(16) + session["config_data"] = config_data + config_data_map[token] = config_data + print(f"Updating config data: {config_data}") + return jsonify({"token": token}), 200 + else: + return "Invalid request format. Only JSON requests are accepted.\n", 400 + + @config_bp.route("/reset_config") + def reset_config(): + """Clears the session cache of any config data.""" + session.pop("config_data", None) + return "Configuration reset successfully!\n", 200 + + return config_bp + + +def create_submit_blueprint() -> Blueprint: + """Builds a Flask Blueprint for all form submission routes.""" + submit_bp = Blueprint("submit", __name__) + + @submit_bp.route(TEST_SERVER_INFO["submit_route"], methods=["POST"]) + def submit_form(): + """Render HTML form data as a response form.""" + # notify what form data was received + print(f"Form data received: {request.form}") + + # notify what data was processed + processed_data = process_form_data(request.form) + print(f"Processed data: {processed_data}") + + # notify what files were added (if any) + process_uploaded_files(processed_data) + print(f"Added uploaded files: {request.files}") + + # render the contact form response + return render_template("form_response_template.html", form_data=processed_data) + + return submit_bp + + +def build_flask_app(serve_directory: Path) -> Flask: + """Assembles Flask app to serve static site.""" + # get instance + app = Flask(__name__) + + # set port + app.config["PORT"] = TEST_SERVER_INFO["port"] + + # set secret key + app.config["SECRET_KEY"] = TEST_SERVER_INFO["secret_key"] + + # set up config data map + config_data_map = CONFIG_DATA_MAP + + # build blueprints + main_bp = create_main_blueprint(serve_directory, config_data_map) + config_bp = create_config_blueprint(config_data_map) + submit_bp = create_submit_blueprint() + + # add blueprints to Flask app + app.register_blueprint(main_bp) + app.register_blueprint(config_bp) + app.register_blueprint(submit_bp) + + return app + + +def run_threaded_flask_app(app: Flask) -> None: + """Run a Flask app using threading.""" + # launch Flask app for project dir in thread + thread = threading.Thread(target=app.run) + thread.daemon = True + thread.start() diff --git a/tests/test_fixtures.py b/tests/test_fixtures.py index 595a326..91a1a9b 100644 --- a/tests/test_fixtures.py +++ b/tests/test_fixtures.py @@ -118,7 +118,6 @@ def test_serve_scripts_route(session_web_app: Flask) -> None: @pytest.mark.fixture def test_submit_form_route( session_web_app: Flask, - submit_route: str, dummy_form_post_data: Dict[str, Any], dummy_txt_file_data_url: str, ) -> None: @@ -128,7 +127,7 @@ def test_submit_form_route( # submit response response = client.post( - submit_route, data=dummy_form_post_data, content_type="multipart/form-data" + "/submit", data=dummy_form_post_data, content_type="multipart/form-data" ) # assert that the response status code is 200 (OK) @@ -180,6 +179,81 @@ def test_submit_form_route( ), "Form data in HTML response does not match expected form data" +@pytest.mark.flask +@pytest.mark.fixture +def test_update_config_route(session_web_app: Flask) -> None: + """Test the route for updating the configuration.""" + client = session_web_app.test_client() + + # send a POST request with JSON data to update the configuration + new_config = {"key": "value"} + post_response = client.post("/update_config", json=new_config) + + # check that the POST request was successful (status code 200) + assert post_response.status_code == 200 + + # check json exists + assert post_response.json is not None + + # retrieve the token from the response + token = post_response.json.get("token") + assert token is not None + + # send a GET request to "/config" to retrieve the updated config + get_response = client.get("/config.json") + + # check if the GET request was successful (status code 200) + assert get_response.status_code == 200 + + # check the response content to verify the updated config data + config_data = json.loads(get_response.data) + assert config_data == new_config + + +@pytest.mark.flask +@pytest.mark.fixture +def test_reset_config_route(session_web_app: Flask) -> None: + """Test the route for resetting the configuration.""" + client = session_web_app.test_client() + + # store original config data + old_config_data_response = client.get("/config.json") + old_config_data = json.loads(old_config_data_response.data) + + # send a POST request with JSON data to update the configuration + new_config = {"key": "value"} + post_response = client.post("/update_config", json=new_config) + + # check that the POST request was successful (status code 200) + assert post_response.status_code == 200 + + # now, send a GET request to "/config" to retrieve the updated config + get_response = client.get("/config.json") + + # check if the GET request was successful (status code 200) + assert get_response.status_code == 200 + + # check the response content to verify the updated config data + config_data = json.loads(get_response.data) + assert config_data == new_config + + # send a GET request to reset the configuration + reset_response = client.get("/reset_config") + + # check that the request was successful (status code 200) + assert reset_response.status_code == 200 + + # now, send a GET request to "/config" to retrieve the reset config + reset_config_response = client.get("/config.json") + + # check if the GET request for reset config was successful (status code 200) + assert reset_config_response.status_code == 200 + + # check the response content to verify the reset config data + reset_config_data = json.loads(reset_config_response.data) + assert reset_config_data == old_config_data + + @pytest.mark.flask @pytest.mark.fixture def test_port_in_app_config(session_web_app: Flask) -> None: diff --git a/tests/test_website.py b/tests/test_website.py index e7471ca..7c869ae 100644 --- a/tests/test_website.py +++ b/tests/test_website.py @@ -11,13 +11,13 @@ from typing import Tuple import pytest +import requests from bs4 import BeautifulSoup from flask import Flask from selenium.webdriver.common.by import By from selenium.webdriver.remote.webelement import WebElement from seleniumbase import BaseCase -from tests.conftest import write_config_file from tests.schema import check_config_schema @@ -290,9 +290,7 @@ def test_custom_title_works( @pytest.mark.website -def test_form_backend_updated( - sb: BaseCase, live_session_web_app_url: str, submit_route: str -) -> None: +def test_form_backend_updated(sb: BaseCase, live_session_web_app_url: str) -> None: """Check that the form backend url has been updated correctly.""" # open the webpage sb.open(live_session_web_app_url) @@ -310,7 +308,7 @@ def test_form_backend_updated( assert form_target is not None # now check that it is the right url - assert form_target == live_session_web_app_url + submit_route + assert form_target == live_session_web_app_url + "/submit" @pytest.mark.website @@ -530,26 +528,31 @@ def test_form_download_required_constraint( sb.save_screenshot_to_logs() +@pytest.mark.debug @pytest.mark.feature def test_select_multiple_options( sb: BaseCase, - live_function_web_app_url: str, - submit_route: str, - function_websrc_tmp_dir: Path, - function_web_app: Flask, + live_session_web_app_url: str, multiple_select_options_config: Dict[str, Any], ) -> None: """Confirm multiple options can be selected.""" - # add form backend - multiple_select_options_config["form_backend_url"] = ( - live_function_web_app_url + submit_route + # update config + response = requests.post( + live_session_web_app_url + "/update_config", json=multiple_select_options_config ) - # update config file for testing multi select options - write_config_file(multiple_select_options_config, function_websrc_tmp_dir) + # check response + assert response.status_code == 200 + + # get token + token = response.json().get("token") + assert token is not None + + # update site URL + site_url = f"{live_session_web_app_url}?token={token}" # open site - sb.open(live_function_web_app_url) + sb.open(site_url) # get question name question_name = multiple_select_options_config["questions"][0]["name"] @@ -611,26 +614,31 @@ def test_select_multiple_options( sb.save_screenshot_to_logs() +@pytest.mark.debug @pytest.mark.feature def test_select_default_submission_rejected( sb: BaseCase, - live_function_web_app_url: str, - submit_route: str, - function_websrc_tmp_dir: Path, - function_web_app: Flask, + live_session_web_app_url: str, multiple_select_options_config: Dict[str, Any], ) -> None: """Confirm that default select options will not pass for submission.""" - # add form backend - multiple_select_options_config["form_backend_url"] = ( - live_function_web_app_url + submit_route + # update config + response = requests.post( + live_session_web_app_url + "/update_config", json=multiple_select_options_config ) - # update config file for testing multi select options - write_config_file(multiple_select_options_config, function_websrc_tmp_dir) + # check response + assert response.status_code == 200 + + # get token + token = response.json().get("token") + assert token is not None + + # update site URL + site_url = f"{live_session_web_app_url}?token={token}" # open site - sb.open(live_function_web_app_url) + sb.open(site_url) # get form form_element = sb.get_element("form")