diff --git a/.gitignore b/.gitignore index c59bbbb2..7f087caa 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ # Python __pycache__/ .venv/ +.hypothesis/ # LSP .vscode/ diff --git a/README.md b/README.md index e321fd1c..1a0042a7 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,70 @@ This is a work-in-progress! +## Installation + +It's highly recommended to install from source at the moment: + +``` +$ pip install git+https://github.com/zerointensity/view.py +``` + +## Examples + +### Simple Hello World + +```py +from view.core.app import App + +from view.dom.core import html_response +from view.dom.components import page +from view.dom.primitives import h1 + +app = App() + + +@app.get("/") +@html_response +async def home(): + with page("Hello, view.py!"): + yield h1("Nobody expects the Spanish Inquisition") + + +app.run() +``` + +### Button Counter + +```py +from view.core.app import App +from view.dom.core import HTMLNode, html_response +from view.dom.components import page +from view.dom.primitives import button, p + +from view.javascript import javascript_compiler, as_javascript_expression + +app = App() + + +@javascript_compiler +def click_button(counter: HTMLNode): + yield f"let node = {as_javascript_expression(counter)};" + yield f"let currentNumber = parseInt(node.innerHTML);" + yield f"node.innerHTML = ++currentNumber;" + + +@app.get("/") +@html_response +async def home(): + with page("Counter"): + count = p("0") + yield count + yield button("Click me!", onclick=click_button(count)) + + +app.run() +``` + ## Copyright `view.py` is distributed under the terms of the [MIT](https://spdx.org/licenses/MIT.html) license. diff --git a/hatch.toml b/hatch.toml index fcfec678..9510317d 100644 --- a/hatch.toml +++ b/hatch.toml @@ -17,11 +17,12 @@ extra-dependencies = [ "daphne", "gunicorn", "werkzeug", + "hypothesis", ] randomize = true retries = 3 retries-delay = 1 -parallel = false +parallel = true [[envs.hatch-test.matrix]] python = ["3.14", "3.13", "3.12", "3.11", "3.10"] diff --git a/pyproject.toml b/pyproject.toml index 64135cf6..e8976046 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,7 +23,7 @@ classifiers = [ dependencies = ["typing_extensions>=4"] dynamic = ["version", "license"] -[project.optional-dependencies] +#[project.optional-dependencies] [project.urls] Documentation = "https://view.zintensity.dev" @@ -31,9 +31,9 @@ Issues = "https://github.com/ZeroIntensity/view.py/issues" Source = "https://github.com/ZeroIntensity/view.py" Funding = "https://github.com/sponsors/ZeroIntensity" -[project.scripts] -view = "view.__main__:main" -view-py = "view.__main__:main" +#[project.scripts] +#view = "view.__main__:main" +#view-py = "view.__main__:main" [tool.ruff] exclude = ["tests/", "docs/"] diff --git a/src/view/__main__.py b/src/view/__main__.py deleted file mode 100644 index cd9ac480..00000000 --- a/src/view/__main__.py +++ /dev/null @@ -1,6 +0,0 @@ -def main(): - pass - - -if __name__ == "__main__": - main() diff --git a/src/view/cache.py b/src/view/cache.py index 2fa1d436..6ca829da 100644 --- a/src/view/cache.py +++ b/src/view/cache.py @@ -98,18 +98,39 @@ async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> Response: def minutes(number: int, /) -> int: + """ + Convert minutes to seconds. + + This is for use in cache decorators. + """ return number * 60 def seconds(number: int, /) -> int: + """ + Do nothing and return ``number``. This only exists for making it + semantically clear that the intended time is seconds. + + This is for use in cache decorators. + """ return number def hours(number: int, /) -> int: + """ + Convert hours to seconds. + + This is for use in cache decorators. + """ return minutes(60) * number def days(number: int, /) -> int: + """ + Convert days to seconds. + + This is for use in cache decorators. + """ return hours(24) * number diff --git a/src/view/core/app.py b/src/view/core/app.py index 5836b474..de9a58f8 100644 --- a/src/view/core/app.py +++ b/src/view/core/app.py @@ -16,7 +16,7 @@ from importlib.metadata import Distribution, PackageNotFoundError from multiprocessing import Process from pathlib import Path -from typing import TYPE_CHECKING, ParamSpec, TypeAlias, TypeVar +from typing import TYPE_CHECKING, ParamSpec, TypeAlias, TypeVar, Unpack from view.core._colors import ColorfulFormatter from view.core.request import Method, Request @@ -35,6 +35,7 @@ ) from view.exceptions import InvalidTypeError from view.responses import FileResponse +from view.run.servers import ServerConfigArgs, run_app_on_any_server from view.utils import reraise if TYPE_CHECKING: @@ -195,14 +196,7 @@ def asgi(self) -> ASGIProtocol: return asgi_for_app(self) - def run( - self, - *, - host: str = "localhost", - port: int = 5000, - production: bool = False, - server_hint: str | None = None, - ) -> None: + def run(self, **kwargs: Unpack[ServerConfigArgs]) -> None: """ Run the app. @@ -210,8 +204,8 @@ def run( finer control over the server settings is desired, explicitly use the server's API with the app's :meth:`asgi` or :meth:`wsgi` method. """ - from view.run.servers import ServerSettings + production = kwargs.get("production", False) # If production is True, then __debug__ should be False. # If production is False, then __debug__ should be True. if production is __debug__: @@ -230,11 +224,11 @@ def run( "If that doesn't sound correct, set VIEW_DEVMODE to 0." ) - self.logger.info("Serving app on http://localhost:%d", port) - self._production = production - settings = ServerSettings(self, host=host, port=port, hint=server_hint) + self.logger.info( + "Serving app on http://localhost:%d", kwargs.get("port") or 5000 + ) try: - settings.run_app_on_any_server() + run_app_on_any_server(self, **kwargs) except KeyboardInterrupt: self.logger.info("CTRL^C received, shutting down") except Exception: @@ -244,11 +238,7 @@ def run( def run_detached( self, - *, - host: str = "localhost", - port: int = 5000, - production: bool = False, - server_hint: str | None = None, + **kwargs: Unpack[ServerConfigArgs], ) -> Process: """ Run the app in a separate process. This means that the server is @@ -257,12 +247,7 @@ def run_detached( process = Process( target=self.run, - kwargs={ - "host": host, - "port": port, - "production": production, - "server_hint": server_hint, - }, + kwargs=kwargs, ) process.start() return process diff --git a/src/view/core/headers.py b/src/view/core/headers.py index 8f94c28b..4981338c 100644 --- a/src/view/core/headers.py +++ b/src/view/core/headers.py @@ -4,7 +4,7 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Iterable, Mapping from typing import TYPE_CHECKING, Any, TypeAlias from typing_extensions import Self @@ -62,6 +62,9 @@ class HTTPHeaders(MultiMap[str, str]): Case-insensitive multi-map of HTTP headers. """ + def __init__(self, items: Iterable[tuple[str, str]] = ()) -> None: + super().__init__((LowerStr(key), value) for key, value in items) + def __getitem__(self, key: str, /) -> str: return super().__getitem__(LowerStr(key)) @@ -71,6 +74,19 @@ def __contains__(self, key: object, /) -> bool: def __repr__(self) -> str: return f"HTTPHeaders({self.as_sequence()})" + def __eq__(self, other: object, /) -> bool: + if isinstance(other, HTTPHeaders): + return other._values == self._values + + if isinstance(other, dict): + return self._as_flat() == { + LowerStr(key): value for key, value in other.items() + } + + return NotImplemented + + __hash__ = MultiMap.__hash__ + def get_exactly_one(self, key: str) -> str: return super().get_exactly_one(LowerStr(key)) diff --git a/src/view/core/router.py b/src/view/core/router.py index 223085f1..70b25780 100644 --- a/src/view/core/router.py +++ b/src/view/core/router.py @@ -67,24 +67,24 @@ class DuplicateRouteError(ViewError): @dataclass(slots=True) -class PathNode: +class _PathNode: """ A node in the "path tree". """ name: str routes: MutableMapping[Method, Route] = field(default_factory=dict) - children: MutableMapping[str, PathNode] = field(default_factory=dict) - path_parameter: PathNode | None = None + children: MutableMapping[str, _PathNode] = field(default_factory=dict) + path_parameter: _PathNode | None = None subrouter: SubRouter | None = None - def parameter(self, name: str) -> PathNode: + def parameter(self, name: str) -> _PathNode: """ Mark this node as having a path parameter (if not already), and return the path parameter node. """ if self.path_parameter is None: - next_node = PathNode(name=name) + next_node = _PathNode(name=name) self.path_parameter = next_node return next_node if __debug__ and name != self.path_parameter.name: @@ -94,7 +94,7 @@ def parameter(self, name: str) -> PathNode: ) return self.path_parameter - def next(self, part: str) -> PathNode: + def next_node(self, part: str) -> _PathNode: """ Get the next node for the given path part, creating it if it doesn't exist. @@ -103,19 +103,19 @@ def next(self, part: str) -> PathNode: if node is not None: return node - new_node = PathNode(name=part) + new_node = _PathNode(name=part) self.children[part] = new_node return new_node -def is_path_parameter(part: str) -> bool: +def _is_path_parameter(part: str) -> bool: """ Is this part a path parameter? """ return part.startswith("{") and part.endswith("}") -def extract_path_parameter(part: str) -> str: +def _extract_path_parameter(part: str) -> str: """ Extract the name of a path parameter from a string given by the user in a route string. @@ -143,11 +143,11 @@ class Router: error_views: MutableMapping[type[HTTPError], RouteView] = field( default_factory=dict ) - parent_node: PathNode = field(default_factory=lambda: PathNode(name="")) + parent_node: _PathNode = field(default_factory=lambda: _PathNode(name="")) def _get_node_for_path( self, path: str, *, allow_path_parameters: bool - ) -> PathNode: + ) -> _PathNode: if __debug__ and not isinstance(path, str): raise InvalidTypeError(path, str) @@ -156,14 +156,14 @@ def _get_node_for_path( parts = path.split("/") for part in parts: - if is_path_parameter(part): + if _is_path_parameter(part): if not allow_path_parameters: raise RuntimeError("Path parameters are not allowed here") parent_node = parent_node.parameter( - extract_path_parameter(part) + _extract_path_parameter(part) ) else: - parent_node = parent_node.next(part) + parent_node = parent_node.next_node(part) return parent_node diff --git a/src/view/dom/core.py b/src/view/dom/core.py index beeb1932..9e7cb2ac 100644 --- a/src/view/dom/core.py +++ b/src/view/dom/core.py @@ -26,6 +26,7 @@ if TYPE_CHECKING: from view.core.router import RouteView + from view.dom.components import Component __all__ = ("HTMLNode", "html_response") @@ -175,7 +176,7 @@ def html_context() -> HTMLTree: P = ParamSpec("P") -HTMLViewResponseItem: TypeAlias = HTMLNode | int +HTMLViewResponseItem: TypeAlias = "HTMLNode | int | Component" HTMLViewResult = ( AsyncIterator[HTMLViewResponseItem] | Iterator[HTMLViewResponseItem] ) diff --git a/src/view/javascript.py b/src/view/javascript.py index 4cc9ca4c..c9c526fa 100644 --- a/src/view/javascript.py +++ b/src/view/javascript.py @@ -67,6 +67,8 @@ def as_javascript_expression(data: object) -> str: if __debug__ and not isinstance(result, str): raise InvalidTypeError(result, str) + return result + raise TypeError( f"Don't know how to convert {data!r} to a JavaScript expression" ) diff --git a/src/view/run/servers.py b/src/view/run/servers.py index 6ab7ffdb..a2f2de71 100644 --- a/src/view/run/servers.py +++ b/src/view/run/servers.py @@ -4,10 +4,17 @@ from __future__ import annotations -from collections.abc import Callable, Sequence +from collections.abc import Callable, MutableMapping from contextlib import suppress from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, ClassVar, TypeAlias +from typing import ( + TYPE_CHECKING, + Any, + NotRequired, + TypeAlias, + TypedDict, + Unpack, +) if TYPE_CHECKING: from view.core.app import BaseApp @@ -15,9 +22,7 @@ from view.exceptions import ViewError -__all__ = ("ServerSettings",) - -StartServer: TypeAlias = Callable[[], None] +__all__ = ("run_app_on_any_server",) class BadServerError(ViewError): @@ -29,138 +34,151 @@ class BadServerError(ViewError): """ +class ServerConfigArgs(TypedDict): + host: NotRequired[str] + port: NotRequired[int] + production: NotRequired[bool] + server_hint: NotRequired[str] + + @dataclass(slots=True, frozen=True) class ServerSettings: + host: str + port: int + production: bool + + @classmethod + def from_kwargs(cls, kwargs: ServerConfigArgs, /) -> ServerSettings: + return cls( + kwargs.get("host") or "localhost", + kwargs.get("port") or 5000, + kwargs.get("production") or False, + ) + + +def run_uvicorn(app: BaseApp, settings: ServerSettings) -> None: """ - Dataclass representing server settings that can be used to start - serving an app. + Run the app using the ``uvicorn`` library. """ + import uvicorn - AVAILABLE_SERVERS: ClassVar[Sequence[str]] = [ - "uvicorn", - "hypercorn", - "daphne", - "gunicorn", - "werkzeug", - "wsgiref", - ] + uvicorn.run(app.asgi(), host=settings.host, port=settings.port) - app: BaseApp - port: int - host: str - hint: str | None = None - - def run_uvicorn(self) -> None: - """ - Run the app using the ``uvicorn`` library. - """ - import uvicorn - - uvicorn.run(self.app.asgi(), host=self.host, port=self.port) - - def run_hypercorn(self) -> None: - """ - Run the app using the ``hypercorn`` library. - """ - import asyncio - - import hypercorn - from hypercorn.asyncio import serve - - config = hypercorn.Config() - config.bind = [f"{self.host}:{self.port}"] - asyncio.run(serve(self.app.asgi(), config)) # type: ignore - - def run_daphne(self) -> None: - """ - Run the app using the ``daphne`` library. - """ - from daphne.endpoints import build_endpoint_description_strings - from daphne.server import Server - - endpoints = build_endpoint_description_strings( - host=self.host, - port=self.port, - ) - server = Server(self.app.asgi(), endpoints=endpoints) - server.run() - - def run_gunicorn(self) -> None: - """ - Run the app using the ``gunicorn`` library. - """ - from gunicorn.app.base import BaseApplication - - class GunicornRunner(BaseApplication): - def __init__( - self, app: WSGIProtocol, options: dict[str, Any] | None = None - ) -> None: - self.options = options or {} - self.application = app - super().__init__() - - def load_config(self): - assert self.cfg is not None - for key, value in self.options.items(): - if key in self.cfg.settings and value is not None: - self.cfg.set(key, value) - - def load(self): - return self.application - - runner = GunicornRunner( - self.app.wsgi(), {"bind": f"{self.host}:{self.port}"} - ) - runner.run() - - def run_werkzeug(self) -> None: - """ - Run the app using the ``werkzeug`` library. - """ - from werkzeug.serving import run_simple - - run_simple(self.host, self.port, self.app.wsgi()) - - def run_wsgiref(self) -> None: - """ - Run the app using the built-in :mod:`wsgiref` module. - """ - from wsgiref.simple_server import make_server - - with make_server(self.host, self.port, self.app.wsgi()) as server: - server.serve_forever() - - def run_app_on_any_server(self) -> None: - """ - Run the app on the nearest available ASGI or WSGI server. - - This will always succeed, as it will fall back to the standard - :mod:`wsgiref` module if no other server is installed. - """ - servers: dict[str, StartServer] = { - "uvicorn": self.run_uvicorn, - "hypercorn": self.run_hypercorn, - "daphne": self.run_daphne, - "gunicorn": self.run_gunicorn, - "werkzeug": self.run_werkzeug, - "wsgiref": self.run_wsgiref, - } - if self.hint is not None: - try: - start_server = servers[self.hint] - except KeyError as key_error: - raise BadServerError( - f"{self.hint!r} is not a known server" - ) from key_error - - try: - return start_server() - except ImportError as error: - raise BadServerError( - f"{self.hint} is not installed" - ) from error - - # I'm not sure what Ruff is complaining about here - for start_server in servers.values(): - with suppress(ImportError): - return start_server() + +def run_hypercorn(app: BaseApp, settings: ServerSettings) -> None: + """ + Run the app using the ``hypercorn`` library. + """ + import asyncio + + import hypercorn + from hypercorn.asyncio import serve + + config = hypercorn.Config() + config.bind = [f"{settings.host}:{settings.port}"] + asyncio.run(serve(app.asgi(), config)) # type: ignore + + +def run_daphne(app: BaseApp, settings: ServerSettings) -> None: + """ + Run the app using the ``daphne`` library. + """ + from daphne.endpoints import build_endpoint_description_strings + from daphne.server import Server + + endpoints = build_endpoint_description_strings( + host=settings.host, + port=settings.port, + ) + server = Server(app.asgi(), endpoints=endpoints) + server.run() + + +def run_gunicorn(app: BaseApp, settings: ServerSettings) -> None: + """ + Run the app using the ``gunicorn`` library. + """ + from gunicorn.app.base import BaseApplication + + class GunicornRunner(BaseApplication): + def __init__( + self, app: WSGIProtocol, options: dict[str, Any] | None = None + ) -> None: + self.options = options or {} + self.application = app + super().__init__() + + def load_config(self): + assert self.cfg is not None + for key, value in self.options.items(): + if key in self.cfg.settings and value is not None: + self.cfg.set(key, value) + + def load(self): + return self.application + + runner = GunicornRunner( + app.wsgi(), {"bind": f"{settings.host}:{settings.port}"} + ) + runner.run() + + +def run_werkzeug(app: BaseApp, settings: ServerSettings) -> None: + """ + Run the app using the ``werkzeug`` library. + """ + from werkzeug.serving import run_simple + + run_simple(settings.host, settings.port, app.wsgi()) + + +def run_wsgiref(app: BaseApp, settings: ServerSettings) -> None: + """ + Run the app using the built-in :mod:`wsgiref` module. + """ + from wsgiref.simple_server import make_server + + with make_server(settings.host, settings.port, app.wsgi()) as server: + server.serve_forever() + + +StartServer: TypeAlias = Callable[["BaseApp", ServerSettings], None] + +ALL_SERVERS: MutableMapping[str, StartServer] = { + "uvicorn": run_uvicorn, + "hypercorn": run_hypercorn, + "daphne": run_daphne, + "gunicorn": run_gunicorn, + "werkzeug": run_werkzeug, + "wsgiref": run_wsgiref, +} + + +def run_app_on_any_server( + app: BaseApp, **kwargs: Unpack[ServerConfigArgs] +) -> None: + """ + Run the app on the nearest available ASGI or WSGI server. + + This will always succeed, as it will fall back to the standard + :mod:`wsgiref` module if no other server is installed. + """ + settings = ServerSettings.from_kwargs(kwargs) + hint = kwargs.get("server_hint") + if hint is not None: + try: + start_server = ALL_SERVERS[hint] + except KeyError as key_error: + raise BadServerError( + f"{hint!r} is not a known server" + ) from key_error + + try: + return start_server(app, settings) + except ImportError as error: + raise BadServerError(f"{hint} is not installed") from error + + # I'm not sure what Ruff is complaining about here + for start_server in ALL_SERVERS.values(): + with suppress(ImportError): + return start_server(app, settings) diff --git a/tests/test_dom.py b/tests/test_dom.py index 5c123ec1..5f677292 100644 --- a/tests/test_dom.py +++ b/tests/test_dom.py @@ -7,6 +7,7 @@ from view.dom.core import HTMLNode, html_context, html_response from view.dom.primitives import ALL_PRIMITIVES, div, html, p from view.testing import AppTestClient +from view.javascript import SupportsJavaScript def html_function( @@ -15,9 +16,12 @@ def html_function( with html(lang="en"): with div(data={"foo": "bar"}): if has_body: - yield node("gotcha", data={"silly": "a"}) + the_node = node("gotcha", data={"silly": "a"}) else: - yield node(data={"silly": "a"}) + the_node = node(data={"silly": "a"}) + + assert isinstance(the_node, SupportsJavaScript) + yield the_node @pytest.mark.parametrize("dom_node", ALL_PRIMITIVES) diff --git a/tests/test_misc.py b/tests/test_misc.py index b7093f5f..2ef52fcb 100644 --- a/tests/test_misc.py +++ b/tests/test_misc.py @@ -58,7 +58,7 @@ def test_empty_multi_map(): def test_multi_map_no_duplicates(): - data = [('a', 1), ('b', 2), ('c', 3)] + data = [("a", 1), ("b", 2), ("c", 3)] multi_map = MultiMap(data) assert multi_map == {"a": 1, "b": 2, "c": 3} @@ -82,9 +82,8 @@ def test_multi_map_no_duplicates(): assert called == 3 - def test_multi_map_with_duplicates(): - data = [('a', 1), ('a', 2), ('a', 3), ('b', 4)] + data = [("a", 1), ("a", 2), ("a", 3), ("b", 4)] multi_map = MultiMap(data) assert len(multi_map) == 2 assert multi_map.as_sequence() == data @@ -95,14 +94,14 @@ def test_multi_map_with_duplicates(): assert "a" in multi_map assert "b" in multi_map - assert list(multi_map.keys()) == ['a', 'b'] + assert list(multi_map.keys()) == ["a", "b"] assert list(multi_map.values()) == [1, 4] - assert list(multi_map.items()) == [('a', 1), ('b', 4)] + assert list(multi_map.items()) == [("a", 1), ("b", 4)] assert list(multi_map.many_values()) == [[1, 2, 3], [4]] - assert list(multi_map.many_items()) == [('a', [1, 2, 3]), ('b', [4])] + assert list(multi_map.many_items()) == [("a", [1, 2, 3]), ("b", [4])] with pytest.raises(HasMultipleValuesError): - multi_map.get_exactly_one('a') + multi_map.get_exactly_one("a") assert multi_map.get_exactly_one("b") == 4 @@ -115,11 +114,11 @@ def test_multi_map_with_duplicates(): def test_multi_map_with_new_value(): - data = [('a', 1), ('b', 2), ('b', 3)] + data = [("a", 1), ("b", 2), ("b", 3)] multi_map = MultiMap(data) assert len(multi_map) == 2 - new_map = multi_map.with_new_value('b', 4) + new_map = multi_map.with_new_value("b", 4) assert len(new_map) == 2 assert "b" in new_map assert multi_map != new_map diff --git a/tests/test_requests.py b/tests/test_requests.py index 7ebc4565..a885c5fb 100644 --- a/tests/test_requests.py +++ b/tests/test_requests.py @@ -1,6 +1,7 @@ import json from collections.abc import AsyncIterator +from hypothesis import given, strategies import pytest from view.core.app import App, as_app from view.core.body import InvalidJSONError @@ -39,33 +40,35 @@ def app(request: Request) -> ResponseLike: @pytest.mark.asyncio -async def test_manual_request(): +@given(strategies.text(), strategies.binary(), strategies.text()) +async def test_manual_request(path: str, body: bytes, content: str): @as_app - def app(request: Request) -> ResponseLike: + async def app(request: Request) -> ResponseLike: assert request.app == app assert request.app.current_request() is request assert isinstance(request.path, str) assert request.method is Method.POST assert request.headers["test"] == "42" + assert (await request.body()) == body - return "1" + return content - async def stream_none() -> AsyncIterator[bytes]: - yield b"" + async def stream_body() -> AsyncIterator[bytes]: + yield body with pytest.raises(LookupError): app.current_request() manual_request = Request( - receive_data=stream_none, + receive_data=stream_body(), app=app, - path="/", + path=path, method=Method.POST, headers=as_real_headers({"test": "42"}), query_parameters=MultiMap(), ) response = await app.process_request(manual_request) - assert (await response.body()) == b"1" + assert (await response.body()) == content.encode("utf-8") @pytest.mark.asyncio @@ -141,41 +144,46 @@ def goodbye(): @pytest.mark.asyncio -async def test_request_path_parameters(): +@given( + # This is super ugly but don't worry about it + a=strategies.text().map(lambda x: x.replace("/", "")).filter(lambda x: (x not in {'', 'a'}) and ('?' not in x)), + b=strategies.text().map(lambda x: x.replace("/", "")).filter(lambda x: (x != '') and ('?' not in x)), +) +async def test_request_path_parameters(a: str, b: str): app = App() @app.get("/") def index(): return "Index" - @app.get("/spanish/{inquisition}") + @app.get("/oneparam/{a}") async def path_param(): request = app.current_request() - assert request.path_parameters["inquisition"] == "42" + assert request.path_parameters["a"] == a return "0" - @app.get("/spanish/inquisition") + @app.get("/oneparam/a") def overwrite_path_param(): return "1" - @app.get("/spanish/inquisition/{nobody}") + @app.get("/nested/param/{b}") def sub_path_param(): request = app.current_request() - assert request.path_parameters["nobody"] == "gotcha" + assert request.path_parameters["b"] == b return "2" - @app.get("/spanish/{inquisition}/{nobody}") + @app.get("/twoparam/{a}/{b}") def double_path_param(): request = app.current_request() - assert request.path_parameters["inquisition"] == "1" - assert request.path_parameters["nobody"] == "2" + assert request.path_parameters["a"] == a + assert request.path_parameters["b"] == b return "3" client = AppTestClient(app) - assert (await into_tuple(client.get("/spanish/42"))) == ok("0") - assert (await into_tuple(client.get("/spanish/inquisition"))) == ok("1") - assert (await into_tuple(client.get("/spanish/inquisition/gotcha"))) == ok("2") - assert (await into_tuple(client.get("/spanish/1/2"))) == ok("3") + assert (await into_tuple(client.get(f"/oneparam/{a}"))) == ok("0") + assert (await into_tuple(client.get("/oneparam/a"))) == ok("1") + assert (await into_tuple(client.get(f"/nested/param/{b}"))) == ok("2") + assert (await into_tuple(client.get(f"/twoparam/{a}/{b}"))) == ok("3") @pytest.mark.asyncio diff --git a/tests/test_responses.py b/tests/test_responses.py index 8afc0ad2..80bced8b 100644 --- a/tests/test_responses.py +++ b/tests/test_responses.py @@ -17,6 +17,8 @@ from view.responses import JSONResponse, FileResponse from view.testing import AppTestClient, bad, into_tuple, ok +from hypothesis import given, strategies + @pytest.mark.asyncio async def test_str_or_bytes_response(): @@ -228,3 +230,35 @@ async def test_static_files(): 200, {"content-type": "text/plain"}, ) + +@pytest.mark.asyncio +async def test_header_case_insensitivity(): + @as_app + async def app(_: Request): + return "a", 200, {"Foo": "bar"} + + + client = AppTestClient(app) + assert (await into_tuple(client.get("/"))) == (b"a", 200, {"foo": "bar"}) + assert (await into_tuple(client.get("/"))) == (b"a", 200, {"FOO": "bar"}) + + +@pytest.mark.asyncio +@given( + strategies.text(), + strategies.integers(min_value=200, max_value=208), + strategies.dictionaries(strategies.text(), strategies.text()), +) +async def test_hypothesis_with_responses( + response: str, status: int, headers: dict[str, str] +): + @as_app + async def app(_: Request): + return response, status, headers + + client = AppTestClient(app) + assert (await into_tuple(client.get("/"))) == ( + response.encode("utf-8"), + status, + headers, + ) diff --git a/tests/test_servers.py b/tests/test_servers.py index a7d9d055..a4084096 100644 --- a/tests/test_servers.py +++ b/tests/test_servers.py @@ -9,12 +9,35 @@ from view.core.request import Request from view.core.response import ResponseLike from view.core.status_codes import Success -from view.run.servers import ServerSettings +from view.run.servers import ALL_SERVERS +from threading import Lock -@pytest.mark.parametrize("server_name", ServerSettings.AVAILABLE_SERVERS) +_PORT_LOCK = Lock() +_PORT: int = 5000 + +@pytest.fixture(scope="function") +def port() -> int: + with _PORT_LOCK: + global _PORT + _PORT += 1 + return _PORT + + +def wait_for_server(port: int, timeout: float = 10.0, interval: float = 0.1) -> bool: + deadline = time.time() + timeout + while time.time() < deadline: + try: + requests.get(f"http://localhost:{port}", timeout=1) + return True + except requests.ConnectionError: + time.sleep(interval) + return False + + +@pytest.mark.parametrize("server_name", ALL_SERVERS) @pytest.mark.skipif(platform.system() != "Linux", reason="this has issues on non-Linux") -def test_run_server(server_name: str): +def test_run_server(server_name: str, port: int): try: __import__(server_name) except ImportError: @@ -29,18 +52,19 @@ def test_run_server(server_name: str): async def index(): return 'ok' - app.run(server_hint={server_name!r}) + app.run(server_hint={server_name!r}, port={port}) """ process = subprocess.Popen([sys.executable, "-c", code]) try: - time.sleep(2) - response = requests.get("http://localhost:5000") + if not wait_for_server(port): + pytest.fail("Server did not start in time") + response = requests.get(f"http://localhost:{port}") assert response.text == "ok" finally: process.kill() -@pytest.mark.parametrize("server_name", ServerSettings.AVAILABLE_SERVERS) +@pytest.mark.parametrize("server_name", ALL_SERVERS) @pytest.mark.skip("some multiprocessing problems at the moment") def test_run_server_detached(server_name: str): @as_app