Start and stop the flask server for your tests
from flask_fixture.decorators.decorator_endpoint import endpoint
from flask_fixture.fixture import local_server_url
from flask_fixture.decorators.decorator_config import config

flask_fixture/__init__.py
pomponchik/flask_fixture
from enum import Enum
from dataclasses import dataclass
from typing import Any


class ChunkType(Enum):
    LOG_RECORD = 1
    STDOUT = 2
    STDERR = 3


@dataclass
class ProcessOutputChunk:
    type: ChunkType
    value: Any

flask_fixture/dataclasses/output_chunk.py
pomponchik/flask_fixture
import inspect
from dataclasses import dataclass
from typing import Tuple, Dict, Callable, Any, Optional


@dataclass
class RouteItem:
    """
    Storage of route signature.
    """
    args: Tuple[Any, ...]
    kwargs: Dict[str, Any]
    function: Callable[[...], Any]

flask_fixture/dataclasses/route_item.py
pomponchik/flask_fixture
import json
from typing import Optional, Dict, Any
from dataclasses import dataclass, asdict


@dataclass
class RunningStartupResult:
    success: bool
    stdout: Optional[str] = None
    stderr: Optional[str] = None
    traceback: Optional[str] = None
    exception_name: Optional[str] = None
    exception_string: Optional[str] = None
    routes: Optional[list[str]] = None
    configs: Dict[str, Any] = None

flask_fixture/dataclasses/running_startup_result.py
pomponchik/flask_fixture
import inspect
from typing import Type
from dataclasses import dataclass, fields

from flask_fixture.errors import NotExpectedConfigFieldError
from flask_fixture.state_storage.collection_of_configs import configs


def config(Class: Type) -> Type:
    Class = dataclass(Class)

    class_fields = {field.name: field.default for field in fields(Class)}

    for field_name, field_value in class_fields.items():
        if field_name not in configs:
            available_fields = ', '.join([f'"{x}"' for x in configs])
            raise NotExpectedConfigFieldError(f'Option "{field_name}" is not provided. Available options: {available_fields}.')
        configs[field_name] = field_value

    return Class

flask_fixture/decorators/decorator_config.py
pomponchik/flask_fixture
import inspect
from typing import Any, Callable

from flask_fixture.dataclasses.route_item import RouteItem
from flask_fixture.state_storage.collection_of_routes import routes
from flask_fixture.state_storage.collection_of_importing_modules import modules
from flask_fixture.errors import NeedToDefineURLError


def endpoint(*args: Any, **kwargs: Any) -> Callable[[Callable[[...], Any]], Callable[[...], Any]]:
    """
    A decorator factory that registers endpoints in the Flask application.

    The URL (in short form) should always be passed here as the first positional argument.
    If it is not passed, the NeedToDefineURLError exception will be raised.

    All arguments that will be passed to this function will be used in the same form when creating routes for the Flask server.
    """
    if not args:
        raise NeedToDefineURLError('The first argument to the decorator is to pass the expected URL.')
    elif len(args) == 1 and callable(args[0]) and not kwargs:
        raise NeedToDefineURLError('The first argument to the decorator is to pass the expected URL.')

    def decorator(function: Callable[[...], Any]) -> Callable[[...], Any]:
        """
        This decorator simply adds the signatures of the registered routes to a global list.
        This list will be recreated in the generated process, and the data from it will be used to create Flask routes.
        """
        routes.append(
            RouteItem(
                args=args,
                kwargs=kwargs,
                function=function,
            )
        )
        modules.add(inspect.getmodule(function).__name__)
        return function
    return decorator

flask_fixture/decorators/decorator_endpoint.py
pomponchik/flask_fixture
class NeedToDefineURLError(Exception):
    """
    This exception is raised if the user has not passed arguments to the decorator.
    """

class NotExpectedConfigFieldError(Exception):
    pass

class UnsuccessfulProcessStartupError(Exception):
    pass

flask_fixture/errors.py
pomponchik/flask_fixture
import logging
import multiprocessing
from threading import Thread

import pytest
from awaits import shoot

from flask_fixture.runner import run_flask
from flask_fixture.state_storage.collection_of_routes import routes
from flask_fixture.state_storage.collection_of_importing_modules import modules
from flask_fixture.state_storage.collection_of_configs import configs
from flask_fixture.errors import NotExpectedConfigFieldError, UnsuccessfulProcessStartupError
from flask_fixture.dataclasses.output_chunk import ChunkType, ProcessOutputChunk



logger: logging.Handler = logging.getLogger('flask_fixture_logger')
handler: logging.StreamHandler = logging.StreamHandler()
logger.addHandler(handler)
handler.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] %(message)s'))

@shoot
def listen_logs(queue: multiprocessing.Queue):
    while True:
        chunk: ProcessOutputChunk = queue.get()

        try:
            if chunk.type == ChunkType.LOG_RECORD:
                record: logging.LogRecord = chunk.value
                logger.handle(record)

            elif chunk.type == ChunkType.STDOUT:
                string: str = chunk.value
                print(string, end='')

            elif chunk.type == ChunkType.STDERR:
                string: str = chunk.value
                print(string, end='')

        except Exception as e:
            print(e)


@pytest.fixture(scope='session')
def local_server_url(local_server_port: int = 5001) -> str:
    """
    A fixture that starts and then stops the Flask server through a separate process.

    A queue is used to synchronize with the generated process.
    The current process makes sure that initialization in the generated process has ended by receiving a message from the queue.

    The names of modules that need to be imported so that all @endpoint decorators are executed are also passed to the generated process.

    The fixture returns the URL of the server, where you can immediately make requests, taking into account the registered routes.
    """
    for config_field_name, config_field_value in configs.items():
        if config_field_value is None:
            raise NotExpectedConfigFieldError()

    queue = multiprocessing.Queue()

    process = multiprocessing.Process(target=run_flask, args=(queue, local_server_port, list(modules), configs))

    process.start()
    startup_result = queue.get()
    if not startup_result.success:
        exception = UnsuccessfulProcessStartupError(f'Error {startup_result.exception_name}("{startup_result.exception_string}") when starting the process with the Flask server.')
        exception.startup_result = startup_result
        exception.traceback_string = startup_result.traceback
        raise exception

    listen_logs(queue)

    yield f'http://localhost:{configs["port"]}/'

    process.terminate()

flask_fixture/fixture.py
pomponchik/flask_fixture
import os
import io
import logging
import traceback
from typing import Dict, List, Any
from contextlib import redirect_stdout, redirect_stderr
from multiprocessing import Queue

from flask import Flask
from awaits import shoot

from flask_fixture.state_storage.collection_of_routes import routes
from flask_fixture.dataclasses.running_startup_result import RunningStartupResult
from flask_fixture.dataclasses.output_chunk import ChunkType, ProcessOutputChunk


class QueueHandler(logging.Handler):
    def __init__(self, queue: Queue):
        logging.Handler.__init__(self)
        self.queue = queue

    def emit(self, record: logging.LogRecord):
        chunk: ProcessOutputChunk = ProcessOutputChunk(value=record, type=ChunkType.LOG_RECORD)
        self.queue.put(chunk)


def run_flask(queue: Queue, port: int, modules: List[str], configs: Dict[str, Any]) -> None:
    """
    The function is designed to run in a separate process. It starts the flask server.

    Before starting the server, it does some manipulations:
    1. Imports all modules that used the @endpoint decorator. This is necessary so that the decorator code is executed in the child process, and the list of endpoints descriptions is filled in.
    2. The Flask application is created and all the endpoints collected in the first step are registered in it.

    The server output (both stderr and stdout) is redirected to the trash.

    The queue is used to synchronize with the process that started this function. After the function is started, that process starts waiting for the function to put something in the queue. Thus, it is guaranteed that the starting process will not continue its execution before the server initialization occurs. This approach allows you to do without using sleep().
    """
    startup_result = RunningStartupResult(success=True, routes=[], configs=configs)
    try:
        for module in modules:
            __import__(module)

        app = Flask('flask_fixture', template_folder=os.path.abspath(configs['template_folder']))

        for route in routes:
            startup_result.routes.append(str(route))
            routing = app.route(*(route.args), **(route.kwargs))
            routing(route.function)

        handler = QueueHandler(queue)
        logging.root.addHandler(handler)

    except Exception as e:
        startup_result.success = False

        buffer = io.StringIO()
        traceback.print_exception(e, file=buffer)
        traceback_string = buffer.getvalue()

        startup_result.traceback = traceback_string
        startup_result.exception_name = e.__class__.__name__
        startup_result.exception_string = str(e)

    finally:
        queue.put(startup_result)
        if not startup_result.success:
            return

    buffer_stdout = io.StringIO()
    buffer_stderr = io.StringIO()

    with open(os.devnull, 'w') as devnull:
        with redirect_stdout(devnull), redirect_stderr(devnull):
            app.run(port=configs['port'])

flask_fixture/runner.py
pomponchik/flask_fixture
from typing import Dict, Any


configs: Dict[str, Any] = {
    'port': 5000,
    'template_folder': 'templates',
}

flask_fixture/state_storage/collection_of_configs.py
pomponchik/flask_fixture
from typing import Set


modules: Set[str] = set()

flask_fixture/state_storage/collection_of_importing_modules.py
pomponchik/flask_fixture
from typing import List

from flask_fixture.dataclasses.route_item import RouteItem


routes: List[RouteItem] = []

flask_fixture/state_storage/collection_of_routes.py
pomponchik/flask_fixture
from setuptools import setup, find_packages


with open('README.md', 'r', encoding='utf8') as readme_file:
    readme = readme_file.read()

requirements = ['flask', 'awaits==0.0.4']

setup(
    name='flask_fixture',
    version='0.0.4',
    author='Evgeniy Blinov',
    author_email='zheni-b@yandex.ru',
    description='Start and stop the flask server for your tests',
    long_description=readme,
    long_description_content_type='text/markdown',
    url='https://github.com/pomponchik/flask_fixture',
    packages=find_packages(exclude=['tests']),
    install_requires=requirements,
    entry_points={
        "pytest11": ["flask_fixture = flask_fixture"],
    },
    classifiers=[
        'Operating System :: MacOS :: MacOS X',
        'Operating System :: Microsoft :: Windows',
        'Operating System :: POSIX',
        'Operating System :: POSIX :: Linux',
        'Programming Language :: Python',
        'Programming Language :: Python :: 3.10',
        'Programming Language :: Python :: 3.11',
        'License :: OSI Approved :: MIT License',
        'Intended Audience :: Developers',
        'Topic :: Software Development :: Libraries',
        'Topic :: Software Development :: Testing',
        'Framework :: Pytest',
        'Framework :: Flask',
    ],
)

setup.py
pomponchik/flask_fixture
import os

import pytest
from flask import render_template

from flask_fixture import endpoint, config


@config
class MyConfig:
    template_folder: str = os.path.join(
        'tests',
        'smokes',
        'data',
        'templates',
    )

@endpoint('/')
def root():
    return 'kek'


@endpoint('/simple_template')
def simple_template():
    return render_template('index.html')

tests/smokes/conftest.py
pomponchik/flask_fixture
import os
import urllib
import time

import requests
import pytest

import urllib.parse


@pytest.mark.timeout(30)
def test_run_server(local_server_url):
    time.sleep(1)
    assert requests.get(local_server_url).text == 'kek'


@pytest.mark.timeout(30)
def test_render_simple_template(local_server_url):
    time.sleep(1)
    page_url = urllib.parse.urljoin(local_server_url, 'simple_template')
    page = requests.get(page_url).text

    with open(
        os.path.join(
            'tests',
            'smokes',
            'data',
            'templates',
            'index.html',
        ),
        'r',
    ) as file:
        file_content = file.read()
        assert file_content.strip() == page.strip()

tests/smokes/test_fixture.py
pomponchik/flask_fixture
import pytest

from flask_fixture import endpoint
from flask_fixture.errors import NeedToDefineURLError


def test_endpoint_without_arguments():
    with pytest.raises(NeedToDefineURLError):
        @endpoint()
        def kek():
            pass


def test_endpoint_without_breakets():
    with pytest.raises(NeedToDefineURLError):
        @endpoint
        def kek():
            pass

tests/units/decorators/test_decorator_endpoint.py
pomponchik/flask_fixture