openstack/codegenerator
Code Issues Proposed changes

Merge "Hardcode barbican container schemas"

Browse Source
This commit is contained in:
Zuul 2025-02-26 18:54:38 +00:00 committed by Gerrit Code Review
commit 7de4bb4f38
2 changed files with 377 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
codegenerator/openapi/barbican.py
View File

@ -20,6 +20,7 @@ import fixtures
from codegenerator.common.schema import SpecSchema
from codegenerator.openapi.base import OpenStackServerSourceBase
from codegenerator.openapi.utils import merge_api_ref_doc
from codegenerator.openapi.barbican_schemas import container
from codegenerator.openapi.barbican_schemas import secret
from codegenerator.openapi.barbican_schemas import secret_store
@ -29,7 +30,7 @@ from ruamel.yaml.scalarstring import LiteralScalarString
class BarbicanGenerator(OpenStackServerSourceBase):
URL_TAG_MAP = {}
RESOURCE_MODULES = [secret, secret_store]
RESOURCE_MODULES = [container, secret, secret_store]
def __init__(self):
pass
@ -345,7 +346,7 @@ class BarbicanGenerator(OpenStackServerSourceBase):
None,
"/v1/containers/{container_id}",
controller=containers.ContainerController.on_get,
action="get",
action="show",
conditions={"method": ["GET"]},
)
mapper.connect(
@ -392,6 +393,35 @@ class BarbicanGenerator(OpenStackServerSourceBase):
action="delete",
conditions={"method": ["DELETE"]},
)
# Container ACL
mapper.connect(
None,
"/v1/containers/{container_id}/acl",
controller=acls.ContainerACLsController.on_get,
action="get",
conditions={"method": ["GET"]},
)
mapper.connect(
None,
"/v1/containers/{container_id}/acl",
controller=acls.ContainerACLsController.on_put,
action="create",
conditions={"method": ["PUT"]},
)
mapper.connect(
None,
"/v1/containers/{container_id}/acl",
controller=acls.ContainerACLsController.on_patch,
action="update",
conditions={"method": ["PATCH"]},
)
mapper.connect(
None,
"/v1/containers/{container_id}/acl",
controller=acls.ContainerACLsController.on_delete,
action="delete",
conditions={"method": ["DELETE"]},
)
# Orders
mapper.connect(

345
codegenerator/openapi/barbican_schemas/container.py Normal file
View File

@ -0,0 +1,345 @@
# 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.
#
from typing import Any
from codegenerator.common.schema import ParameterSchema
from codegenerator.common.schema import TypeSchema
CONTAINER_CONSUMER_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "The name of the consumer set by the user.",
},
"url": {
"type": "string",
"description": "The URL for the user or service using the container.",
},
},
"required": ["name", "url"],
}
CONTAINER_CONSUMERS_LIST_PARAMETERS: dict[str, dict] = {
"offset": {
"in": "query",
"name": "offset",
"description": "The starting index within the total list of the secrets that you would like to retrieve.",
"schema": {"type": "integer"},
},
"limit": {
"in": "query",
"name": "limit",
"description": "The maximum number of records to return (up to 100). The default limit is 10.",
"schema": {"type": "integer"},
},
}
CONTAINER_CONSUMERS_RESPONSE: dict[str, Any] = {
"type": "object",
"properties": {
"next": {
"type": "string",
"description": "A HATEOAS URL to retrieve the next set of secrets based on the offset and limit parameters. This attribute is only available when the total number of secrets is greater than offset and limit parameter combined.",
},
"previous": {
"type": "string",
"description": "A HATEOAS URL to retrieve the previous set of secrets based on the offset and limit parameters. This attribute is only available when the request offset is greater than 0.",
},
"total": {
"type": "integer",
"description": "The total number of secrets available to the user.",
},
"consumers": {"type": "array", "items": CONTAINER_CONSUMER_SCHEMA},
},
"required": ["total", "consumers"],
"additionalProperties": False,
}
CONTAINER_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {
"consumers": {"type": "array", "items": CONTAINER_CONSUMER_SCHEMA},
"container_ref": {
"type": "string",
"description": "URL for referencing a specific container",
},
"created": {
"type": "string",
"format": "date-time",
"description": "Date and time container was created",
},
"name": {"type": "string", "description": "Container name"},
"secret_refs": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {
"type": ["string", "null"],
"description": "The name of the secret set by the user.",
},
"secret_ref": {
"type": "string",
"description": "Reference of the secret",
},
},
},
"description": "URL for referencing a specific secret store",
},
"status": {"type": "string", "description": "Container status"},
"type": {
"type": "string",
"enum": ["generic", "rsa", "certificate"],
"description": "Container type",
},
"updated": {
"type": ["string", "null"],
"format": "date-time",
"description": "Date and time container was last updated",
},
},
"required": ["name", "container_ref"],
"additionalProperties": False,
}
CONTAINER_LIST_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {
"next": {
"type": "string",
"description": "A HATEOAS URL to retrieve the next set of containers based on the offset and limit parameters. This attribute is only available when the total number of containers is greater than offset and limit parameter combined.",
},
"previous": {
"type": "string",
"description": "A HATEOAS URL to retrieve the previous set of containers based on the offset and limit parameters. This attribute is only available when the request offset is greater than 0.",
},
"total": {
"type": "integer",
"description": "The total number of containers available to the user.",
},
"containers": {
"type": "array",
"items": CONTAINER_SCHEMA,
"description": "A list of secret store references",
},
},
"required": ["containers"],
"additionalProperties": False,
}
CONTAINER_CREATE_REQUEST: dict[str, Any] = {
"type": "object",
"properties": {
"name": {"type": "string", "description": "Container name"},
"secret_refs": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {
"type": ["string", "null"],
"description": "The name of the secret set by the user.",
},
"secret_ref": {
"type": "string",
"description": "Reference of the secret",
},
},
"required": ["secret_ref"],
"additionalProperties": False,
},
"description": "URL for referencing a specific secrets",
},
"type": {
"type": "string",
"enum": ["generic", "rsa", "certificate"],
"description": "Container type",
},
},
"required": ["type", "secret_refs"],
"additionalProperties": False,
}
CONTAINER_CREATE_RESPONSE: dict[str, Any] = {
"type": "object",
"properties": {
"container_ref": {
"type": "string",
"description": "URL for referencing a specific container",
}
},
"required": ["container_ref"],
"additionalProperties": False,
}
CONTAINER_SECRET_CREATE_REQUEST: dict[str, Any] = {
"type": "object",
"properties": {
"name": {"type": "string", "description": "Container name"},
"secret_ref": {
"type": "string",
"description": "Reference of the secret",
},
},
"required": ["secret_ref"],
"additionalProperties": False,
}
CONTAINER_LIST_PARAMETERS: dict[str, dict] = {
"offset": {
"in": "query",
"name": "offset",
"description": "The starting index within the total list of the secrets that you would like to retrieve.",
"schema": {"type": "integer"},
},
"limit": {
"in": "query",
"name": "limit",
"description": "The maximum number of records to return (up to 100). The default limit is 10.",
"schema": {"type": "integer"},
},
}
CONTAINER_ACL_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {
"users": {
"type": "array",
"items": {"type": "string"},
"description": "List of user ids. This needs to be a user id as returned by Keystone.",
},
"project_access": {
"type": "boolean",
"description": "Flag to mark a container private so that the user who created the secret and users specified in above list can only access the container. Pass false to mark the container private.",
},
},
"additionalProperties": False,
}
CONTAINER_ACL_CONTAINER_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {"read": CONTAINER_ACL_SCHEMA},
"required": ["read"],
"additionalProperties": False,
}
CONTAINER_ACL_CREATE_RESPONSE: dict[str, Any] = {
"type": "object",
"properties": {"acl_ref": {"type": "string"}},
"required": ["acl_ref"],
"additionalProperties": False,
}
def _post_process_operation_hook(
openapi_spec, operation_spec, path: str | None = None
):
"""Hook to allow service specific generator to modify details"""
operationId = operation_spec.operationId
if operationId == "containers:get":
for key, val in CONTAINER_LIST_PARAMETERS.items():
openapi_spec.components.parameters.setdefault(
key, ParameterSchema(**val)
)
ref = f"#/components/parameters/{key}"
if ref not in [x.ref for x in operation_spec.parameters]:
operation_spec.parameters.append(ParameterSchema(ref=ref))
elif operationId == "containers/container_id/consumers:get":
for key, val in CONTAINER_CONSUMERS_LIST_PARAMETERS.items():
openapi_spec.components.parameters.setdefault(
key, ParameterSchema(**val)
)
ref = f"#/components/parameters/{key}"
if ref not in [x.ref for x in operation_spec.parameters]:
operation_spec.parameters.append(ParameterSchema(ref=ref))
elif operationId == "containers/container_id/consumers:delete":
operation_spec.requestBody = {
"content": {
"application/json": {"schema": CONTAINER_CONSUMER_SCHEMA}
}
}
operation_spec.responses = {"204": {"description": "ok"}}
def _get_schema_ref(
openapi_spec, name, description=None, schema_def=None, action_name=None
) -> tuple[str | None, str | None, bool]:
mime_type: str = "application/json"
ref: str
if name == "ContainersListResponse":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**CONTAINER_LIST_SCHEMA)
)
ref = f"#/components/schemas/{name}"
elif name in ["ContainerShowResponse", "ContainersConsumersPostResponse"]:
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**CONTAINER_SCHEMA)
)
ref = f"#/components/schemas/{name}"
elif name == "ContainersCreateRequest":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**CONTAINER_CREATE_REQUEST)
)
ref = f"#/components/schemas/{name}"
elif name == "ContainersCreateResponse":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**CONTAINER_CREATE_RESPONSE)
)
ref = f"#/components/schemas/{name}"
elif name == "ContainersSecretsCreateRequest":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**CONTAINER_SECRET_CREATE_REQUEST)
)
ref = f"#/components/schemas/{name}"
elif name == "ContainersSecretsCreateResponse":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**CONTAINER_CREATE_RESPONSE)
)
ref = f"#/components/schemas/{name}"
elif name == "ContainersConsumersGetResponse":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**CONTAINER_CONSUMERS_RESPONSE)
)
ref = f"#/components/schemas/{name}"
elif name == "ContainersConsumersPostRequest":
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**CONTAINER_CONSUMER_SCHEMA)
)
ref = f"#/components/schemas/{name}"
elif name in [
"ContainersAclGetResponse",
"ContainersAclCreateRequest",
"ContainersAclUpdateRequest",
]:
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**CONTAINER_ACL_CONTAINER_SCHEMA)
)
ref = f"#/components/schemas/{name}"
elif name in [
"ContainersAclCreateResponse",
"ContainersAclUpdateResponse",
]:
openapi_spec.components.schemas.setdefault(
name, TypeSchema(**CONTAINER_ACL_CREATE_RESPONSE)
)
ref = f"#/components/schemas/{name}"
else:
return (None, None, False)
return (ref, mime_type, True)