test

WSGIServerForm dataclass

Dataclass to register a server form.

This dataclass would be used to register as a form into test classes such as WSGITestExecutor.

Parameters:

Name	Type	Description	Default
host		Hostname of the server.	required
port		Port of the server.	required
app		WSGIApp object with implemented Endpoints.	required
path_log		Path log of the server will be written.	required

WSGITestExecutor

Utility class that can execute server applications at child processes.

This class can be used in test scripts for holding several server forms and running the application based on the forms at the child processes. Also this object has feature of context manager and developers may use it to kill the processes safely.

__init__(self, *forms) special

Parameters:

Name	Type	Description	Default
*forms	WSGIServerForm	Dataclass describing information of the server application.	()

Source code in bamboo/test.py

def __init__(self, *forms: WSGIServerForm) -> None:
    """
    Args:
        *forms: Dataclass describing information of
            the server application.
    """
    self._forms: t.List[WSGIServerForm] = []
    self._children: t.List[multiprocessing.Process] = []

    self.add_forms(*forms)

add_forms(self, *forms)

Add forms with information of server applications.

Parameters:

Name	Type	Description	Default
*foms		Dataclass describing information of the server application	required

Source code in bamboo/test.py

def add_forms(self, *forms: WSGIServerForm) -> None:
    """Add forms with information of server applications.

    Args:
        *foms: Dataclass describing information of
            the server application
    """
    for form in forms:
        self._forms.append(form)

close(self, pop=True)

Kill the all child processes derived from registered forms.

Parameters:

Name	Type	Description	Default
pop	bool	If the registered forms is to be removed.	True

Source code in bamboo/test.py

def close(self, pop: bool = True) -> None:
    """Kill the all child processes derived from registered forms.

    Args:
        pop: If the registered forms is to be removed.
    """
    for child in self._children:
        child.terminate()
        child.join()
        child.close()

    if pop:
        self._children.clear()

debug(app, host='localhost', port=8000) staticmethod

Executes a server application for debug.

This method is a kind of shorcut for launching a server application and can be used to debug the application. If you want to deploy an application made with Bamboo, consider to use another WSGI server application for production.

Parameters:

Name	Type	Description	Default
app	WSGIApp	WSGIApp object with implemented Endpoints.	required
host	str	Hostname of the server.	'localhost'
port	int	Port of the server.	8000

Source code in bamboo/test.py

@staticmethod
def debug(
    app: WSGIApp,
    host: str = "localhost",
    port: int = 8000,
) -> None:
    """Executes a server application for debug.

    This method is a kind of shorcut for launching a server application
    and can be used to debug the application. If you want to deploy an
    application made with Bamboo, consider to use another WSGI server
    application for production.

    Args:
        app: WSGIApp object with implemented Endpoints.
        host: Hostname of the server.
        port: Port of the server.
    """
    server = wsgiref.simple_server.make_server(host, port, app)
    try:
        print(f"Hosting on {host}:{port} ...")
        print(insert_colorcode(
            "WARNING: This is debug mode. "
            "Do not use it in your production deployment.",
            ColorCode.RED
        ))
        server.serve_forever()
    except KeyboardInterrupt:
        server.server_close()
        print()

exec(self, func, args=(), waiting=0.1)

Executes a simple client-server test.

Parameters:

Name	Type	Description	Default
func	t.Callable[[t.Tuple[t.Any, ...]], None]	Function executed after all the server applications start.	required
args	t.Tuple[t.Any, ...]	Arguments of the func.	()
waiting	float	Waiting time after running the applications.	0.1

Source code in bamboo/test.py

def exec(
    self,
    func: t.Callable[[t.Tuple[t.Any, ...]], None],
    args: t.Tuple[t.Any, ...] = (),
    waiting: float = 0.1,
) -> None:
    """Executes a simple client-server test.

    Args:
        func: Function executed after all the server applications start.
        args: Arguments of the func.
        waiting: Waiting time after running the applications.
    """
    with self.start_serve(waiting=waiting):
        func(*args)

start_serve(self, waiting=0.05)

Run registered server applications at child processes.

This object has feature of context manager and the method returns the object itself. So developer can use the with sentence and in it, can define logic of clients.

waiting: Waiting time after running the processes.

Returns:

Type	Description
WSGITestExecutor	This object itself.

Examples:

>>> holder = WSGITestExecutor(form)
>>> with holder.start_serve():
...    res = http_get("http://localhost:8000/image")
>>> print(res.body)

Source code in bamboo/test.py

def start_serve(self, waiting: float = 0.05) -> WSGITestExecutor:
    """Run registered server applications at child processes.

    This object has feature of context manager and the method
    returns the object itself. So developer can use the with
    sentence and in it, can define logic of clients.

    Args:
    waiting: Waiting time after running the processes.

    Returns:
        This object itself.

    Examples:
        ```python
        >>> holder = WSGITestExecutor(form)
        >>> with holder.start_serve():
        ...    res = http_get("http://localhost:8000/image")
        >>> print(res.body)
        ```
    """
    for form in self._forms:
        child = multiprocessing.Process(target=serve_at, args=(form,))
        child.start()
        self._children.append(child)
    time.sleep(waiting)

    return self

serve_at(form)

Subroutine for server application called at a child process.

Parameters:

Name	Type	Description	Default
form	WSGIServerForm	Dataclass describing information of the server application.	required

Source code in bamboo/test.py

def serve_at(form: WSGIServerForm) -> None:
    """Subroutine for server application called at a child process.

    Args:
        form: Dataclass describing information of the server application.
    """
    server = wsgiref.simple_server.make_server(form.host, form.port, form.app)
    f_log = open(form.path_log, "wt")
    sys.stdout = f_log
    sys.stderr = f_log

    def server_close(signalnum, frame):
        print()
        f_log.flush()
        f_log.close()
        sys.exit()

    signal.signal(signal.SIGTERM, server_close)
    signal.signal(signal.SIGINT, server_close)
    server.serve_forever()