diff --git a/.cspell/dictionary.txt b/.cspell/dictionary.txt index 4c4b52e..e5fc8a7 100644 --- a/.cspell/dictionary.txt +++ b/.cspell/dictionary.txt @@ -1,10 +1,12 @@ accesslog camelcase cpus +creds docstrings errmsg gthread gunicorn +héllo isort keepalive Meiczinger @@ -19,3 +21,4 @@ setuptools tesserocr tryceratops venv +wörld diff --git a/README.md b/README.md index 9e207de..7c6c621 100644 --- a/README.md +++ b/README.md @@ -1,99 +1,441 @@ -This is sample template to create a quick Python Falcon API application. It uses a schema defined in the resource for the route path and OpenAPI documentation. The base application will handle creating the routes and documentation automatically at runtime. +# Reliqua -You define a resource and the documentation will be auto-generated at startup using the docstrings. +A simple, efficient, and intuitive Python REST API framework built on [Falcon](https://falconframework.org/) and [Gunicorn](https://gunicorn.org/). Define resources with Sphinx-style docstrings and Reliqua handles routing, parameter validation, type coercion, and OpenAPI 3.1 documentation automatically. -```python +## Features + +- **Automatic route discovery** — Resource modules are auto-loaded from a directory +- **OpenAPI 3.1 generation** — API docs generated from docstrings at startup +- **Built-in Swagger UI** — Interactive API explorer served at `/docs` +- **Parameter validation & type coercion** — Declared parameter types are automatically enforced +- **Pluggable authentication & authorization** — Basic, Cookie, Header, Query, and Bearer auth with role-based access control +- **CORS support** — Enabled by default +- **Database helpers** — MySQL, PostgreSQL, and SQLite via Peewee ORM +- **Custom media handlers** — JSON (with datetime support), YAML, and plain text + +## Installation + +```bash +pip install reliqua +``` + +### Dependencies + +- `falcon>=3.0.0` +- `falcon-cors` +- `gunicorn>=19.6.0` +- `pyyaml` + +Optional for database support: `peewee` + +## Quick Start +### 1. Create a Resource + +Resources are classes that subclass `Resource`. Routes, parameters, and responses are all defined via class attributes and docstrings. + +```python from reliqua.resources.base import Resource -from reliqua import status_codes as status +from reliqua.exceptions import HTTPNotFound -class User(Resource): +users = [ + {"username": "ted", "email": "ted@nowhere.com"}, + {"username": "bob", "email": "bob@nowhere.com"}, +] - __routes__ = [ - '/users/{id}', - ] +USER = { + "type": "object", + "properties": { + "username": {"type": "string", "examples": ["billy"]} + }, + "required": ["username"], +} - phone = phone +USERS = {"type": "array", "items": {"$ref": "#/components/schemas/user"}} - def on_get(self, req, resp, id=None): - """ - Retrieve a user. This value - is awesome - :param str id: [in=path, required] User ID - :param str email: [in=query] User Email - :param str phone: [in=query enum] Phone Numbers +class User(Resource): + """User resource.""" + + __routes__ = { + "/users/{id}": {"suffix": "by_id"}, + } - :response 200: - :response 400: + __tags__ = ["users"] + __auth__ = {"GET": ["admin"]} + user = USER + + def on_get_by_id(self, req, resp, id=None): + """ + Return a user. + + :param str id: [in=path required] User ID + :response 200 user: User was retrieved + :response 404: User not found :return json: """ try: resp.media = users[int(id)] - except IndexError: - resp.status = status.HTTP_404 + except IndexError as exc: + raise HTTPNotFound("User not found") from exc - def on_delete(self, req, resp, id=None): - try: - users.pop(int(id)) - resp.media = {'success': True} - except IndexError - resp.status = status.HTTP_400 + +class Users(Resource): + """Users resource.""" + + __routes__ = {"/users": {}} + __tags__ = ["users"] + + users = USERS + + def on_get(self, req, resp): + """ + Return users. + + :param str username: [in=query] Username + :param str email: [in=query default=ted@nowhere.com] Email + + :response 200 users: Users were retrieved + :response 401: Invalid Authorization + :return json: + """ + resp.media = {"results": users} + + def on_post(self, req, resp): + """ + Create a new user. + + :param str username: [in=body required] Username + :param str email: [in=body required] Email + + :accepts json: + :return json: + """ + users.append(req.params) + resp.media = len(users) - 1 ``` -This will create the /users/{id} endpoint and corresponding API documentation. The OpenAPI documentation and swagger.json will be dynamically generate at application startup. The OpenAPI ui will be available at: +### 2. Create the Application +```python +from reliqua import Application + +app = Application( + resource_path="/path/to/resources", + info={ + "title": "My API", + "version": "1.0.0", + "description": "My awesome API", + }, + gunicorn={ + "bind": "127.0.0.1:8000", + "workers": 2, + }, +) +app.run() ``` -http:///docs/ + +The OpenAPI UI will be available at: + +``` +http://127.0.0.1:8000/docs ``` -and the swagger.json file located at: +The OpenAPI JSON spec will be available at: ``` -http:///docs/swagger.json +http://127.0.0.1:8000/openapi/openapi.json ``` -To create an application, you import the application template. +## Configuration + +The `Application` constructor accepts the following option groups: + +### `info` — API Metadata + +| Key | Default | Description | +|---|---|---| +| `title` | `"Application Title"` | API title shown in docs | +| `version` | `"1.0.0"` | API version | +| `description` | `""` | API description | +| `license` | `""` | License name | +| `license_url` | `""` | License URL | +| `contact_name` | `""` | Contact name | + +### `openapi` — Documentation Options + +| Key | Default | Description | +|---|---|---| +| `ui_url` | `None` | Public URL for the API (e.g., behind a proxy). Defaults to `http://{bind}` | +| `highlight` | `True` | Enable syntax highlighting in Swagger UI | +| `sort` | `"alpha"` | Sort order for operations | +| `path` | `"/openapi"` | Path for OpenAPI spec endpoint | +| `docs` | `"/docs"` | Path for Swagger UI | +| `servers` | `[]` | Additional server entries for OpenAPI spec | + +### `gunicorn` — Server Options + +| Key | Default | Description | +|---|---|---| +| `bind` | `"127.0.0.1:8000"` | Address and port to listen on | +| `workers` | `1` | Number of worker processes | +| `worker_class` | `"gthread"` | Gunicorn worker class | +| `timeout` | `30` | Worker timeout in seconds | +| `keepalive` | `2` | Keep-alive timeout | +| `loglevel` | `"critical"` | Log level | +| `accesslog` | `"-"` | Access log target | +| `errorlog` | `"-"` | Error log target | + +### Other Parameters + +| Parameter | Default | Description | +|---|---|---| +| `resource_path` | `""` | Directory containing resource modules | +| `middleware` | `[]` | List of middleware instances | +| `config` | `None` | App config dict accessible on resources as `self.app_config` | +| `resource_attributes` | `{}` | Key/value pairs set as attributes on every resource | + +## Resources + +### Routes + +Routes are defined as a dictionary on the `__routes__` class attribute. Keys are URL patterns and values are option dicts. ```python -from reliqua.app import Application +class MyResource(Resource): + __routes__ = { + "/items": {}, + "/items/{id}": {"suffix": "by_id"}, + } - -app = Application( - bind='0.0.0.0:8000', - ui_url = 'http://example.com/api', - workers=1, - resource_path='/var/www/html/resources' + def on_get(self, req, resp): + """Handle GET /items.""" + + def on_get_by_id(self, req, resp, id=None): + """Handle GET /items/{id}.""" +``` + +When a resource has multiple routes, the `suffix` option distinguishes handler methods (e.g., `on_get_by_id`, `on_delete_by_id`). + +### Docstring Parameters + +Parameters are declared using Sphinx-style docstrings: + +``` +:param : [] +``` + +**Supported types:** `str`, `int`, `float`, `bool`, `list`, `list[int]`, `list[str]`, `list[dict]`, `object`, `dict`, `json` + +**Options** (inside `[]`): + +| Option | Description | +|---|---| +| `in=query\|path\|body\|form` | Parameter location (default: `query`) | +| `required` | Mark as required (path params are always required) | +| `default=` | Default value | +| `enum=` | Resource attribute containing allowed values | +| `min=` | Minimum value | +| `max=` | Maximum value | + +### Responses + +``` +:response []: +``` + +The optional `` name references a class attribute containing an OpenAPI schema dict. The schema is automatically added to the OpenAPI `components/schemas` section. + +### Content Types + +``` +:return [json, yaml, xml]: Response content types +:accepts [json, xml]: Accepted request content types +``` + +Supported aliases: `json`, `yaml`, `xml`, `html`, `text`, `binary`, `gzip`, `form`, `jpeg`, `png`, `gif` + +### Resource Attributes + +| Attribute | Description | +|---|---| +| `__routes__` | Dict mapping URL patterns to route options | +| `__tags__` | List of OpenAPI tags (defaults to class name) | +| `__auth__` | Dict mapping HTTP methods to allowed roles | +| `__components__` | Dict of OpenAPI component schemas | +| `no_auth` | Set to `True` to bypass authentication | + +### Helper Methods + +The `Resource` base class provides `get_params(req, keys=None, exclude=None)` to extract request parameters. + +## Authentication & Authorization + +Reliqua separates authentication (identity verification) from authorization (access control). + +### Authentication + +```python +from reliqua.auth import ( + AuthenticationContext, + AuthMiddleware, + BasicAuthentication, + CookieAuthentication, + HeaderAuthentication, + QueryAuthentication, + MultiAuthentication, ) -app.run() + + +def validate_user(username, password): + if username == "admin" and password == "secret": + return AuthenticationContext(user="admin", role="admin") + return None + + +def validate_api_key(api_key): + if api_key == "abc123": + return AuthenticationContext(name="api_user", role="admin") + return None + + +# HTTP Basic auth +basic_auth = BasicAuthentication(validation=validate_user) + +# API key from cookie +cookie_auth = CookieAuthentication("api_key", validation=validate_api_key) + +# API key from header +header_auth = HeaderAuthentication("X-API-Key", validation=validate_api_key) + +# API key from query parameter +query_auth = QueryAuthentication("api_key", validation=validate_api_key) + +# Try multiple authenticators in order +multi_auth = MultiAuthentication([basic_auth, cookie_auth]) ``` +Validation callbacks receive credentials and must return an `AuthenticationContext` on success or a falsy value on failure. + +### Authorization + +Four access control strategies are available: + +**AccessResource** — Per-resource authorization via `__auth__` class attributes: + +```python +from reliqua.auth import AccessResource, AuthMiddleware + +control = AccessResource(default_mode="deny") +auth = AuthMiddleware([basic_auth], control=control) + +class SecureResource(Resource): + __auth__ = { + "GET": ["admin", "user"], + "POST": ["admin"], + } ``` -Where: -bind: Address and port to listen for requests. [host:port] -ui_url: The URL to the API when being used with a proxy, like nginx. If not supplied, - then the bind address is used. -workers: Number of worker threads to start. -resource_path: This is where your python resource files are located. +**AccessMap** — A centralized route-to-role mapping: + +```python +from reliqua.auth import AccessMap + +access_map = { + "/users": {"GET": ["admin", "user"], "POST": ["admin"]}, + "/public": {"*": ["*"]}, # wildcard: any role, any method +} +control = AccessMap(access_map) ``` -Refer to the example application for more examples. You can install the library and example application +**AccessList** — Simple allow/deny lists of routes and methods: +```python +from reliqua.auth import AccessList + +control = AccessList( + routes=["/secret"], + methods=["DELETE"], + default_mode="allow", # everything open except listed routes/methods +) ``` -python setup.py install + +**AccessCallback** — Delegate to custom functions: + +```python +from reliqua.auth import AccessCallback + +control = AccessCallback( + authenticate_callback=my_auth_check, + authorized_callback=my_role_check, +) +``` + +### Applying Auth Middleware + +```python +auth_middleware = AuthMiddleware([basic_auth, cookie_auth], control=control) + +app = Application( + resource_path="/path/to/resources", + middleware=[auth_middleware], +) ``` -You can execute the example application +Resources with `no_auth = True` bypass authentication entirely. The authenticated context is available in handlers via `req.context["authentication"]`. +## Database Support + +Database helpers use the [Peewee ORM](http://docs.peewee-orm.com/) with a proxy pattern. + +```python +from reliqua.database import BaseModel, mysql_connect, DatabaseConnection +import peewee + +# Connect to the database +db = mysql_connect("localhost", "mydb", "user", "password", 3306) + +# Define a model +class UserModel(BaseModel): + username = peewee.CharField() + email = peewee.CharField() + +# Add DatabaseConnection middleware to manage connections per request +app = Application( + resource_path="/path/to/resources", + middleware=[DatabaseConnection()], +) ``` -$ reliqua-example + +Available connection helpers: `mysql_connect()`, `psql_connect()`, `sqlite_connect()` + +## Media Handlers + +Built-in handlers are registered automatically: + +| Content Type | Handler | Notes | +|---|---|---| +| `application/json` | `JSONHandler` | Serializes `datetime`, `date`, and `time` objects | +| `application/yaml` | `YAMLHandler` | YAML via PyYAML | +| `text/html` | `TextHandler` | Plain text | +| `text/plain` | `TextHandler` | Plain text | + +## Example Application + +A full example application is included in the `example/` directory demonstrating resources, authentication, form handling, and binary media streaming. + +```bash +# Run the example +python -m example + +# Custom bind address and port +python -m example --address 0.0.0.0 --port 9000 ``` -From here the openapi-ui will be available at +The Swagger UI will be available at `http://127.0.0.1:8000/docs`. + +## License -```` -http://localhost:8000/docs/ -```` +MIT diff --git a/example/__main__.py b/example/__main__.py index 1ead7d4..e8cc020 100644 --- a/example/__main__.py +++ b/example/__main__.py @@ -36,7 +36,7 @@ def check_api_key(api_key): def main(): """Execute main method.""" - bind_address = "127.0.0.01" + bind_address = "127.0.0.1" bind_port = 8000 workers = 2 parser = argparse.ArgumentParser() @@ -50,7 +50,7 @@ def main(): parser.add_argument("--port", help="Bind port to listen for requests", default=bind_port) parser.add_argument("--ui-url", help="OpenAPI UI URL (ie Swagger index) default is address:port") parser.add_argument("--docs", help="Docs", default="/docs") - parser.add_argument("--server", help="Additional server(s) usable in the API docs", nargs="*", action="append") + parser.add_argument("--server", help="Additional server(s) usable in the API docs", nargs="*") parser.add_argument("--resource-path", help="Path to API resource modules", default=resource_path) parser.add_argument("--workers", help="Number of worker threads", default=workers) parser.add_argument("--config", help="Configuration file", default=None) @@ -100,14 +100,12 @@ def main(): app = Application( resource_path=args.resource_path, - loglevel="info", - accesslog=None, middleware=middleware, config=vars(args), resource_attributes={"random": "example"}, info=info, openapi=openapi, - gunicorn_options=gunicorn, + gunicorn=gunicorn, ) app.run() diff --git a/example/resources/media.py b/example/resources/media.py index ee77565..82c611c 100644 --- a/example/resources/media.py +++ b/example/resources/media.py @@ -24,11 +24,11 @@ def on_get(self, _req, resp): :response 200 binary: All good :return gzip: Return data """ - fh = open("/tmp/hello.txt.gz", "rb") - resp.append_header("Content-Disposition", "attachment; filename=hello.txt.gz") - resp.content_type = "application/gzip" - resp.content_encoding = "gzip" - resp.data = fh.read() + with open("/tmp/hello.txt.gz", "rb") as fh: + resp.append_header("Content-Disposition", "attachment; filename=hello.txt.gz") + resp.content_type = "application/gzip" + resp.content_encoding = "gzip" + resp.data = fh.read() class Binary(Resource): diff --git a/pyproject.toml b/pyproject.toml index 922b15a..173c940 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -68,6 +68,7 @@ add-ignore = "D412" [tool.pylint."messages control"] disable = [ + "C0116", "too-many-instance-attributes", "too-many-arguments", "too-few-public-methods", @@ -160,6 +161,9 @@ ignore = [ "TRY400", # tryceratops ] +[tool.ruff.lint.per-file-ignores] +"**tests/**" = ["S101", "D102", "D103", "D104", "D105", "D107", "D400", "DTZ001"] + [tool.ruff.lint.pydocstyle] convention = "pep257" diff --git a/src/reliqua/auth.py b/src/reliqua/auth.py index 2969cf5..8cd0fde 100644 --- a/src/reliqua/auth.py +++ b/src/reliqua/auth.py @@ -133,8 +133,8 @@ def __init__(self, routes=None, methods=None, default_mode="allow"): :param list methods: List of methods to apply rules :param str default_mode: Default mode (allow|deny) """ - self.routes = [x.lower() for x in routes] - self.methods = [x.lower() for x in methods] + self.routes = [x.lower() for x in (routes or [])] + self.methods = [x.lower() for x in (methods or [])] self.default_mode = default_mode def authorized(self, context, _route, _method, _resource): @@ -161,10 +161,10 @@ def authentication_required(self, route, method, _resource): # check if authentication is required if self.default_mode == "allow": # if default mode is allow, then a match means auth required - required = not matched + required = matched else: # default mode is deny, then a match means auth not required - required = matched + required = not matched return required @@ -213,8 +213,10 @@ def authorized(self, context, route, method, _resource): :param str method: HTTP method being invoked :return bool: True if authorized """ - route = self.access_map.get(route) - roles = route.get("*") or route.get(method) + entry = self.access_map.get(route) + if not entry: + return False + roles = entry.get("*") or entry.get(method) or [] if context.role in roles or "*" in roles: return True @@ -229,10 +231,12 @@ def authentication_required(self, route, method, _resource): :param str route: The route being called :param str method: The http method invoked - :return bool: True if e + :return bool: True if authentication is required """ - route = self.access_map.get(route) - roles = route.get("*") or route.get(method) + entry = self.access_map.get(route) + if not entry: + return True + roles = entry.get("*") or entry.get(method) # if method is specified and has a wildcard role # then authentication is not required @@ -289,13 +293,13 @@ def authorized(self, context, _route, method, resource): :return bool: True if authorized """ auth = getattr(resource, "__auth__", {}) - roles = auth.get("*", []) or auth.get(method, []) + roles = auth.get("*", []) or auth.get(method.upper(), []) or auth.get(method.lower(), []) # if no roles are defined skip authorization # or raise an exception if not roles: if self.raise_on_undefined: - raise falcon.HTTPNotImplemented(f"no roles are defined for {resource.name} {method}") + raise falcon.HTTPNotImplemented(description=f"no roles are defined for {resource.name} {method}") return True diff --git a/src/reliqua/database.py b/src/reliqua/database.py index d87b3da..08a03e1 100644 --- a/src/reliqua/database.py +++ b/src/reliqua/database.py @@ -27,6 +27,8 @@ def is_b64(string): :return: True if string is base64 encoded """ try: + if isinstance(string, str): + string = string.encode("utf-8") return b64encode(b64decode(string)) == string except (ValueError, TypeError, binascii.Error): return False diff --git a/src/reliqua/middleware.py b/src/reliqua/middleware.py index 9ae7e30..3089610 100644 --- a/src/reliqua/middleware.py +++ b/src/reliqua/middleware.py @@ -4,6 +4,7 @@ Copyright 2016-2024. """ +import builtins import json import re @@ -51,7 +52,7 @@ def to_bool(value): def python_type(s): """Return Python type from string.""" - return __builtins__.get(s) + return getattr(builtins, s, None) class Parameter: @@ -185,7 +186,7 @@ def as_array(req, name, default=None, required=False, transform=None, **_kwargs) :param transform: Function to transform the list items :return: Converted parameter value """ - Converter.as_list(req, name, default=default, required=required, transform=transform) + return Converter.as_list(req, name, default=default, required=required, transform=transform) @staticmethod def convert(req, parameter, transform=None): @@ -236,8 +237,8 @@ def _check_required(self, request, parameter): :return: None :raises falcon.HTTPBadRequest: Falcon bad request exception """ - value = parameter.default or request.params.get(parameter.name) - if parameter.required and not value: + value = parameter.default if parameter.default is not None else request.params.get(parameter.name) + if parameter.required and value is None: raise falcon.HTTPBadRequest( title="Bad Request", description=f"Missing parameter '{parameter.name}'", diff --git a/src/reliqua/openapi.py b/src/reliqua/openapi.py index 38b3f23..98d176a 100644 --- a/src/reliqua/openapi.py +++ b/src/reliqua/openapi.py @@ -245,7 +245,7 @@ def __init__(self, code=None, description=None, content=None, schema=None): self.code = code self.description = description self.content = content - self.schema = schema if schema else "default_response" + self.schema = schema or "default_response" def __repr__(self): """Return a printable representational string.""" @@ -319,7 +319,7 @@ def accepts(self): def binary_body(self): """Return binary body.""" - accepts = CONTENT_MAP.get(self.accepts) + accepts = CONTENT_MAP.get(self.accepts[0]) if self.accepts else "application/octet-stream" return { "content": { accepts: { @@ -355,7 +355,7 @@ def body(self): def request_body(self): """Return request body.""" - if self.accepts in BINARY_TYPES: + if any(a in BINARY_TYPES for a in self.accepts): return self.binary_body() return self.body() diff --git a/src/reliqua/status_codes.py b/src/reliqua/status_codes.py index 07c34e8..c535db1 100644 --- a/src/reliqua/status_codes.py +++ b/src/reliqua/status_codes.py @@ -92,4 +92,4 @@ def http(code): "511": "Network Authentication Required", } -MESSAGES = {(v.upper().replace(" ", "_"), k) for k, v in CODES.items()} +MESSAGES = {v.upper().replace(" ", "_"): k for k, v in CODES.items()} diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..e304e45 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,92 @@ +"""Tests for reliqua.app utility functions.""" + +import os +import tempfile + +from reliqua.app import load_config, update_dict + + +class TestUpdateDict: + """Tests for the update_dict function.""" + + def test_keeps_only_common_keys_from_a(self): + a = {"x": 1, "y": 2, "z": 3} + b = {"x": 10, "y": 20} + result = update_dict(a, b) + assert result == {"x": 1, "y": 2} + + def test_adds_keys_from_b_not_in_a(self): + a = {"x": 1} + b = {"x": 10, "y": 20, "z": 30} + result = update_dict(a, b) + assert result == {"x": 1, "y": 20, "z": 30} + + def test_empty_a(self): + a = {} + b = {"x": 1, "y": 2} + result = update_dict(a, b) + assert result == {"x": 1, "y": 2} + + def test_empty_b(self): + a = {"x": 1, "y": 2} + b = {} + result = update_dict(a, b) + assert result == {} + + def test_both_empty(self): + result = update_dict({}, {}) + assert result == {} + + def test_no_overlap(self): + a = {"x": 1} + b = {"y": 2} + result = update_dict(a, b) + assert result == {"y": 2} + + def test_does_not_mutate_inputs(self): + a = {"x": 1, "y": 2} + b = {"x": 10} + a_copy = a.copy() + b_copy = b.copy() + update_dict(a, b) + assert a == a_copy + assert b == b_copy + + +class TestLoadConfig: + """Tests for the load_config function.""" + + def test_load_valid_config(self): + config_content = "[config]\ntitle = My App\nversion = 2.0.0\n" + with tempfile.NamedTemporaryFile(mode="w", suffix=".ini", delete=False) as f: + f.write(config_content) + f.flush() + result = load_config(f.name) + os.unlink(f.name) + assert result["title"] == "My App" + assert result["version"] == "2.0.0" + + def test_load_nonexistent_file(self): + result = load_config("/nonexistent/config.ini") + assert not result + + def test_load_none(self): + result = load_config(None) + assert not result + + def test_load_empty_config(self): + with tempfile.NamedTemporaryFile(mode="w", suffix=".ini", delete=False) as f: + f.write("") + f.flush() + result = load_config(f.name) + os.unlink(f.name) + assert not result + + def test_load_config_missing_section(self): + config_content = "[other]\nkey = value\n" + with tempfile.NamedTemporaryFile(mode="w", suffix=".ini", delete=False) as f: + f.write(config_content) + f.flush() + result = load_config(f.name) + os.unlink(f.name) + assert result == {} diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..e9df57f --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,345 @@ +"""Tests for reliqua.auth module.""" + +import base64 +from unittest.mock import MagicMock + +import falcon +import pytest + +from reliqua.auth import ( + AccessCallback, + AccessList, + AccessMap, + AccessResource, + AuthenticationContext, + AuthMiddleware, + BasicAuthentication, + MultiAuthentication, + is_base64, +) + + +class TestIsBase64: + """Tests for is_base64 helper.""" + + def test_valid_base64(self): + assert is_base64("aGVsbG8=") is True + + def test_invalid_base64(self): + assert is_base64("not-base64!!!") is False + + def test_empty_string(self): + assert is_base64("") is True + + +class TestAuthenticationContext: + """Tests for AuthenticationContext.""" + + def test_default_role_is_none(self): + ctx = AuthenticationContext() + assert ctx.role is None + + def test_kwargs_set_attributes(self): + ctx = AuthenticationContext(user="alice", role="admin") + assert ctx.user == "alice" + assert ctx.role == "admin" + + +class TestAccessList: + """Tests for AccessList access control.""" + + def test_init_with_none_defaults(self): + """Should not crash when routes/methods are None.""" + ac = AccessList() + assert ac.routes == [] + assert ac.methods == [] + + def test_init_with_values(self): + ac = AccessList(routes=["/users", "/Admin"], methods=["GET", "POST"]) + assert ac.routes == ["/users", "/admin"] + assert ac.methods == ["get", "post"] + + def test_allow_mode_matched_route_requires_auth(self): + """In allow mode, listed routes require authentication.""" + ac = AccessList(routes=["/secret"], methods=[], default_mode="allow") + assert ac.authentication_required("/secret", "GET", None) is True + + def test_allow_mode_unmatched_route_no_auth(self): + """In allow mode, unlisted routes do NOT require authentication.""" + ac = AccessList(routes=["/secret"], methods=[], default_mode="allow") + assert ac.authentication_required("/public", "GET", None) is False + + def test_allow_mode_matched_method_requires_auth(self): + """In allow mode, listed methods require authentication.""" + ac = AccessList(routes=[], methods=["post"], default_mode="allow") + assert ac.authentication_required("/any", "POST", None) is True + + def test_deny_mode_matched_route_no_auth(self): + """In deny mode, listed routes are exempt from auth.""" + ac = AccessList(routes=["/public"], methods=[], default_mode="deny") + assert ac.authentication_required("/public", "GET", None) is False + + def test_deny_mode_unmatched_route_requires_auth(self): + """In deny mode, unlisted routes require authentication.""" + ac = AccessList(routes=["/public"], methods=[], default_mode="deny") + assert ac.authentication_required("/secret", "GET", None) is True + + def test_authorized_not_implemented(self): + ac = AccessList() + with pytest.raises(NotImplementedError): + ac.authorized(None, None, None, None) + + +class TestAccessCallback: + """Tests for AccessCallback access control.""" + + def test_authorized_delegates_to_callback(self): + callback = MagicMock(return_value=True) + ac = AccessCallback(authenticate_callback=MagicMock(), authorized_callback=callback) + ctx = AuthenticationContext(role="admin") + assert ac.authorized(ctx, "/users", "GET", None) is True + callback.assert_called_once_with(ctx, "/users", "GET") + + def test_authentication_required_delegates_to_callback(self): + callback = MagicMock(return_value=True) + ac = AccessCallback(authenticate_callback=callback, authorized_callback=MagicMock()) + assert ac.authentication_required("/users", "GET", None) is True + callback.assert_called_once_with(route="/users", method="GET") + + +class TestAccessMap: + """Tests for AccessMap access control.""" + + ac = None + access_map = None + + def setup_method(self): + self.access_map = { + "/users": { + "GET": ["admin", "user"], + "POST": ["admin"], + }, + "/public": { + "*": ["*"], + }, + } + self.ac = AccessMap(self.access_map) + + def test_authorized_with_valid_role(self): + ctx = AuthenticationContext(role="admin") + assert self.ac.authorized(ctx, "/users", "GET", None) is True + + def test_authorized_with_invalid_role(self): + ctx = AuthenticationContext(role="guest") + assert self.ac.authorized(ctx, "/users", "GET", None) is False + + def test_authorized_wildcard_role(self): + ctx = AuthenticationContext(role="anyone") + assert self.ac.authorized(ctx, "/public", "GET", None) is True + + def test_authorized_missing_route_returns_false(self): + """Missing routes should return False, not crash.""" + ctx = AuthenticationContext(role="admin") + assert self.ac.authorized(ctx, "/nonexistent", "GET", None) is False + + def test_authentication_required_normal_route(self): + assert self.ac.authentication_required("/users", "GET", None) is True + + def test_authentication_not_required_wildcard(self): + assert self.ac.authentication_required("/public", "GET", None) is False + + def test_authentication_required_missing_route(self): + """Missing routes should require authentication, not crash.""" + assert self.ac.authentication_required("/nonexistent", "GET", None) is True + + +class TestAccessResource: + """Tests for AccessResource access control.""" + + def _make_resource(self, auth_dict=None): + resource = MagicMock() + resource.name = "TestResource" + if auth_dict is not None: + resource.__auth__ = auth_dict + else: + del resource.__auth__ + return resource + + def test_authorized_with_matching_role(self): + ac = AccessResource() + resource = self._make_resource({"GET": ["admin"]}) + ctx = AuthenticationContext(role="admin") + assert ac.authorized(ctx, "/test", "GET", resource) is True + + def test_authorized_with_wrong_role(self): + ac = AccessResource() + resource = self._make_resource({"GET": ["admin"]}) + ctx = AuthenticationContext(role="guest") + assert ac.authorized(ctx, "/test", "GET", resource) is False + + def test_authorized_wildcard_method(self): + ac = AccessResource() + resource = self._make_resource({"*": ["admin"]}) + ctx = AuthenticationContext(role="admin") + assert ac.authorized(ctx, "/test", "GET", resource) is True + + def test_authorized_no_auth_defined_allows(self): + ac = AccessResource() + resource = self._make_resource({}) + ctx = AuthenticationContext(role="admin") + assert ac.authorized(ctx, "/test", "GET", resource) is True + + def test_authorized_no_auth_raises_when_configured(self): + ac = AccessResource(raise_on_undefined=True) + resource = self._make_resource({}) + ctx = AuthenticationContext(role="admin") + with pytest.raises(falcon.HTTPNotImplemented): + ac.authorized(ctx, "/test", "GET", resource) + + def test_authorized_case_insensitive_method(self): + """Authorized should find roles regardless of method case.""" + ac = AccessResource() + resource = self._make_resource({"GET": ["admin"]}) + ctx = AuthenticationContext(role="admin") + # method passed as lowercase should match "GET" key + assert ac.authorized(ctx, "/test", "get", resource) is True + + def test_authentication_required_deny_mode_with_roles(self): + ac = AccessResource(default_mode="deny") + resource = self._make_resource({"GET": ["admin"]}) + assert ac.authentication_required("/test", "GET", resource) is True + + def test_authentication_required_allow_mode_no_roles(self): + ac = AccessResource(default_mode="allow") + resource = self._make_resource({}) + assert ac.authentication_required("/test", "GET", resource) is False + + def test_authentication_required_deny_mode_no_roles(self): + ac = AccessResource(default_mode="deny") + resource = self._make_resource({}) + assert ac.authentication_required("/test", "GET", resource) is True + + +class TestBasicAuthentication: + """Tests for BasicAuthentication.""" + + def test_authenticate_success(self): + auth = BasicAuthentication(validation=lambda u, p: AuthenticationContext(user=u, role="admin")) + token = base64.b64encode(b"alice:secret").decode("utf-8") + req = MagicMock() + req.get_header.return_value = f"Basic {token}" + result = auth.authenticate(req, None, None) + assert result.user == "alice" + + def test_authenticate_missing_header_raises(self): + auth = BasicAuthentication(validation=lambda u, p: None) + req = MagicMock() + req.get_header.return_value = None + with pytest.raises(falcon.HTTPUnauthorized): + auth.authenticate(req, None, None) + + def test_authenticate_invalid_credentials_raises(self): + auth = BasicAuthentication(validation=lambda u, p: None) + token = base64.b64encode(b"bad:creds").decode("utf-8") + req = MagicMock() + req.get_header.return_value = f"Basic {token}" + with pytest.raises(falcon.HTTPUnauthorized): + auth.authenticate(req, None, None) + + def test_dict_returns_openapi_schema(self): + auth = BasicAuthentication(validation=lambda u, p: None) + schema = auth.dict() + assert "BasicAuthentication" in schema + assert schema["BasicAuthentication"]["type"] == "http" + assert schema["BasicAuthentication"]["scheme"] == "basic" + + +class TestMultiAuthentication: + """Tests for MultiAuthentication.""" + + def test_first_authenticator_succeeds(self): + auth1 = MagicMock() + auth1.authenticate.return_value = AuthenticationContext(role="admin") + auth2 = MagicMock() + multi = MultiAuthentication([auth1, auth2]) + result = multi.authenticate(None, None, None) + assert result.role == "admin" + auth2.authenticate.assert_not_called() + + def test_falls_through_on_failure(self): + auth1 = MagicMock() + auth1.authenticate.side_effect = falcon.HTTPUnauthorized(description="fail") + auth2 = MagicMock() + auth2.authenticate.return_value = AuthenticationContext(role="user") + multi = MultiAuthentication([auth1, auth2]) + result = multi.authenticate(None, None, None) + assert result.role == "user" + + def test_all_fail_raises(self): + auth1 = MagicMock() + auth1.authenticate.side_effect = falcon.HTTPUnauthorized(description="fail") + auth2 = MagicMock() + auth2.authenticate.side_effect = falcon.HTTPUnauthorized(description="fail") + multi = MultiAuthentication([auth1, auth2]) + with pytest.raises(falcon.HTTPUnauthorized): + multi.authenticate(None, None, None) + + def test_dict_merges_schemas(self): + auth1 = MagicMock() + auth1.dict.return_value = {"BasicAuth": {"type": "http"}} + auth2 = MagicMock() + auth2.dict.return_value = {"ApiKey": {"type": "apiKey"}} + multi = MultiAuthentication([auth1, auth2]) + schema = multi.dict() + assert "BasicAuth" in schema + assert "ApiKey" in schema + + +class TestAuthMiddleware: + """Tests for AuthMiddleware.""" + + def test_skips_no_auth_resource(self): + """Resources with no_auth=True should never be authenticated.""" + auth = MagicMock() + control = MagicMock() + mw = AuthMiddleware([auth], control=control) + + resource = MagicMock() + resource.no_auth = True + + mw.process_resource(MagicMock(), MagicMock(), resource, {}) + control.authentication_required.assert_not_called() + + def test_skips_when_auth_not_required(self): + auth = MagicMock() + control = MagicMock() + control.authentication_required.return_value = False + mw = AuthMiddleware([auth], control=control) + + resource = MagicMock(spec=[]) # no no_auth attribute + req = MagicMock() + req.uri_template = "/public" + req.method = "GET" + + mw.process_resource(req, MagicMock(), resource, {}) + auth.authenticate.assert_not_called() + + def test_unauthorized_role_raises(self): + auth_obj = MagicMock() + ctx = AuthenticationContext(role="guest") + auth_obj.authenticate.return_value = ctx + + control = MagicMock() + control.authentication_required.return_value = True + control.authorized.return_value = False + + mw = AuthMiddleware([auth_obj], control=control) + + resource = MagicMock(spec=[]) + req = MagicMock() + req.uri_template = "/secret" + req.method = "DELETE" + req.context = {} + + with pytest.raises(falcon.HTTPUnauthorized): + mw.process_resource(req, MagicMock(), resource, {}) diff --git a/tests/test_database.py b/tests/test_database.py new file mode 100644 index 0000000..32ab99f --- /dev/null +++ b/tests/test_database.py @@ -0,0 +1,41 @@ +"""Tests for reliqua.database module.""" + +from base64 import b64encode + +from reliqua.database import is_b64 + + +class TestIsB64: + """Tests for the is_b64 helper.""" + + def test_valid_base64_string(self): + """A valid base64 str should return True.""" + encoded = b64encode(b"hello world").decode("utf-8") + assert is_b64(encoded) is True + + def test_valid_base64_bytes(self): + """A valid base64 bytes should return True.""" + encoded = b64encode(b"hello world") + assert is_b64(encoded) is True + + def test_invalid_base64_string(self): + """A non-base64 string should return False.""" + assert is_b64("not-base64!!!") is False + + def test_empty_string(self): + """An empty string is valid base64.""" + assert is_b64("") is True + + def test_none_returns_false(self): + """None should return False.""" + assert is_b64(None) is False + + def test_plain_text_not_base64(self): + """Arbitrary plain text that isn't valid base64 returns False.""" + assert is_b64("hello world") is False + + def test_password_roundtrip(self): + """A base64-encoded password should be detectable.""" + password = "s3cr3t_p@ssw0rd" + encoded = b64encode(password.encode("utf-8")).decode("utf-8") + assert is_b64(encoded) is True diff --git a/tests/test_media_handlers.py b/tests/test_media_handlers.py new file mode 100644 index 0000000..a7695e4 --- /dev/null +++ b/tests/test_media_handlers.py @@ -0,0 +1,87 @@ +"""Tests for reliqua.media_handlers module.""" + +import io +import json +from datetime import date, datetime, time + +from reliqua.media_handlers import JSONHandler, TextHandler, YAMLHandler + + +class TestJSONHandler: + """Tests for the JSONHandler.""" + + def test_serialize_basic(self): + handler = JSONHandler() + data = {"name": "alice", "age": 30} + result = handler.serialize(data, "application/json") + parsed = json.loads(result) + assert parsed == data + + def test_serialize_datetime(self): + handler = JSONHandler() + dt = datetime(2024, 1, 15, 10, 30, 0) + data = {"created": dt} + result = handler.serialize(data, "application/json") + parsed = json.loads(result) + assert parsed["created"] == str(dt) + + def test_serialize_date(self): + handler = JSONHandler() + d = date(2024, 1, 15) + data = {"date": d} + result = handler.serialize(data, "application/json") + parsed = json.loads(result) + assert parsed["date"] == str(d) + + def test_serialize_time(self): + handler = JSONHandler() + t = time(10, 30, 0) + data = {"time": t} + result = handler.serialize(data, "application/json") + parsed = json.loads(result) + assert parsed["time"] == str(t) + + def test_serialize_unicode(self): + handler = JSONHandler() + data = {"name": "héllo wörld"} + result = handler.serialize(data, "application/json") + # ensure_ascii=False means unicode chars are preserved + assert "héllo" in result.decode("utf-8") if isinstance(result, bytes) else "héllo" in result + + +class TestTextHandler: + """Tests for the TextHandler.""" + + def test_serialize_string(self): + handler = TextHandler() + result = handler.serialize("hello world", "text/plain") + assert result == b"hello world" + + def test_serialize_number(self): + handler = TextHandler() + result = handler.serialize(42, "text/plain") + assert result == b"42" + + def test_deserialize(self): + handler = TextHandler() + stream = io.BytesIO(b"hello world") + result = handler.deserialize(stream, "text/plain", len(b"hello world")) + assert result == "hello world" + + +class TestYAMLHandler: + """Tests for the YAMLHandler.""" + + def test_serialize(self): + handler = YAMLHandler() + data = {"name": "alice", "age": 30} + result = handler.serialize(data, "application/yaml") + assert b"name: alice" in result + + def test_deserialize(self): + handler = YAMLHandler() + yaml_content = b"name: alice\nage: 30\n" + stream = io.BytesIO(yaml_content) + result = handler.deserialize(stream, "application/yaml", len(yaml_content)) + assert result["name"] == "alice" + assert result["age"] == 30 diff --git a/tests/test_middleware.py b/tests/test_middleware.py new file mode 100644 index 0000000..c67d4a6 --- /dev/null +++ b/tests/test_middleware.py @@ -0,0 +1,246 @@ +"""Tests for reliqua.middleware module.""" + +from unittest.mock import MagicMock + +import falcon +import pytest + +from reliqua.middleware import Converter, Parameter, ProcessParams, python_type, to_bool + + +class TestToBool: + """Tests for the to_bool helper.""" + + def test_truthy_strings(self): + assert to_bool("true") is True + assert to_bool("yes") is True + assert to_bool("1") is True + assert to_bool("on") is True + assert to_bool("t") is True + assert to_bool("y") is True + + def test_falsy_strings(self): + assert to_bool("false") is False + assert to_bool("no") is False + assert to_bool("0") is False + assert to_bool("off") is False + assert to_bool("f") is False + assert to_bool("n") is False + assert to_bool("") is False + + def test_case_insensitive(self): + assert to_bool("TRUE") is True + assert to_bool("False") is False + assert to_bool("YES") is True + + def test_whitespace_handling(self): + assert to_bool(" true ") is True + assert to_bool(" false ") is False + + def test_non_string_truthy(self): + assert to_bool(1) is True + assert to_bool([1]) is True + + def test_non_string_falsy(self): + assert to_bool(0) is False + + +class TestPythonType: + """Tests for the python_type helper.""" + + def test_returns_builtin_types(self): + assert python_type("str") is str + assert python_type("int") is int + assert python_type("float") is float + assert python_type("list") is list + assert python_type("dict") is dict + assert python_type("bool") is bool + + def test_returns_none_for_unknown(self): + assert python_type("nonexistent_type") is None + + def test_returns_none_for_empty_string(self): + assert python_type("") is None + + +class TestParameter: + """Tests for the Parameter class.""" + + def test_default_attributes(self): + p = Parameter() + assert p.default is None + assert p.required is False + + def test_kwargs_update(self): + p = Parameter(name="test", datatype="str", required=True, default="hello") + assert p.name == "test" + assert p.datatype == "str" + assert p.required is True + assert p.default == "hello" + + +class TestConverter: + """Tests for the Converter class.""" + + def _make_request(self, params=None): + """Create a mock Falcon request with params.""" + req = MagicMock(spec=falcon.Request) + req.params = params or {} + req.get_param = MagicMock(side_effect=lambda name, **kw: req.params.get(name, kw.get("default"))) + req.get_param_as_int = MagicMock( + side_effect=lambda name, **kw: int(req.params[name]) if name in req.params else kw.get("default") + ) + req.get_param_as_float = MagicMock( + side_effect=lambda name, **kw: float(req.params[name]) if name in req.params else kw.get("default") + ) + req.get_param_as_bool = MagicMock( + side_effect=lambda name, **kw: bool(req.params[name]) if name in req.params else kw.get("default") + ) + return req + + def test_as_str(self): + req = self._make_request({"name": "alice"}) + result = Converter.as_str(req, "name") + assert result == "alice" + + def test_as_int(self): + req = self._make_request({"count": "42"}) + result = Converter.as_int(req, "count") + assert result == 42 + + def test_as_float(self): + req = self._make_request({"rate": "3.14"}) + result = Converter.as_float(req, "rate") + assert result == pytest.approx(3.14) + + def test_as_bool(self): + req = self._make_request({"active": True}) + result = Converter.as_bool(req, "active") + assert result is True + + def test_as_int_default(self): + req = self._make_request({}) + result = Converter.as_int(req, "missing", default=10) + assert result == 10 + + def test_as_list_splits_csv_string(self): + req = self._make_request({"items": "a, b, c"}) + req.get_param_as_list = MagicMock(return_value=["a", "b", "c"]) + result = Converter.as_list(req, "items") + assert result == ["a", "b", "c"] + + def test_as_list_filters_empty_strings(self): + req = self._make_request({"items": ["a", "", "c"]}) + req.get_param_as_list = MagicMock(return_value=["a", "", "c"]) + result = Converter.as_list(req, "items") + assert result == ["a", "c"] + + def test_as_array_returns_value(self): + """as_array should return the result (not None).""" + req = self._make_request({"items": ["a", "b"]}) + req.get_param_as_list = MagicMock(return_value=["a", "b"]) + result = Converter.as_array(req, "items") + assert result == ["a", "b"] + + +class TestProcessParams: + """Tests for the ProcessParams middleware.""" + + def test_check_required_raises_on_missing(self): + pp = ProcessParams() + req = MagicMock() + req.params = {} + param = Parameter(name="username", required=True, default=None) + with pytest.raises(falcon.HTTPBadRequest): + pp._check_required(req, param) + + def test_check_required_passes_with_value(self): + pp = ProcessParams() + req = MagicMock() + req.params = {"username": "alice"} + param = Parameter(name="username", required=True, default=None) + pp._check_required(req, param) # should not raise + + def test_check_required_passes_with_default(self): + pp = ProcessParams() + req = MagicMock() + req.params = {} + param = Parameter(name="username", required=True, default="guest") + pp._check_required(req, param) # should not raise + + def test_check_required_passes_with_empty_string(self): + """An empty string is a valid provided value, should not raise.""" + pp = ProcessParams() + req = MagicMock() + req.params = {"tags": ""} + param = Parameter(name="tags", required=True, default=None) + pp._check_required(req, param) # should not raise + + def test_check_required_passes_with_empty_list(self): + """An empty list is a valid provided value, should not raise.""" + pp = ProcessParams() + req = MagicMock() + req.params = {"items": []} + param = Parameter(name="items", required=True, default=None) + pp._check_required(req, param) # should not raise + + def test_check_required_passes_with_empty_dict(self): + """An empty dict is a valid provided value, should not raise.""" + pp = ProcessParams() + req = MagicMock() + req.params = {"config": {}} + param = Parameter(name="config", required=True, default=None) + pp._check_required(req, param) # should not raise + + def test_check_required_passes_with_zero(self): + """Zero is a valid provided value, should not raise.""" + pp = ProcessParams() + req = MagicMock() + req.params = {"count": 0} + param = Parameter(name="count", required=True, default=None) + pp._check_required(req, param) # should not raise + + def test_check_required_passes_with_false(self): + """False is a valid provided value, should not raise.""" + pp = ProcessParams() + req = MagicMock() + req.params = {"active": False} + param = Parameter(name="active", required=True, default=None) + pp._check_required(req, param) # should not raise + + def test_parse_operators(self): + pp = ProcessParams() + req = MagicMock() + req.params = {"name__in": ["a", "b"], "age__between": [10, 20]} + operators = pp._parse_operators(req) + assert operators["name"] == "in" + assert operators["age"] == "between" + assert "name" in req.params + assert "age" in req.params + + def test_get_resource_parameters_returns_empty_on_missing(self): + req = MagicMock() + req.uri_template = "/missing" + req.method = "GET" + resource = MagicMock() + resource.__data__ = {} + result = ProcessParams.get_resource_parameters(req, resource) + assert result == [] + + def test_get_resource_parameters_returns_parameters(self): + req = MagicMock() + req.uri_template = "/users" + req.method = "get" + resource = MagicMock() + resource.__data__ = { + "/users": { + "get": { + "parameters": [ + {"name": "username", "datatype": "str", "required": False}, + ] + } + } + } + result = ProcessParams.get_resource_parameters(req, resource) + assert len(result) == 1 + assert result[0].name == "username" diff --git a/tests/test_openapi.py b/tests/test_openapi.py new file mode 100644 index 0000000..d7ee531 --- /dev/null +++ b/tests/test_openapi.py @@ -0,0 +1,268 @@ +"""Tests for reliqua.openapi module.""" + +from reliqua.openapi import ( + BINARY_TYPES, + CONTENT_MAP, + TYPE_MAP, + Contact, + License, + Operation, + Parameter, + Response, + camelcase, +) + + +class TestCamelcase: + """Tests for the camelcase helper.""" + + def test_single_word(self): + assert camelcase("hello") == "hello" + + def test_two_words(self): + assert camelcase("hello_world") == "helloWorld" + + def test_three_words(self): + assert camelcase("one_two_three") == "oneTwoThree" + + def test_empty_string(self): + assert camelcase("") == "" + + +class TestParameter: + """Tests for the OpenAPI Parameter class.""" + + def test_basic_string_parameter(self): + p = Parameter(name="username", datatype="str", location="query") + assert p.datatype == "string" + assert p.name == "username" + assert p.location == "query" + + def test_path_parameter_always_required(self): + p = Parameter(name="id", datatype="int", location="path", required=False) + assert p.required is True + + def test_list_datatype(self): + p = Parameter(name="ids", datatype="list[int]") + assert p.datatype == "array" + assert p.items_type == "integer" + + def test_pipe_separated_type(self): + p = Parameter(name="value", datatype="str|int") + assert p.datatype == "string|integer" + + def test_no_items_type_for_non_list(self): + p = Parameter(name="name", datatype="str") + assert p.items_type is None + + def test_schema_includes_enum(self): + p = Parameter(name="status", datatype="str", enum=["active", "inactive"]) + assert p.schema["enum"] == ["active", "inactive"] + + def test_schema_includes_default(self): + p = Parameter(name="page", datatype="int", default=1) + assert p.schema["default"] == 1 + + def test_in_request_body_for_body_location(self): + p = Parameter(name="data", datatype="str", location="body") + assert p.in_request_body() is True + + def test_in_request_body_for_form_location(self): + p = Parameter(name="data", datatype="str", location="form") + assert p.in_request_body() is True + + def test_not_in_request_body_for_query(self): + p = Parameter(name="data", datatype="str", location="query") + assert p.in_request_body() is False + + def test_parameter_dict_for_query(self): + p = Parameter(name="q", datatype="str", location="query", description="Search") + d = p.dict() + assert d["name"] == "q" + assert d["in"] == "query" + + def test_parameter_dict_for_body(self): + p = Parameter(name="data", datatype="str", location="body", description="Body data") + d = p.dict() + # body parameters return the request_body schema + assert "type" in d + + +class TestResponse: + """Tests for the Response class.""" + + def test_basic_response(self): + r = Response(code="200", description="Success", content=["json"]) + d = r.dict() + assert d["description"] == "Success" + assert "application/json" in d["content"] + + def test_default_schema(self): + r = Response(code="200", description="OK", content=["json"]) + assert r.schema == "default_response" + + def test_custom_schema(self): + r = Response(code="200", description="OK", content=["json"], schema="user") + assert r.schema == "user" + + def test_multiple_content_types(self): + r = Response(code="200", description="OK", content=["json", "xml"]) + d = r.dict() + assert "application/json" in d["content"] + assert "application/xml" in d["content"] + + +class TestOperation: + """Tests for the Operation class.""" + + def test_basic_operation(self): + op = Operation( + operation="get", + summary="Get users", + description="Returns all users", + operation_id="getUsers", + parameters=[], + responses=[{"code": "200", "description": "OK"}], + return_types=["json"], + ) + d = op.dict() + assert "get" in d + assert d["get"]["summary"] == "Get users" + assert d["get"]["operationId"] == "getUsers" + + def test_operation_with_parameters(self): + op = Operation( + operation="get", + summary="Test", + description="Test", + operation_id="test", + parameters=[{"name": "id", "datatype": "int", "location": "path"}], + responses=[], + return_types=["json"], + ) + d = op.dict() + assert len(d["get"]["parameters"]) == 1 + assert d["get"]["parameters"][0]["name"] == "id" + + def test_operation_with_body_parameters(self): + op = Operation( + operation="post", + summary="Create", + description="Create item", + operation_id="createItem", + parameters=[ + {"name": "data", "datatype": "str", "location": "body", "description": "Data", "required": True} + ], + responses=[], + return_types=["json"], + ) + d = op.dict() + # body params should not appear in parameters list + assert len(d["post"]["parameters"]) == 0 + assert "requestBody" in d["post"] + + def test_request_body_binary_type(self): + """Binary accept types should produce binary body schema.""" + op = Operation( + operation="post", + summary="Upload", + description="Upload file", + operation_id="upload", + parameters=[{"name": "file", "datatype": "str", "location": "body", "description": "File"}], + responses=[], + return_types=["binary"], + accepts=["binary"], + ) + body = op.request_body() + assert "content" in body + assert "application/octet-stream" in body["content"] + + def test_request_body_json_type(self): + """JSON accept types should produce normal body schema.""" + op = Operation( + operation="post", + summary="Create", + description="Create", + operation_id="create", + parameters=[{"name": "data", "datatype": "str", "location": "body", "description": "Data"}], + responses=[], + return_types=["json"], + accepts=["json"], + ) + body = op.request_body() + assert "content" in body + assert "application/json" in body["content"] + + def test_has_form(self): + op = Operation( + operation="post", + summary="Form", + description="Form", + operation_id="form", + parameters=[{"name": "field", "datatype": "str", "location": "form", "description": "Field"}], + responses=[], + return_types=["json"], + ) + assert op.has_form() is True + + def test_no_form(self): + op = Operation( + operation="get", + summary="Get", + description="Get", + operation_id="get", + parameters=[{"name": "q", "datatype": "str", "location": "query"}], + responses=[], + return_types=["json"], + ) + assert op.has_form() is False + + +class TestContact: + """Tests for the Contact class.""" + + def test_contact_dict(self): + c = Contact(name="Alice", url="https://example.com", email="alice@example.com") + d = c.dict() + assert d["name"] == "Alice" + assert d["url"] == "https://example.com" + assert d["email"] == "alice@example.com" + + def test_contact_repr(self): + c = Contact(name="Alice") + assert "Alice" in repr(c) + + +class TestLicense: + """Tests for the License class.""" + + def test_license_dict(self): + lic = License(name="MIT", url="https://mit.edu") + d = lic.dict() + assert d["name"] == "MIT" + assert d["url"] == "https://mit.edu" + + def test_license_defaults(self): + lic = License() + assert lic.name == "" + assert lic.url == "" + + +class TestContentMap: + """Tests for module-level constants.""" + + def test_json_mapping(self): + assert CONTENT_MAP["json"] == "application/json" + + def test_binary_mapping(self): + assert CONTENT_MAP["binary"] == "application/octet-stream" + + def test_binary_types_list(self): + assert "binary" in BINARY_TYPES + assert "gzip" in BINARY_TYPES + + def test_type_map(self): + assert TYPE_MAP["str"] == "string" + assert TYPE_MAP["int"] == "integer" + assert TYPE_MAP["list"] == "array" + assert TYPE_MAP["dict"] == "object" diff --git a/tests/test_sphinx_parser.py b/tests/test_sphinx_parser.py new file mode 100644 index 0000000..a176006 --- /dev/null +++ b/tests/test_sphinx_parser.py @@ -0,0 +1,226 @@ +"""Tests for reliqua.sphinx_parser module.""" + +from reliqua.sphinx_parser import SphinxParser + + +def make_method(docstring): + """Create a function with the given docstring for testing.""" + + def on_get(self, _req, _resp): + pass + + on_get.__doc__ = docstring + on_get.__qualname__ = "TestResource.on_get" + return on_get + + +class TestSphinxParserOptions: + """Tests for parse_options.""" + + def test_parse_location(self): + result = SphinxParser.parse_options("in=path required") + assert result["location"] == "path" + assert result["required"] is True + + def test_parse_query_default(self): + result = SphinxParser.parse_options("") + assert result["location"] == "query" + assert result["required"] is False + + def test_parse_enum(self): + result = SphinxParser.parse_options("enum=colors") + assert result["enum"] == "colors" + + def test_parse_default(self): + result = SphinxParser.parse_options("default=hello") + assert result["default"] == "hello" + + def test_optional_overrides_required(self): + result = SphinxParser.parse_options("required optional") + assert result["required"] is False + + def test_parse_min_max(self): + result = SphinxParser.parse_options("min=1 max=100") + assert result["min"] == "1" + assert result["max"] == "100" + + +class TestSphinxParserParameter: + """Tests for parse_parameter.""" + + def test_basic_parameter(self): + parser = SphinxParser() + result = parser.parse_parameter(":param str username: [in=query] User name") + assert result is not None + assert result["name"] == "username" + assert result["datatype"] == "str" + assert result["description"] == "User name" + + def test_parameter_with_path(self): + parser = SphinxParser() + result = parser.parse_parameter(":param int id: [in=path required] User ID") + assert result["name"] == "id" + assert result["datatype"] == "int" + assert result["location"] == "path" + assert result["required"] is True + + def test_parameter_no_options(self): + parser = SphinxParser() + result = parser.parse_parameter(":param str name: User's name") + assert result is not None + assert result["name"] == "name" + assert result["required"] is False + + def test_parameter_list_type(self): + parser = SphinxParser() + result = parser.parse_parameter(":param list[int] ids: [in=query] List of IDs") + assert result["datatype"] == "list[int]" + + def test_invalid_parameter_returns_none(self): + parser = SphinxParser() + result = parser.parse_parameter("not a parameter string") + assert result is None + + +class TestSphinxParserProperties: + """Tests for summary, description, parameters, responses, content, accepts properties.""" + + def test_summary(self): + parser = SphinxParser() + parser.doc = "Return all users.\n\nGet a list of users.\n\n:return json:" + assert parser.summary == "Return all users." + + def test_description(self): + parser = SphinxParser() + parser.doc = "Return all users.\n\nGet a list of users.\n\n:return json:" + assert parser.description == "Get a list of users." + + def test_empty_summary(self): + parser = SphinxParser() + parser.doc = "" + assert parser.summary == "" + + def test_parameters_parsed(self): + parser = SphinxParser() + parser.doc = ( + "Get users.\n\n:param str username: [in=query] Username\n:param int limit: [in=query] Limit\n:return json:" + ) + params = parser.parameters + assert len(params) == 2 + assert params[0]["name"] == "username" + assert params[1]["name"] == "limit" + + def test_responses_parsed(self): + parser = SphinxParser() + parser.doc = "Get users.\n\n:response 200: Success\n:response 400: Bad request\n:return json:" + responses = parser.responses + assert len(responses) == 2 + assert responses[0]["code"] == "200" + assert responses[1]["code"] == "400" + + def test_response_with_schema(self): + parser = SphinxParser() + parser.doc = "Get user.\n\n:response 200 user: User found\n:return json:" + responses = parser.responses + assert responses[0]["schema"] == "user" + + def test_content_default_json(self): + parser = SphinxParser() + parser.doc = "Get data.\n\nSome description." + assert parser.content == ["json"] + + def test_content_explicit(self): + parser = SphinxParser() + parser.doc = "Get data.\n\n:return [json xml]:" + assert "json" in parser.content + assert "xml" in parser.content + + def test_accepts_empty(self): + parser = SphinxParser() + parser.doc = "Get data.\n\n:return json:" + assert parser.accepts == [] + + def test_accepts_explicit(self): + parser = SphinxParser() + parser.doc = "Post data.\n\n:accepts [json,xml]:\n:return json:" + assert "json" in parser.accepts + assert "xml" in parser.accepts + + +class TestSphinxParserParse: + """Tests for the full parse method.""" + + def test_parse_full_method(self): + parser = SphinxParser() + + method = make_method( + "Return users.\n\n" + "Get a list of users.\n\n" + ":param str username: [in=query] Username\n" + ":response 200: Success\n" + ":return json:" + ) + + result = parser.parse(method) + assert result["operation"] == "get" + assert result["summary"] == "Return users." + assert result["description"] == "Get a list of users." + assert len(result["parameters"]) == 1 + assert len(result["responses"]) == 1 + assert result["return_types"] == ["json"] + + def test_parse_operation_detection(self): + parser = SphinxParser() + + for verb in ["get", "post", "put", "patch", "delete"]: + + def method(self): + """Test.""" + + method.__qualname__ = f"Res.on_{verb}" + method.__doc__ = "Test.\n\n:return json:" + result = parser.parse(method) + assert result["operation"] == verb + + def test_parse_suffix_detection(self): + parser = SphinxParser() + + def method(self): + r"""Test.\n\n:return json:""" + + method.__qualname__ = "Res.on_get_by_id" + method.__doc__ = "Test.\n\n:return json:" + + result = parser.parse(method) + assert result["suffix"] == "by_id" + + def test_parse_no_suffix(self): + parser = SphinxParser() + method = make_method("Test.\n\n:return json:") + result = parser.parse(method) + assert result["suffix"] is None + + def test_generate_operation_id(self): + method = make_method("Test.") + assert SphinxParser.generate_operation_id(method) == "TestResource.on_get" + + +class TestSphinxParserEdgeCases: + """Edge case tests.""" + + def test_multiline_parameter_description(self): + parser = SphinxParser() + parser.doc = "Get data.\n\n:param str name: [in=query] The user's\n full name\n:return json:" + params = parser.parameters + assert len(params) == 1 + assert "name" in params[0]["description"].lower() or params[0]["name"] == "name" + + def test_no_parameters(self): + parser = SphinxParser() + parser.doc = "Simple method.\n\n:return json:" + assert not parser.parameters + + def test_no_responses(self): + parser = SphinxParser() + parser.doc = "Simple method.\n\n:return json:" + assert not parser.responses diff --git a/tests/test_status_codes.py b/tests/test_status_codes.py new file mode 100644 index 0000000..9c84192 --- /dev/null +++ b/tests/test_status_codes.py @@ -0,0 +1,62 @@ +"""Tests for reliqua.status_codes module.""" + +from reliqua.status_codes import CODES, MESSAGES, http + + +class TestHttp: + """Tests for the http() function.""" + + def test_http_with_numeric_code(self): + """Return formatted status string for a numeric code.""" + assert http(200) == "200 OK" + + def test_http_with_string_code(self): + """Return formatted status string for a string code.""" + assert http("404") == "404 Not Found" + + def test_http_common_codes(self): + """Verify common HTTP status codes.""" + assert http(201) == "201 Created" + assert http(400) == "400 Bad Request" + assert http(401) == "401 Unauthorized" + assert http(403) == "403 Forbidden" + assert http(500) == "500 Internal Server Error" + + def test_http_with_message_key(self): + """Return status string when given a message name.""" + assert http("NOT_FOUND") == "404 NOT_FOUND" + assert http("OK") == "200 OK" + assert http("BAD_REQUEST") == "400 BAD_REQUEST" + + +class TestMessages: + """Tests for the MESSAGES dict.""" + + def test_messages_is_dict(self): + """MESSAGES should be a dict, not a set.""" + assert isinstance(MESSAGES, dict) + + def test_messages_lookup(self): + """MESSAGES keys should map to status code strings.""" + assert MESSAGES["OK"] == "200" + assert MESSAGES["NOT_FOUND"] == "404" + assert MESSAGES["BAD_REQUEST"] == "400" + assert MESSAGES["INTERNAL_SERVER_ERROR"] == "500" + + def test_messages_has_all_codes(self): + """MESSAGES should have an entry for every code.""" + assert len(MESSAGES) == len(CODES) + + +class TestCodes: + """Tests for the CODES dict.""" + + def test_codes_is_dict(self): + """CODES should be a dict.""" + assert isinstance(CODES, dict) + + def test_codes_has_common_entries(self): + """CODES should contain common HTTP status codes.""" + assert CODES["200"] == "OK" + assert CODES["404"] == "Not Found" + assert CODES["500"] == "Internal Server Error"