284 lines
11 KiB
Python
284 lines
11 KiB
Python
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
#
|
|
import inspect
|
|
import logging
|
|
from multiprocessing import Process
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import fixtures
|
|
|
|
from codegenerator.common.schema import SpecSchema
|
|
from codegenerator.openapi.base import OpenStackServerSourceBase
|
|
from codegenerator.openapi.base import UNSET
|
|
from codegenerator.openapi import designate_schemas
|
|
from codegenerator.openapi.utils import merge_api_ref_doc
|
|
|
|
from ruamel.yaml.scalarstring import LiteralScalarString
|
|
|
|
|
|
class DesignateGenerator(OpenStackServerSourceBase):
|
|
URL_TAG_MAP = {
|
|
"/zones/tasks/transfer_accepts": "zone-ownership-transfers-accepts",
|
|
"/zones/tasks/transfer_requests": "zone-ownership-transfers-requests",
|
|
"/zones/{zone_id}/tasks/transfer_requests": "zone-ownership-transfers-requests",
|
|
"/zones/tasks/imports": "zone-imports",
|
|
"/zones/tasks/exports": "zone-exports",
|
|
"/zones/{zone_id}/tasks/export": "zone-exports",
|
|
"/zones/{zone_id}/tasks": "zone-tasks",
|
|
"/zones/{zone_id}/recordsets": "recordsets",
|
|
"/zones/{zone_id}/shares": "shared-zones",
|
|
"/service_statuses": "service-statuses",
|
|
"/tlds": "tld",
|
|
"/tsigkeys": "tsigkey",
|
|
"/reverse/floatingips": "floatingips",
|
|
}
|
|
|
|
def __init__(self):
|
|
self.api_version = "2.1"
|
|
self.min_api_version = "2.0"
|
|
|
|
def _api_ver_major(self, ver):
|
|
return ver.ver_major
|
|
|
|
def _api_ver_minor(self, ver):
|
|
return ver.ver_minor
|
|
|
|
def _api_ver(self, ver):
|
|
return (ver.ver_major, ver.ver_minor)
|
|
|
|
def _build_routes(
|
|
self, mapper, node, path: str = "", param_url_map: dict[str, str] = {}
|
|
):
|
|
logging.debug(f"Processing controller {node} with {param_url_map}")
|
|
for part in [x for x in dir(node) if callable(getattr(node, x))]:
|
|
# Iterate over functions to find what is exposed on the current
|
|
# level
|
|
if part == "_route":
|
|
continue
|
|
obj = getattr(node, part)
|
|
_pecan = getattr(obj, "_pecan", None)
|
|
exposed = getattr(obj, "exposed", None)
|
|
if _pecan and exposed:
|
|
argspec = _pecan.get("argspec", None)
|
|
args: list[str] = []
|
|
if argspec:
|
|
args = argspec.args
|
|
if "self" in args:
|
|
args.remove("self")
|
|
# Only whatever is pecan exposed is of interest
|
|
conditions = {}
|
|
action = None
|
|
url: str = path
|
|
resource = None
|
|
# Check whether for mandatory params there is path defined
|
|
# If there is entry with all parameters we are most likely on
|
|
# the subcontroller level, however on the subcontroller there
|
|
# might be parent_id and id for which we do not have
|
|
# combination yet, thus take parent_id as base url
|
|
# for "zone_id" => /zones/{zone_id}/
|
|
# for "zone_id zone_transfer_request_id" => /zones/{zone_id}/tasks/transfer_requests/{transfer_request_id}
|
|
if " ".join(args) in param_url_map:
|
|
url = param_url_map[" ".join(args)]
|
|
elif " ".join(args[0:-1]) in param_url_map:
|
|
url = param_url_map[" ".join(args[0:-1])]
|
|
url += f"/{{{args[-1]}}}"
|
|
elif len(args) > 0:
|
|
url += f"/{{{args[-1]}}}"
|
|
|
|
# Identify the action from function name
|
|
# https://pecan.readthedocs.io/en/latest/rest.html#url-mapping
|
|
if part == "get_one":
|
|
conditions["method"] = ["GET"]
|
|
action = "show"
|
|
# if " ".join(args) not in param_url_map:
|
|
param_url_map[" ".join(args)] = url
|
|
elif part == "get_all":
|
|
conditions["method"] = ["GET"]
|
|
action = "list"
|
|
elif part in ["post", "post_all"]:
|
|
conditions["method"] = ["POST"]
|
|
action = "create"
|
|
elif part in ["put"]:
|
|
conditions["method"] = ["PUT"]
|
|
action = "update"
|
|
elif part in ["patch_one"]:
|
|
conditions["method"] = ["PATCH"]
|
|
action = "update"
|
|
elif part in ["delete", "delete_one"]:
|
|
conditions["method"] = ["DELETE"]
|
|
action = "delete"
|
|
|
|
if action:
|
|
# If we identified method as "interesting" register it into
|
|
# the routes mapper
|
|
mapper.connect(
|
|
None,
|
|
url,
|
|
controller=obj,
|
|
action=action,
|
|
conditions=conditions,
|
|
)
|
|
# yield part
|
|
if not hasattr(node, "__dict__"):
|
|
return
|
|
|
|
for subcontroller, v in (
|
|
node.__dict__.items()
|
|
if isinstance(node, type)
|
|
else node.__class__.__dict__.items()
|
|
):
|
|
# Iterate over node attributes for subcontrollers
|
|
if subcontroller.startswith("_"):
|
|
continue
|
|
if subcontroller in ["central_api", "__wrapped__", "SORT_KEYS"]:
|
|
# Not underested in those
|
|
continue
|
|
subpath = f"{path}/{subcontroller}"
|
|
next_param_url_map: dict = {
|
|
k: f"{v}/{subcontroller}" for k, v in param_url_map.items()
|
|
}
|
|
self._build_routes(mapper, v, subpath, next_param_url_map)
|
|
|
|
return
|
|
|
|
def generate(self, target_dir, args):
|
|
proc = Process(target=self._generate, args=[target_dir, args])
|
|
proc.start()
|
|
proc.join()
|
|
if proc.exitcode != 0:
|
|
raise RuntimeError("Error generating Designate OpenAPI schema")
|
|
|
|
def _generate(self, target_dir, args):
|
|
from designate.api.v2.controllers import root
|
|
from designate.api.v2.controllers import zones
|
|
from designate.api.v2.controllers.zones import nameservers
|
|
from designate.api.v2.controllers.zones import recordsets
|
|
from designate.api.v2.controllers.zones import sharedzones
|
|
from designate.api.v2.controllers.zones import tasks
|
|
|
|
from oslo_config import cfg
|
|
|
|
from pecan import make_app as pecan_make_app
|
|
from routes import Mapper
|
|
|
|
work_dir = Path(target_dir)
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
impl_path = Path(
|
|
work_dir, "openapi_specs", "dns", f"v{self.api_version}.yaml"
|
|
)
|
|
impl_path.parent.mkdir(parents=True, exist_ok=True)
|
|
openapi_spec = self.load_openapi(Path(impl_path))
|
|
if not openapi_spec:
|
|
openapi_spec = SpecSchema(
|
|
info={
|
|
"title": "OpenStack DNS API",
|
|
"description": LiteralScalarString(
|
|
"DNS API provided by Designate service"
|
|
),
|
|
"version": self.api_version,
|
|
},
|
|
openapi="3.1.0",
|
|
security=[{"ApiKeyAuth": []}],
|
|
components={
|
|
"securitySchemes": {
|
|
"ApiKeyAuth": {
|
|
"type": "apiKey",
|
|
"in": "header",
|
|
"name": "X-Auth-Token",
|
|
}
|
|
},
|
|
"parameters": {
|
|
"x-auth-all-projects": {
|
|
"name": "x-auth-all-projects",
|
|
"in": "header",
|
|
"schema": {"type": "boolean"},
|
|
"description": "If enabled this will show results from all projects in Designate",
|
|
},
|
|
"x-auth-sudo-project-id": {
|
|
"name": "x-auth-sudo-project-id",
|
|
"in": "header",
|
|
"schema": {"type": "string", "format": "uuid"},
|
|
"description": "This allows a user to impersonate another project",
|
|
},
|
|
},
|
|
},
|
|
)
|
|
|
|
self._buses = {}
|
|
|
|
self.useFixture(
|
|
fixtures.MonkeyPatch(
|
|
"designate.central.rpcapi.CentralAPI", mock.MagicMock()
|
|
)
|
|
)
|
|
|
|
self.app = pecan_make_app(root.RootController())
|
|
self.root = self.app.application.root
|
|
|
|
mapper = Mapper()
|
|
|
|
self._build_routes(mapper, root.RootController, "/v2")
|
|
for route in mapper.matchlist:
|
|
if (
|
|
route.routepath == "/v2/zones/{zone_id}/tasks/abandon"
|
|
and route.conditions.get("method", "GET")[0] == "GET"
|
|
):
|
|
# returns 405 in Designate anyway
|
|
continue
|
|
if route.routepath in [
|
|
"/v2/service_status",
|
|
"/v2/service_status/{service_id}",
|
|
]:
|
|
# backwards compatibility for mistake
|
|
continue
|
|
self._process_route(route, openapi_spec, framework="pecan")
|
|
|
|
if args.api_ref_src:
|
|
merge_api_ref_doc(
|
|
openapi_spec, args.api_ref_src, allow_strip_version=False
|
|
)
|
|
|
|
self.dump_openapi(openapi_spec, Path(impl_path), args.validate)
|
|
|
|
lnk = Path(impl_path.parent, "v2.yaml")
|
|
lnk.unlink(missing_ok=True)
|
|
lnk.symlink_to(impl_path.name)
|
|
|
|
return impl_path
|
|
|
|
def _get_schema_ref(
|
|
self,
|
|
openapi_spec,
|
|
name,
|
|
description: str | None = None,
|
|
schema_def=UNSET,
|
|
action_name=None,
|
|
):
|
|
(ref, mime_type, matched) = designate_schemas._get_schema_ref(
|
|
openapi_spec, name, description
|
|
)
|
|
if matched:
|
|
return (ref, mime_type)
|
|
|
|
# Default
|
|
(ref, mime_type) = super()._get_schema_ref(
|
|
openapi_spec,
|
|
name,
|
|
description,
|
|
schema_def=schema_def,
|
|
action_name=action_name,
|
|
)
|
|
return (ref, mime_type)
|