blob: 7e66bb2732cce570780088815c05d3757c5f3aaf [file]
"""The OpenAPI document, and the checks that keep it honest.
A hand-maintained API description rots. The tests here are the reason this one
should not: they compare the published document against the application's real
URL map and against the dataclass the API actually returns, so adding a route or
a field without describing it fails the build.
"""
from __future__ import annotations
import dataclasses
import json
import pathlib
from typing import Any
import pytest
import yaml
from openapi_spec_validator import validate as validate_spec
from asfcalendar import __version__, openapi
from asfcalendar.app import create_app
from asfcalendar.config import Config, from_dict
from asfcalendar.models import CATEGORIES, MAX_TITLE_LENGTH, VISIBILITIES, Event
from asfcalendar.storage import DEFAULT_SORT, MAX_LIMIT, SORT_COLUMNS
# Methods Quart adds by itself, which are not part of the described API.
IMPLICIT_METHODS = {"HEAD", "OPTIONS"}
@pytest.fixture
def document(config: Config) -> dict[str, Any]:
return openapi.build(config)
def api_operations(app: Any, base_path: str = "") -> set[tuple[str, str]]:
"""Every (path, method) the application actually serves under /api."""
prefix = f"{base_path}/api/"
found: set[tuple[str, str]] = set()
for rule in app.url_map.iter_rules():
if not rule.rule.startswith(prefix):
continue
path = openapi.openapi_path(rule.rule[len(base_path) :] if base_path else rule.rule)
for method in rule.methods or set():
if method not in IMPLICIT_METHODS:
found.add((path, method.lower()))
return found
def documented_operations(document: dict[str, Any]) -> set[tuple[str, str]]:
return {
(path, method)
for path, operations in document["paths"].items()
for method in operations
if method in {"get", "post", "put", "patch", "delete", "head", "options", "trace"}
}
class TestItIsAValidDocument:
def test_it_passes_a_real_openapi_validator(self, document: dict[str, Any]) -> None:
# The checks below are about this API in particular; this one is about
# the document being OpenAPI at all, so that Swagger UI and any other
# tool a consumer points at it will accept it.
validate_spec(document)
def test_it_is_still_valid_when_mounted_in_a_subdirectory(self, tmp_path: pathlib.Path) -> None:
cfg = from_dict(
{"server": {"base_url": "https://calendar.apache.org", "base_path": "/calendar"}},
root_dir=tmp_path,
)
validate_spec(openapi.build(cfg))
def test_it_is_still_valid_with_no_configured_origin(self, tmp_path: pathlib.Path) -> None:
validate_spec(openapi.build(from_dict({}, root_dir=tmp_path)))
def test_top_level_shape(self, document: dict[str, Any]) -> None:
assert document["openapi"].startswith("3.")
assert set(document) >= {"openapi", "info", "servers", "paths", "components"}
def test_info(self, document: dict[str, Any]) -> None:
assert document["info"]["version"] == __version__
assert "Test Calendar" in document["info"]["title"]
assert document["info"]["license"]["identifier"] == "Apache-2.0"
assert len(document["info"]["description"]) > 200
def test_every_operation_has_a_summary_and_at_least_one_response(self, document: dict[str, Any]) -> None:
for path, operations in document["paths"].items():
for method, operation in operations.items():
where = f"{method.upper()} {path}"
assert operation.get("summary"), f"{where} has no summary"
assert operation.get("responses"), f"{where} has no responses"
assert operation.get("tags"), f"{where} has no tag"
def test_every_operation_uses_a_declared_tag(self, document: dict[str, Any]) -> None:
declared = {tag["name"] for tag in document["tags"]}
for operations in document["paths"].values():
for operation in operations.values():
assert set(operation["tags"]) <= declared
def test_every_reference_resolves(self, document: dict[str, Any]) -> None:
schemas = document["components"]["schemas"]
def walk(node: Any) -> None:
if isinstance(node, dict):
target = node.get("$ref")
if isinstance(target, str):
assert target.startswith("#/components/schemas/"), target
assert target.rsplit("/", 1)[-1] in schemas, f"dangling {target}"
for value in node.values():
walk(value)
elif isinstance(node, list):
for value in node:
walk(value)
walk(document)
def test_no_schema_is_declared_and_never_used(self, document: dict[str, Any]) -> None:
text = json.dumps(document["paths"])
for name in document["components"]["schemas"]:
referenced = f'#/components/schemas/{name}"' in text
nested = f'#/components/schemas/{name}"' in json.dumps(document["components"]["schemas"])
assert referenced or nested, f"{name} is described but never referenced"
def test_it_survives_a_round_trip_through_json_and_yaml(self, document: dict[str, Any]) -> None:
assert json.loads(json.dumps(document)) == document
assert yaml.safe_load(yaml.safe_dump(document)) == document
class TestItMatchesTheRealApi:
"""The anti-drift checks. These are the point of the whole module."""
def test_every_route_is_documented(self, config: Config, document: dict[str, Any]) -> None:
app = create_app(config=config, testing=True, token_file=None)
undocumented = api_operations(app) - documented_operations(document)
assert not undocumented, "These endpoints exist but are missing from openapi.py: " + ", ".join(
f"{method.upper()} {path}" for path, method in sorted(undocumented)
)
def test_nothing_is_documented_that_does_not_exist(self, config: Config, document: dict[str, Any]) -> None:
app = create_app(config=config, testing=True, token_file=None)
imaginary = documented_operations(document) - api_operations(app)
assert not imaginary, "These endpoints are documented but not served: " + ", ".join(
f"{method.upper()} {path}" for path, method in sorted(imaginary)
)
def test_the_check_works_when_the_app_is_mounted_in_a_subdirectory(self, tmp_path: pathlib.Path) -> None:
cfg = from_dict({"server": {"base_path": "/calendar"}}, root_dir=tmp_path)
app = create_app(config=cfg, testing=True, token_file=None)
document = openapi.build(cfg)
# Paths in the document stay mount-point free; the prefix lives in
# `servers`, so the same document describes either deployment.
assert api_operations(app, "/calendar") == documented_operations(document)
assert all(path.startswith("/api/") for path in document["paths"])
def test_the_event_schema_matches_what_the_api_returns(self, document: dict[str, Any]) -> None:
event = Event(
id=1,
shortlink="AbCd2345",
title="t",
category="foundation",
visibility="public",
start=0,
end=3600,
all_day=False,
description="",
location="",
url="",
project=None,
timezone="UTC",
owner="carol",
created_at=0,
updated_at=0,
)
returned = set(event.to_json("https://example.org/e/AbCd2345"))
described = set(document["components"]["schemas"]["Event"]["properties"])
assert returned == described, (
f"Event.to_json() and the Event schema disagree: "
f"only in the API {returned - described}, only in the docs {described - returned}"
)
def test_the_schema_lists_every_dataclass_field_as_required(self, document: dict[str, Any]) -> None:
required = set(document["components"]["schemas"]["Event"]["required"])
assert {field.name for field in dataclasses.fields(Event)} <= required
def test_the_writable_fields_are_a_subset_of_the_event(self, document: dict[str, Any]) -> None:
schemas = document["components"]["schemas"]
writable = set(schemas["EventInput"]["properties"])
readable = set(schemas["Event"]["properties"])
assert writable <= readable, writable - readable
def test_read_only_fields_cannot_be_written(self, document: dict[str, Any]) -> None:
schemas = document["components"]["schemas"]
read_only = {name for name, schema in schemas["Event"]["properties"].items() if schema.get("readOnly")}
assert read_only & set(schemas["EventInput"]["properties"]) == set()
# The fields the API refuses to take from a client really are these.
assert read_only == {"id", "shortlink", "shortlink_url", "owner", "created_at", "updated_at"}
class TestItTracksTheConstants:
"""Change a limit in the code and the published schema follows."""
def test_categories_and_visibilities(self, document: dict[str, Any]) -> None:
properties = document["components"]["schemas"]["Event"]["properties"]
assert properties["category"]["enum"] == list(CATEGORIES)
assert properties["visibility"]["enum"] == list(VISIBILITIES)
def test_field_lengths(self, document: dict[str, Any]) -> None:
properties = document["components"]["schemas"]["Event"]["properties"]
assert properties["title"]["maxLength"] == MAX_TITLE_LENGTH
def test_sort_keys_and_the_limit_cap(self, document: dict[str, Any]) -> None:
parameters = {
parameter["name"]: parameter for parameter in document["paths"]["/api/events"]["get"]["parameters"]
}
assert parameters["sort"]["schema"]["enum"] == sorted(SORT_COLUMNS)
assert parameters["sort"]["schema"]["default"] == DEFAULT_SORT
assert parameters["limit"]["schema"]["maximum"] == MAX_LIMIT
def test_the_feed_takes_the_same_filters_as_the_listing(self, document: dict[str, Any]) -> None:
listing = document["paths"]["/api/events"]["get"]["parameters"]
feed = document["paths"]["/api/events.ics"]["get"]["parameters"]
assert [parameter["name"] for parameter in listing] == [parameter["name"] for parameter in feed]
def test_the_shortlink_pattern_accepts_a_real_token(self, document: dict[str, Any]) -> None:
import re
from asfcalendar import shortlink
pattern = document["components"]["schemas"]["Event"]["properties"]["shortlink"]["pattern"]
for _attempt in range(20):
assert re.match(pattern, shortlink.generate())
assert not re.match(pattern, "has space")
class TestServers:
def test_it_advertises_the_configured_origin(self, tmp_path: pathlib.Path) -> None:
cfg = from_dict({"server": {"base_url": "https://calendar.apache.org"}}, root_dir=tmp_path)
assert openapi.build(cfg)["servers"][0]["url"] == "https://calendar.apache.org"
def test_it_includes_the_mount_point(self, tmp_path: pathlib.Path) -> None:
cfg = from_dict(
{"server": {"base_url": "https://calendar.apache.org", "base_path": "/calendar"}},
root_dir=tmp_path,
)
assert openapi.build(cfg)["servers"][0]["url"] == "https://calendar.apache.org/calendar"
def test_it_falls_back_to_the_caller_host(self, tmp_path: pathlib.Path) -> None:
cfg = from_dict({}, root_dir=tmp_path)
assert openapi.build(cfg, origin="http://localhost:8080/")["servers"][0]["url"] == ("http://localhost:8080")
def test_it_is_never_empty(self, tmp_path: pathlib.Path) -> None:
cfg = from_dict({}, root_dir=tmp_path)
assert openapi.build(cfg)["servers"][0]["url"] == "/"
class TestTheEndpoints:
async def test_the_json_document_is_served(self, client: Any) -> None:
response = await client.get("/api/openapi.json")
assert response.status_code == 200
assert response.headers["Content-Type"].startswith("application/json")
body = await response.get_json()
assert body["openapi"].startswith("3.")
assert "/api/events" in body["paths"]
async def test_the_yaml_document_is_served(self, client: Any) -> None:
response = await client.get("/api/openapi.yaml")
assert response.status_code == 200
assert response.headers["Content-Type"].startswith("application/yaml")
body = yaml.safe_load(await response.get_data())
assert body["openapi"].startswith("3.")
async def test_both_formats_describe_the_same_api(self, client: Any) -> None:
as_json = await (await client.get("/api/openapi.json")).get_json()
as_yaml = yaml.safe_load(await (await client.get("/api/openapi.yaml")).get_data())
assert as_json == as_yaml
async def test_it_needs_no_session(self, client: Any) -> None:
# The docs page is public; an anonymous reader should be able to load it.
assert (await client.get("/api/openapi.json")).status_code == 200
async def test_the_server_url_follows_the_request(self, client: Any) -> None:
body = await (await client.get("/api/openapi.json")).get_json()
# The test client calls http://localhost, and base_url is set in the
# fixture config, so the configured origin wins.
assert body["servers"][0]["url"] == "https://calendar.example.org"
async def test_what_is_served_validates_too(self, client: Any) -> None:
# The document the tests build and the document a caller receives are
# produced the same way, but only one of them is what anybody uses.
validate_spec(await (await client.get("/api/openapi.json")).get_json())
async def test_it_describes_itself(self, client: Any) -> None:
body = await (await client.get("/api/openapi.json")).get_json()
assert "/api/openapi.json" in body["paths"]
assert "/api/openapi.yaml" in body["paths"]