#!/usr/bin/env python
"""This module contains tests for metadata API handlers."""

import json

from absl import app
from absl.testing import absltest

from grr_response_core.lib import rdfvalue
from grr_response_core.lib.rdfvalues import paths as rdf_paths
from grr_response_core.lib.rdfvalues import structs as rdf_structs
from grr_response_proto import tests_pb2
from grr_response_server.gui import api_call_router
from grr_response_server.gui import api_test_lib
from grr_response_server.gui.api_plugins import metadata as metadata_plugin
from grr.test_lib import test_lib


class MetadataSimpleMessage(rdf_structs.RDFProtoStruct):
  protobuf = tests_pb2.MetadataSimpleMessage


class MetadataPrimitiveTypesMessage(rdf_structs.RDFProtoStruct):
  protobuf = tests_pb2.MetadataPrimitiveTypesMessage


class MetadataRepeatedFieldMessage(rdf_structs.RDFProtoStruct):
  protobuf = tests_pb2.MetadataRepeatedFieldMessage


class MetadataEnumFieldMessage(rdf_structs.RDFProtoStruct):
  protobuf = tests_pb2.MetadataEnumFieldMessage


class MetadataTypesHierarchyRoot(rdf_structs.RDFProtoStruct):
  protobuf = tests_pb2.MetadataTypesHierarchyRoot
  rdf_deps = [
      "MetadataTypesHierarchyCyclic",
      "MetadataTypesHierarchyLeaf",
  ]


class MetadataTypesHierarchyCyclic(rdf_structs.RDFProtoStruct):
  protobuf = tests_pb2.MetadataTypesHierarchyCyclic
  rdf_deps = [
      MetadataTypesHierarchyRoot,
      "MetadataTypesHierarchyLeaf",
  ]


class MetadataTypesHierarchyLeaf(rdf_structs.RDFProtoStruct):
  protobuf = tests_pb2.MetadataTypesHierarchyLeaf


class MetadataOneofMessage(rdf_structs.RDFProtoStruct):
  protobuf = tests_pb2.MetadataOneofMessage
  rdf_deps = [
      MetadataSimpleMessage,
  ]


class MetadataMapMessage(rdf_structs.RDFProtoStruct):
  protobuf = tests_pb2.MetadataMapMessage
  rdf_deps = [
      "FieldMapEntry",
  ]


class MetadataSemTypeMessage(rdf_structs.RDFProtoStruct):
  protobuf = tests_pb2.MetadataSemTypeMessage
  rdf_deps = [
      rdfvalue.RDFDatetime,
      rdfvalue.RDFDatetimeSeconds,
      rdfvalue.Duration,
      rdfvalue.DurationSeconds,
      rdfvalue.RDFBytes,
      rdfvalue.HashDigest,
      rdf_paths.GlobExpression,
      rdfvalue.ByteSize,
      rdfvalue.RDFURN,
      rdfvalue.SessionID,
  ]


class MetadataDummyApiCallRouter(api_call_router.ApiCallRouter):
  """Dummy `ApiCallRouter` implementation used for Metadata testing."""

  @api_call_router.ArgsType(MetadataSimpleMessage)
  @api_call_router.Http("GET", "/metadata_test/method1/<metadata_id>")
  @api_call_router.Http("HEAD", "/metadata_test/method1/<metadata_id>")
  @api_call_router.Http("POST", "/metadata_test/method1/<metadata_id>")
  def Method1WithArgsType(self, args, context=None):
    """Method 1 description."""

  @api_call_router.ResultType(MetadataSimpleMessage)
  @api_call_router.Http("GET", "/metadata_test/method2")
  @api_call_router.Http("HEAD", "/metadata_test/method2")
  @api_call_router.Http("POST", "/metadata_test/method2")
  def Method2WithResultType(self, args, context=None):
    """Method 2 description."""

  @api_call_router.ArgsType(MetadataPrimitiveTypesMessage)
  @api_call_router.ResultBinaryStream()
  @api_call_router.Http("GET", "/metadata_test/method3")
  def Method3PrimitiveTypes(self, args, context=None):
    """Method 3 description."""

  @api_call_router.ArgsType(MetadataRepeatedFieldMessage)
  @api_call_router.Http("GET", "/metadata_test/method4")
  @api_call_router.Http("POST", "/metadata_test/method4")
  def Method4RepeatedField(self, args, context=None):
    """Method 4 description."""

  @api_call_router.ArgsType(MetadataEnumFieldMessage)
  @api_call_router.Http("GET", "/metadata_test/method5")
  @api_call_router.Http("POST", "/metadata_test/method5")
  def Method5EnumField(self, args, context=None):
    """Method 5 description."""

  @api_call_router.ArgsType(MetadataTypesHierarchyRoot)
  @api_call_router.ResultType(MetadataTypesHierarchyLeaf)
  @api_call_router.Http("GET", "/metadata_test/method6")
  def Method6TypeReferences(self, args, context=None):
    """Method 6 description."""

  @api_call_router.ArgsType(MetadataOneofMessage)
  @api_call_router.ResultType(MetadataOneofMessage)
  @api_call_router.Http("GET", "/metadata_test/method7")
  @api_call_router.Http("POST", "/metadata_test/method7")
  def Method7ProtobufOneof(self, args, context=None):
    """Method 7 description."""

  @api_call_router.ArgsType(MetadataMapMessage)
  @api_call_router.ResultType(MetadataMapMessage)
  @api_call_router.Http("GET", "/metadata_test/method8")
  @api_call_router.Http("POST", "/metadata_test/method8")
  def Method8ProtobufMap(self, args, context=None):
    """Method 8 description."""

  @api_call_router.ArgsType(MetadataSimpleMessage)
  @api_call_router.ResultType(MetadataSimpleMessage)
  @api_call_router.Http("GET", "/metadata_test/method9")
  @api_call_router.Http("GET", "/metadata_test/method9/<metadata_id>")
  @api_call_router.Http("POST",
                        "/metadata_test/method9/<metadata_id>/<metadata_arg1>")
  @api_call_router.Http(
      "GET",
      "/metadata_test/method9/<metadata_id>/<metadata_arg1>/<metadata_arg2>")
  @api_call_router.Http("GET", "/metadata_test/method9/<metadata_id>/fixed1")
  @api_call_router.Http(
      "GET", "/metadata_test/method9/<metadata_id>/fixed1/<metadata_arg1>")
  @api_call_router.Http("GET", "/metadata_test/method9/fixed2/")  # Trailing /.
  @api_call_router.Http("GET", "/metadata_test/method9/fixed2/<metadata_arg1>/")
  @api_call_router.Http(
      "GET", "/metadata_test/method9/fixed2/<metadata_arg1>/<metadata_arg2>/")
  def Method9OptionalPathArgs(self, args, context=None):
    """Method 9 description."""

  @api_call_router.ArgsType(MetadataSemTypeMessage)
  @api_call_router.ResultType(MetadataSemTypeMessage)
  @api_call_router.Http("GET", "/metadata_test/method10")
  @api_call_router.Http("POST", "/metadata_test/method10")
  def Method10SemTypeProtobufOption(self, args, context=None):
    """Method 10 description."""


class ApiGetOpenApiDescriptionHandlerTest(api_test_lib.ApiCallHandlerTest):
  """Test for `ApiGetOpenApiDescriptionHandler`."""

  def setUp(self):
    super().setUp()
    self.router = MetadataDummyApiCallRouter()
    self.router_methods = self.router.__class__.GetAnnotatedMethods()

    self.handler = metadata_plugin.ApiGetOpenApiDescriptionHandler(self.router)

    result = self.handler.Handle(None, context=self.context)
    self.openapi_desc = result.openapi_description
    self.openapi_desc_dict = json.loads(self.openapi_desc)

  def testRouterAnnotatedMethodsAreAllExtracted(self):
    expected_methods = {
        "Method1WithArgsType",
        "Method2WithResultType",
        "Method3PrimitiveTypes",
        "Method4RepeatedField",
        "Method5EnumField",
        "Method6TypeReferences",
        "Method7ProtobufOneof",
        "Method8ProtobufMap",
        "Method9OptionalPathArgs",
        "Method10SemTypeProtobufOption",
    }
    extracted_methods = {method.name for method in self.router_methods.values()}

    self.assertEqual(expected_methods, extracted_methods)

  def testGeneratedOpenApiDescriptionIsValid(self):
    # TODO(user): Move this import to the top when the GitHub
    # issue #813 (https://github.com/google/grr/issues/813) is resolved.
    try:
      # pytype: disable=import-error
      # pylint: disable=g-import-not-at-top
      import openapi_spec_validator
      # pytype: enable=import-error
      # pylint: enable=g-import-not-at-top
    except ImportError:
      raise absltest.SkipTest("`openapi-spec-validator` not installed")

    # Will raise exceptions when the OpenAPI specification is invalid.
    openapi_spec_validator.validate_spec(self.openapi_desc_dict)

  def testAllRoutesAreInOpenApiDescription(self):
    openapi_paths_dict = self.openapi_desc_dict["paths"]

    # Check that there are no extra/missing paths.
    self.assertCountEqual(
        {
            "/metadata_test/method1/{metadataId}",
            "/metadata_test/method2",
            "/metadata_test/method3",
            "/metadata_test/method4",
            "/metadata_test/method5",
            "/metadata_test/method6",
            "/metadata_test/method7",
            "/metadata_test/method8",
            "/metadata_test/method9/{metadataId}",
            "/metadata_test/method9/{metadataId}/{metadataArg1}",
            "/metadata_test/method9/{metadataId}/{metadataArg1}/{metadataArg2}",
            "/metadata_test/method9/{metadataId}/fixed1/{metadataArg1}",
            "/metadata_test/method9/fixed2/{metadataArg1}/{metadataArg2}",
            "/metadata_test/method10",
        }, openapi_paths_dict.keys())

    # Check that there are no extra/missing HTTP methods for each path (routes).
    self.assertCountEqual(
        {"get", "head", "post"},
        openapi_paths_dict["/metadata_test/method1/{metadataId}"].keys())
    self.assertCountEqual({"get", "head", "post"},
                          openapi_paths_dict["/metadata_test/method2"].keys())
    self.assertCountEqual({"get"},
                          openapi_paths_dict["/metadata_test/method3"].keys())
    self.assertCountEqual({"get", "post"},
                          openapi_paths_dict["/metadata_test/method4"].keys())
    self.assertCountEqual({"get", "post"},
                          openapi_paths_dict["/metadata_test/method5"].keys())
    self.assertCountEqual({"get"},
                          openapi_paths_dict["/metadata_test/method6"].keys())
    self.assertCountEqual({"get", "post"},
                          openapi_paths_dict["/metadata_test/method7"].keys())
    self.assertCountEqual({"get", "post"},
                          openapi_paths_dict["/metadata_test/method8"].keys())
    self.assertCountEqual(
        {"get"},
        openapi_paths_dict["/metadata_test/method9/{metadataId}"].keys())
    self.assertCountEqual(
        {"post"},
        openapi_paths_dict.get(
            "/metadata_test/method9/{metadataId}/{metadataArg1}").keys())
    self.assertCountEqual(
        {"get"},
        openapi_paths_dict.get(
            "/metadata_test/method9/{metadataId}/{metadataArg1}/{metadataArg2}")
        .keys())
    self.assertCountEqual(
        {"get"},
        openapi_paths_dict.get(
            "/metadata_test/method9/{metadataId}/fixed1/{metadataArg1}").keys())

  def testRouteArgsAreCorrectlySeparated(self):
    # Check that the parameters are separated correctly in path, query and
    # request body parameters.

    openapi_paths_dict = self.openapi_desc_dict["paths"]

    # Check the OpenAPI parameters of `Method1WithArgsType` routes.
    method1_path_dict = (
        openapi_paths_dict["/metadata_test/method1/{metadataId}"])

    # Parameters of `GET /metadata_test/method1/{metadata_id}`.
    get_method1_dict = method1_path_dict["get"]

    get_method1_params_path = [
        param["name"]
        for param in get_method1_dict["parameters"]
        if param["in"] == "path"
    ]
    self.assertCountEqual(["metadataId"], get_method1_params_path)

    get_method1_params_query = [
        param["name"]
        for param in get_method1_dict["parameters"]
        if param["in"] == "query"
    ]
    self.assertCountEqual(["metadataArg1", "metadataArg2"],
                          get_method1_params_query)

    # `parameters` field is always present, even if it is an empty array,
    # `requestBody` should not be, unless there are arguments in the body.
    self.assertIsNone(get_method1_dict.get("requestBody"))

    # Parameters of `HEAD /metadata_test/method1/{metadata_id}`.
    head_method1_dict = method1_path_dict["head"]

    head_method1_params_path = [
        param["name"]
        for param in head_method1_dict["parameters"]
        if param["in"] == "path"
    ]
    self.assertCountEqual(["metadataId"], head_method1_params_path)

    head_method1_params_query = [
        param["name"]
        for param in get_method1_dict["parameters"]
        if param["in"] == "query"
    ]
    self.assertCountEqual(["metadataArg1", "metadataArg2"],
                          head_method1_params_query)

    self.assertIsNone(head_method1_dict.get("requestBody"))

    # Parameters of `POST /metadata_test/method1/{metadata_id}`.
    post_method1_dict = method1_path_dict["post"]

    post_method1_params_path = [
        param["name"]
        for param in post_method1_dict["parameters"]
        if param["in"] == "path"
    ]
    self.assertCountEqual(["metadataId"], post_method1_params_path)

    post_method1_params_query = [
        param["name"]
        for param in post_method1_dict["parameters"]
        if param["in"] == "query"
    ]
    self.assertEmpty(post_method1_params_query)

    # requestBody parameters use their names as keys.
    post_method1_params_body = list(
        post_method1_dict.get("requestBody").get("content").get(
            "application/json").get("schema").get("properties"))
    self.assertCountEqual(["metadataArg1", "metadataArg2"],
                          post_method1_params_body)

    # Check the OpenAPI parameters of `Method4RepeatedField` routes.
    method4_path_dict = openapi_paths_dict["/metadata_test/method4"]

    # Parameters of `GET /metadata_test/method4`.
    get_method4_dict = method4_path_dict["get"]

    get_method4_params_path = [
        param["name"]
        for param in get_method4_dict["parameters"]
        if param["in"] == "path"
    ]
    self.assertEmpty(get_method4_params_path)

    get_method4_params_query = [
        param["name"]
        for param in get_method4_dict["parameters"]
        if param["in"] == "query"
    ]
    self.assertCountEqual(["fieldRepeated"], get_method4_params_query)

    self.assertIsNone(get_method4_dict.get("requestBody"))

    # Parameters of POST /metadata_test/method4.
    post_method4_dict = method4_path_dict["post"]

    post_method4_params_path = [
        param["name"]
        for param in post_method4_dict["parameters"]
        if param["in"] == "path"
    ]
    self.assertEmpty(post_method4_params_path)

    post_method4_params_query = [
        param["name"]
        for param in post_method4_dict["parameters"]
        if param["in"] == "query"
    ]
    self.assertEmpty(post_method4_params_query)

    post_method4_params_body = list(
        post_method4_dict.get("requestBody").get("content").get(
            "application/json").get("schema").get("properties"))
    self.assertCountEqual(["fieldRepeated"], post_method4_params_body)

    # Check the OpenAPI parameters of `Method5EnumField` routes.
    method5_path_dict = openapi_paths_dict["/metadata_test/method5"]

    # Parameters of `GET /metadata_test/method5`.
    get_method5_dict = method5_path_dict["get"]

    get_method5_params_path = [
        param["name"]
        for param in get_method5_dict["parameters"]
        if param["in"] == "path"
    ]
    self.assertEmpty(get_method5_params_path)

    get_method5_params_query = [
        param["name"]
        for param in get_method5_dict["parameters"]
        if param["in"] == "query"
    ]
    self.assertCountEqual(["fieldEnum"], get_method5_params_query)

    self.assertIsNone(get_method5_dict.get("requestBody"))

    # Parameters of `POST /metadata_test/method5`.
    post_method5_dict = method5_path_dict["post"]

    post_method5_params_path = [
        param["name"]
        for param in post_method5_dict["parameters"]
        if param["in"] == "path"
    ]
    self.assertEmpty(post_method5_params_path)

    post_method5_params_query = [
        param["name"]
        for param in post_method5_dict["parameters"]
        if param["in"] == "query"
    ]
    self.assertEmpty(post_method5_params_query)

    post_method5_params_body = list(
        post_method5_dict.get("requestBody").get("content").get(
            "application/json").get("schema").get("properties"))
    self.assertCountEqual(["fieldEnum"], post_method5_params_body)

    # Check the OpenAPI parameters of `Method6TypeReferences` routes.
    method6_path_dict = openapi_paths_dict["/metadata_test/method6"]

    # Parameters of `GET /metadata_test/method6`.
    get_method6_dict = method6_path_dict["get"]

    get_method6_params_path = [
        param["name"]
        for param in get_method6_dict["parameters"]
        if param["in"] == "path"
    ]
    self.assertEmpty(get_method6_params_path)

    get_method6_params_query = [
        param["name"]
        for param in get_method6_dict["parameters"]
        if param["in"] == "query"
    ]
    self.assertCountEqual(["fieldInt64", "child1", "child2"],
                          get_method6_params_query)

    self.assertIsNone(get_method6_dict.get("requestBody"))

    # Check the OpenAPI parameters of `Method7ProtobufOneof` routes.
    method7_path_dict = openapi_paths_dict["/metadata_test/method7"]

    # Parameters of `GET /metadata_test/method7`.
    get_method7_dict = method7_path_dict["get"]

    get_method7_params_path = [
        param["name"]
        for param in get_method7_dict["parameters"]
        if param["in"] == "path"
    ]
    self.assertEmpty(get_method7_params_path)

    get_method7_params_query = [
        param["name"]
        for param in get_method7_dict["parameters"]
        if param["in"] == "query"
    ]
    self.assertCountEqual(["oneofInt64", "oneofSimplemsg", "fieldInt64"],
                          get_method7_params_query)

    self.assertIsNone(get_method7_dict.get("requestBody"))

  def testRoutesResultsAreCorrectlyDescribedInOpenApiDescription(self):
    # Verify the OpenAPI schemas of the response objects.
    # Response types are usually protobuf messages. The expectation in these
    # cases is that the route descriptions include a reference to the type
    # schemas in the `components` field of the root `OpenAPI Object`.
    openapi_paths_dict = self.openapi_desc_dict["paths"]

    # `Method2WithResultType (GET, HEAD, POST)` => `MetadataSimpleMessage`
    method2_path_dict = openapi_paths_dict["/metadata_test/method2"]
    # Check responses for `GET /metadata_test/method2`.
    self.assertEqual(
        {
            "200": {
                "description":
                    f"The call to the Method2WithResultType API method "
                    f"succeeded and it returned an instance of "
                    f"{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}.",
                "content": {
                    "application/json": {
                        "schema": {
                            "$ref":
                                f"#/components/schemas/{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}",
                        },
                    },
                },
            },
            "default": {
                "description":
                    "The call to the Method2WithResultType API method did "
                    "not succeed.",
            },
        }, method2_path_dict["get"]["responses"])
    # Check responses for HEAD /metadata_test/method2.
    self.assertEqual(
        {
            "200": {
                "description":
                    f"The call to the Method2WithResultType API method "
                    f"succeeded and it returned an instance of "
                    f"{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}.",
                "content": {
                    "application/json": {
                        "schema": {
                            "$ref":
                                f"#/components/schemas/{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}",
                        },
                    },
                },
            },
            "default": {
                "description":
                    "The call to the Method2WithResultType API method did "
                    "not succeed.",
            },
        }, method2_path_dict["head"]["responses"])
    # Check responses for `POST /metadata_test/method2`.
    self.assertEqual(
        {
            "200": {
                "description":
                    f"The call to the Method2WithResultType API method "
                    f"succeeded and it returned an instance of "
                    f"{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}.",
                "content": {
                    "application/json": {
                        "schema": {
                            "$ref":
                                f"#/components/schemas/{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}",
                        },
                    },
                },
            },
            "default": {
                "description":
                    "The call to the Method2WithResultType API method did "
                    "not succeed.",
            },
        }, method2_path_dict["post"]["responses"])

    # Method3PrimitiveTypes (GET) => BinaryStream result type.
    method3_path_dict = openapi_paths_dict["/metadata_test/method3"]
    # Check responses for GET /metadata_test/method3.
    self.assertEqual(
        {
            "200": {
                "description":
                    "The call to the Method3PrimitiveTypes API method "
                    "succeeded and it returned an instance of "
                    "BinaryStream.",
                "content": {
                    "application/octet-stream": {
                        "schema": {
                            "$ref": "#/components/schemas/BinaryStream",
                        },
                    },
                },
            },
            "default": {
                "description":
                    "The call to the Method3PrimitiveTypes API method did "
                    "not succeed.",
            },
        }, method3_path_dict["get"]["responses"])

    # `Method4RepeatedField (GET, POST)` => No result type.
    method4_path_dict = openapi_paths_dict["/metadata_test/method4"]
    # Check responses for `GET /metadata_test/method4`.
    self.assertEqual(
        {
            "200": {
                "description":
                    "The call to the Method4RepeatedField API method "
                    "succeeded.",
            },
            "default": {
                "description":
                    "The call to the Method4RepeatedField API method did "
                    "not succeed.",
            },
        }, method4_path_dict["get"]["responses"])
    # Check responses for `POST /metadata_test/method4`.
    self.assertEqual(
        {
            "200": {
                "description":
                    "The call to the Method4RepeatedField API method "
                    "succeeded.",
            },
            "default": {
                "description":
                    "The call to the Method4RepeatedField API method did "
                    "not succeed.",
            },
        }, method4_path_dict["post"]["responses"])

  def testPrimitiveTypesAreCorrectlyDescribedAndUsedInOpenApiDescription(self):
    # Primitive type schemas are described in the `components` field of the
    # root `OpenAPI Object`.

    # Firstly, verify that the descriptions of the fields of the
    # `MetadataPrimitiveTypesMessage` (which is the `ArgsType` of
    # `Method3PrimitiveTypes`) include references to the primitive type
    # descriptions.
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_DOUBLE"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldDouble"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_FLOAT"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldFloat"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_INT64"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldInt64"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_UINT64"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldUint64"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_INT32"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldInt32"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_FIXED64"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldFixed64"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_FIXED32"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldFixed32"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_BOOL"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldBool"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_STRING"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldString"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_BYTES"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldBytes"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_UINT32"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldUint32"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_SFIXED32"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldSfixed32"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_SFIXED64"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldSfixed64"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_SINT32"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldSint32"))
    self.assertEqual({"$ref": "#/components/schemas/protobuf2.TYPE_SINT64"},
                     self._GetParamSchema("/metadata_test/method3", "get",
                                          "fieldSint64"))

    # Extract `BinaryStream` type reference from the response schema.
    self.assertEqual(
        {"$ref": "#/components/schemas/BinaryStream"},
        self.openapi_desc_dict.get("paths").get("/metadata_test/method3").get(
            "get").get("responses").get("200").get("content").get(
                "application/octet-stream").get("schema"))

    # Secondly, verify the descriptions of primitive types from the `components`
    # field of the root `OpenAPI Object`.
    component_schemas = self.openapi_desc_dict["components"]["schemas"]

    self.assertEqual({
        "type": "number",
        "format": "double"
    }, component_schemas["protobuf2.TYPE_DOUBLE"])
    self.assertEqual({
        "type": "number",
        "format": "float"
    }, component_schemas["protobuf2.TYPE_FLOAT"])
    self.assertEqual({
        "type": "string",
        "format": "int64"
    }, component_schemas["protobuf2.TYPE_INT64"])
    self.assertEqual({
        "type": "string",
        "format": "uint64"
    }, component_schemas["protobuf2.TYPE_UINT64"])
    self.assertEqual({
        "type": "integer",
        "format": "int32"
    }, component_schemas["protobuf2.TYPE_INT32"])
    self.assertEqual({
        "type": "string",
        "format": "fixed64"
    }, component_schemas["protobuf2.TYPE_FIXED64"])
    self.assertEqual({
        "type": "number",
        "format": "fixed32"
    }, component_schemas["protobuf2.TYPE_FIXED32"])
    self.assertEqual({"type": "boolean"},
                     component_schemas["protobuf2.TYPE_BOOL"])
    self.assertEqual({"type": "string"},
                     component_schemas["protobuf2.TYPE_STRING"])
    self.assertEqual({
        "type": "string",
        "format": "byte"
    }, component_schemas["protobuf2.TYPE_BYTES"])
    self.assertEqual({
        "type": "number",
        "format": "uint32"
    }, component_schemas["protobuf2.TYPE_UINT32"])
    self.assertEqual({
        "type": "number",
        "format": "sfixed32"
    }, component_schemas["protobuf2.TYPE_SFIXED32"])
    self.assertEqual({
        "type": "string",
        "format": "sfixed64"
    }, component_schemas["protobuf2.TYPE_SFIXED64"])
    self.assertEqual({
        "type": "integer",
        "format": "int32"
    }, component_schemas["protobuf2.TYPE_INT32"])
    self.assertEqual({
        "type": "string",
        "format": "sint64"
    }, component_schemas["protobuf2.TYPE_SINT64"])
    self.assertEqual({
        "type": "string",
        "format": "binary"
    }, component_schemas["BinaryStream"])

  def testRepeatedFieldIsDescribedCorrectlyInOpenApiDescription(self):
    # Extract the `repeated` field from the description of the `GET` route
    # associated with `Method4RepeatedField`.
    # The `repeated` field should be the only parameter and be a query
    # parameter. This aspect is tested by `testRouteArgsAreCorrectlySeparated`.
    get_method4_repeated_field_schema = (
        self.openapi_desc_dict.get("paths").get("/metadata_test/method4").get(
            "get").get("parameters")[0].get("schema"))
    self.assertEqual(
        {
            "type": "array",
            "items": {
                "$ref": "#/components/schemas/protobuf2.TYPE_INT64",
            },
        }, get_method4_repeated_field_schema)

    # Extract the `repeated` field from the description of the `POST` route
    # associated with `Method4RepeatedField`.
    # The repeated field should be the only parameter and be a request body
    # parameter. This aspect is tested by `testRouteArgsAreCorrectlySeparated`.
    post_method4_repeated_field_schema = (
        self.openapi_desc_dict.get("paths").get("/metadata_test/method4").get(
            "post").get("requestBody").get("content").get("application/json")
        .get("schema").get("properties").get("fieldRepeated"))
    self.assertEqual(
        {
            "type": "array",
            "items": {
                "$ref": "#/components/schemas/protobuf2.TYPE_INT64",
            },
        }, post_method4_repeated_field_schema)

  def testEnumFieldIsDescribedCorrectlyInOpenApiDescription(self):
    # The enum type is user defined, so it is reusable and described separately
    # from the field itself, in the `component` field of the root `OpenAPI
    # Object`. Therefore, the expectation is that this field has a reference to
    # the enum type description.
    # We test both the references and the type definition.

    # Test the OpenAPI schema description of the *enum field*.

    # Extract the enum field schema from the description of the `GET` route
    # associated with `Method5EnumField`.
    get_method5_enum_field_schema = (
        self.openapi_desc_dict.get("paths").get("/metadata_test/method5").get(
            "get").get("parameters")[0].get("schema"))
    self.assertEqual(
        {
            "description":
                "UNKNOWN == 0\nFOO == 1\nBAR == 2",
            "allOf": [{
                "$ref":
                    f"#/components/schemas/{tests_pb2.MetadataEnumFieldMessage.MetadataEnum.DESCRIPTOR.full_name}",
            },],
        }, get_method5_enum_field_schema)

    # Extract the enum field schema from the description of the POST route
    # associated with `Method5EnumField`.
    post_method5_enum_field_schema = (
        self.openapi_desc_dict.get("paths").get("/metadata_test/method5").get(
            "post").get("requestBody").get("content").get("application/json")
        .get("schema").get("properties").get("fieldEnum"))
    self.assertEqual(
        {
            "description":
                "UNKNOWN == 0\nFOO == 1\nBAR == 2",
            "allOf": [{
                "$ref":
                    f"#/components/schemas/{tests_pb2.MetadataEnumFieldMessage.MetadataEnum.DESCRIPTOR.full_name}",
            },],
        }, post_method5_enum_field_schema)

    # Test the OpenAPI schema description of the *enum field's type*.
    openapi_enum_type_schema = (
        self.openapi_desc_dict.get("components").get("schemas").get(
            f"{tests_pb2.MetadataEnumFieldMessage.MetadataEnum.DESCRIPTOR.full_name}"
        ))

    self.assertEqual(
        {
            "type": "string",
            "enum": ["UNKNOWN", "FOO", "BAR"],
            "description": "UNKNOWN == 0\nFOO == 1\nBAR == 2",
        }, openapi_enum_type_schema)

  def testTypeReferencesAreUsedInOpenApiDescriptionsOfCompositeTypes(self):
    # Tests that references are used for composite types and that the referenced
    # types are declared, including the case of cyclic dependencies.

    # Test that composite types are defined.
    openapi_component_schemas = (
        self.openapi_desc_dict.get("components").get("schemas"))
    root = openapi_component_schemas[
        tests_pb2.MetadataTypesHierarchyRoot.DESCRIPTOR.full_name]
    cyclic = openapi_component_schemas[
        tests_pb2.MetadataTypesHierarchyCyclic.DESCRIPTOR.full_name]

    # Test the references between message types.
    root_ref = f"#/components/schemas/{tests_pb2.MetadataTypesHierarchyRoot.DESCRIPTOR.full_name}"
    cyclic_ref = f"#/components/schemas/{tests_pb2.MetadataTypesHierarchyCyclic.DESCRIPTOR.full_name}"
    leaf_ref = f"#/components/schemas/{tests_pb2.MetadataTypesHierarchyLeaf.DESCRIPTOR.full_name}"

    self.assertEqual(root["properties"]["child1"]["$ref"], cyclic_ref)
    self.assertEqual(root["properties"]["child2"]["$ref"], leaf_ref)

    self.assertEqual(cyclic["properties"]["root"]["$ref"], root_ref)  # Cycle.
    self.assertEqual(cyclic["properties"]["child1"]["$ref"], leaf_ref)

  def testProtobufOneofIsDescribedCorrectlyInOpenApiDescription(self):
    # The semantic of `protobuf.oneof` is not currently supported by the
    # OpenAPI Specification (see this GitHub issue [1] for more details) and
    # for the moment we just add the `oneof`'s fields as regular `Parameter
    # Object`s or as schema properties with a `description` field stating that
    # only one of the `protobuf.oneof`'s fields should be present at a time.
    #
    # [1]: github.com/google/grr/issues/822

    # Check the description of the `protobuf.oneof` from the `parameters` field
    # of the `Operation Object` associated with `GET /metadata-test/method7`.
    # Check the `oneof_int64` inner field of the `metadata_oneof`.
    self.assertEqual(
        {
            "description":
                "This field is part of the \"metadata_oneof\" oneof. "
                "Only one field per oneof should be present.",
            "allOf": [{
                "$ref": "#/components/schemas/protobuf2.TYPE_INT64",
            },],
        }, self._GetParamSchema("/metadata_test/method7", "get", "oneofInt64"))
    # Check the `oneof_simplemsg` inner field of the `metadata_oneof`.
    self.assertEqual(
        {
            "description":
                "This field is part of the \"metadata_oneof\" oneof. "
                "Only one field per oneof should be present.",
            "allOf": [{
                "$ref":
                    f"#/components/schemas/{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}",
            },],
        },
        self._GetParamSchema("/metadata_test/method7", "get", "oneofSimplemsg"))

    # Check the description of the `protobuf.oneof` from the `requestBody` field
    # of the `Operation Object` associated with `POST /metadata-test/method7`.
    # Check the `oneof_int64` inner field of the `metadata_oneof`.
    self.assertEqual(
        {
            "description":
                "This field is part of the \"metadata_oneof\" oneof. "
                "Only one field per oneof should be present.",
            "allOf": [{
                "$ref": "#/components/schemas/protobuf2.TYPE_INT64",
            },],
        }, self._GetParamSchema("/metadata_test/method7", "post", "oneofInt64"))
    # Check the `oneof_simplemsg` inner field of the `metadata_oneof`.
    self.assertEqual(
        {
            "description":
                "This field is part of the \"metadata_oneof\" oneof. "
                "Only one field per oneof should be present.",
            "allOf": [{
                "$ref":
                    f"#/components/schemas/{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}",
            },],
        },
        self._GetParamSchema("/metadata_test/method7", "post",
                             "oneofSimplemsg"))

  def testProtobufMapIsDescribedCorrectlyInOpenApiDescription(self):
    # The semantic of `protobuf.map` is partially supported by the OpenAPI
    # Specification, as OAS supports only `string` for the keys' type, while
    # `protobuf.map`s key types can be any of a set of primitive types [1].
    #
    # [1]: https://developers.google.com/protocol-buffers/docs/proto#maps

    # Firstly, check the map schema definition from the `Components Object`.
    openapi_map_type_schema = (
        self.openapi_desc_dict.get("components").get("schemas").get(
            f"{tests_pb2.MetadataMapMessage.DESCRIPTOR.full_name}.FieldMapMap_"
            f"protobuf2.TYPE_SFIXED64:{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}"
        ))
    self.assertEqual(
        {
            "description":
                f"This is a map with real key "
                f"type=\"protobuf2.TYPE_SFIXED64\" and value "
                f"type=\"{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}\"",
            "type": "object",
            "additionalProperties": {
                "$ref":
                    f"#/components/schemas/{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}",
            },
        }, openapi_map_type_schema)

    # Secondly, check the description of the `field_map` route parameter.
    # Check the description of `field_map` from the `parameters` field of the
    # `Operation Object` associated with `GET /metadata-test/method8`.
    self.assertEqual(
        {
            "description":
                f"This is a map with real key "
                f"type=\"protobuf2.TYPE_SFIXED64\" and value "
                f"type=\"{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}\"",
            "allOf": [{
                "$ref":
                    f"#/components/schemas/{tests_pb2.MetadataMapMessage.DESCRIPTOR.full_name}.FieldMapMap_"
                    f"protobuf2.TYPE_SFIXED64:{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}",
            },],
        }, self._GetParamSchema("/metadata_test/method8", "get", "fieldMap"))

    # Check the description of `field_map` from the `requestBody` field of the
    # `Operation Object` associated with `POST /metadata-test/method8`.
    self.assertEqual(
        {
            "description":
                f"This is a map with real key "
                f"type=\"protobuf2.TYPE_SFIXED64\" and value "
                f"type=\"{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}\"",
            "allOf": [{
                "$ref":
                    f"#/components/schemas/{tests_pb2.MetadataMapMessage.DESCRIPTOR.full_name}.FieldMapMap_"
                    f"protobuf2.TYPE_SFIXED64:{tests_pb2.MetadataSimpleMessage.DESCRIPTOR.full_name}",
            },],
        }, self._GetParamSchema("/metadata_test/method8", "post", "fieldMap"))

  def testOptionalPathParamsAreCorrectlyDescribedInOpenApiDescription(self):
    # This test verifies that path arguments are marked correctly as optional or
    # required.
    # The fact that the API routes that point out optional path arguments are
    # grouped under the same route described in the OpenAPI description is
    # tested by `testAllRoutesAreInOpenApiDescription`.

    # Test `GET /metadata_test/method9/{metadataId}` parameters.
    self.assertCountEqual([
        {
            "name": "metadataId",
            "in": "path",
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_STRING",
            },
        },
        {
            "name": "metadataArg1",
            "in": "query",
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_INT64",
            },
        },
        {
            "name": "metadataArg2",
            "in": "query",
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_BOOL",
            },
        },
    ], [
        self._GetParamDescription("/metadata_test/method9/{metadataId}", "get",
                                  "metadataId"),
        self._GetParamDescription("/metadata_test/method9/{metadataId}", "get",
                                  "metadataArg1"),
        self._GetParamDescription("/metadata_test/method9/{metadataId}", "get",
                                  "metadataArg2"),
    ])

    # Test `GET /metadata_test/method9/{metadataId}/fixed1/{metadataArg1}`
    # parameters.
    self.assertCountEqual([
        {
            "name": "metadataId",
            "in": "path",
            "required": True,
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_STRING",
            },
        },
        {
            "name": "metadataArg1",
            "in": "path",
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_INT64",
            },
        },
        {
            "name": "metadataArg2",
            "in": "query",
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_BOOL",
            },
        },
    ], [
        self._GetParamDescription(
            "/metadata_test/method9/{metadataId}/fixed1/{metadataArg1}", "get",
            "metadataId"),
        self._GetParamDescription(
            "/metadata_test/method9/{metadataId}/fixed1/{metadataArg1}", "get",
            "metadataArg1"),
        self._GetParamDescription(
            "/metadata_test/method9/{metadataId}/fixed1/{metadataArg1}", "get",
            "metadataArg2"),
    ])

    # Test
    # `GET /metadata_test/method9/{metadataId}/{metadataArg1}/{metadataArg2}`
    # parameters.
    self.assertCountEqual([
        {
            "name": "metadataId",
            "in": "path",
            "required": True,
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_STRING",
            },
        },
        {
            "name": "metadataArg1",
            "in": "path",
            "required": True,
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_INT64",
            },
        },
        {
            "name": "metadataArg2",
            "in": "path",
            "required": True,
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_BOOL",
            },
        },
    ], [
        self._GetParamDescription(
            "/metadata_test"
            "/method9/{metadataId}/{metadataArg1}/{metadataArg2}", "get",
            "metadataId"),
        self._GetParamDescription(
            "/metadata_test"
            "/method9/{metadataId}/{metadataArg1}/{metadataArg2}", "get",
            "metadataArg1"),
        self._GetParamDescription(
            "/metadata_test"
            "/method9/{metadataId}/{metadataArg1}/{metadataArg2}", "get",
            "metadataArg2"),
    ])

    # Test `POST /metadata_test/method9/{metadataId}/{metadataArg1}`
    # parameters.
    self.assertCountEqual([
        {
            "name": "metadataId",
            "in": "path",
            "required": True,
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_STRING",
            },
        },
        {
            "name": "metadataArg1",
            "in": "path",
            "required": True,
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_INT64",
            },
        },
    ], [
        self._GetParamDescription(
            "/metadata_test/method9/{metadataId}/{metadataArg1}", "post",
            "metadataId"),
        self._GetParamDescription(
            "/metadata_test/method9/{metadataId}/{metadataArg1}", "post",
            "metadataArg1"),
    ])

    # Test `GET /metadata_test/method9/fixed2/{metadataArg1}/{metadataArg2}`
    # parameters.
    self.assertCountEqual([
        {
            "name": "metadataId",
            "in": "query",
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_STRING",
            },
        },
        {
            "name": "metadataArg1",
            "in": "path",
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_INT64",
            },
        },
        {
            "name": "metadataArg2",
            "in": "path",
            "schema": {
                "$ref": "#/components/schemas/protobuf2.TYPE_BOOL",
            },
        },
    ], [
        self._GetParamDescription(
            "/metadata_test/method9/fixed2/{metadataArg1}/{metadataArg2}",
            "get", "metadataId"),
        self._GetParamDescription(
            "/metadata_test/method9/fixed2/{metadataArg1}/{metadataArg2}",
            "get", "metadataArg1"),
        self._GetParamDescription(
            "/metadata_test/method9/fixed2/{metadataArg1}/{metadataArg2}",
            "get", "metadataArg2"),
    ])

  def _GetParamDescription(self, method_path, http_method, param_name):
    params = (
        self.openapi_desc_dict.get("paths").get(method_path).get(
            http_method).get("parameters"))

    for param in params:
      if param["name"] == param_name:
        return param

    return None

  def _GetParamSchema(self, method_path, http_method, param_name):
    param_description = self._GetParamDescription(method_path, http_method,
                                                  param_name)

    if param_description is not None:
      return param_description["schema"]

    if http_method == "post":
      # Try finding the param in the `requestBody`.
      return (self.openapi_desc_dict.get("paths").get(method_path).get(
          http_method).get("requestBody").get("content").get("application/json")
              .get("schema").get("properties").get(param_name))

    return None

  def testRDFTypesAreCorrectlyDescribedAndUsedInOpenApiDescription(self):
    # First, check that fields which have a `sem_type` protobuf field option are
    # described using the `sem_type.type`'s schema.
    expected_field_datetime = {
        "description": "RDF type is `RDFDatetime` and it represents the number "
                       "of microseconds since epoch to a timestamp.",
        "allOf": [{
            "$ref": "#/components/schemas/RDFDatetime",
        },],
    }
    self.assertEqual(
        expected_field_datetime,
        self._GetParamSchema("/metadata_test/method10", "get", "fieldDatetime"))
    self.assertEqual(
        expected_field_datetime,
        self._GetParamSchema("/metadata_test/method10", "post",
                             "fieldDatetime"))

    expected_field_datetimeseconds = {
        "description": "RDF type is `RDFDatetimeSeconds` and it represents the "
                       "number of seconds since epoch to a timestamp.",
        "allOf": [{
            "$ref": "#/components/schemas/RDFDatetimeSeconds",
        },],
    }
    self.assertEqual(
        expected_field_datetimeseconds,
        self._GetParamSchema("/metadata_test/method10", "get",
                             "fieldDatetimeseconds"))
    self.assertEqual(
        expected_field_datetimeseconds,
        self._GetParamSchema("/metadata_test/method10", "post",
                             "fieldDatetimeseconds"))

    expected_field_duration = {
        "description": "RDF type is `Duration` and it represents the number of "
                       "microseconds between two timestamps.",
        "allOf": [{
            "$ref": "#/components/schemas/Duration",
        },],
    }
    self.assertEqual(
        expected_field_duration,
        self._GetParamSchema("/metadata_test/method10", "get",
                             "fieldDurationmicros"))
    self.assertEqual(
        expected_field_duration,
        self._GetParamSchema("/metadata_test/method10", "post",
                             "fieldDurationmicros"))

    expected_field_durationseconds = {
        "description": "RDF type is `DurationSeconds` and it represents the "
                       "number of seconds between two timestamps.",
        "allOf": [{
            "$ref": "#/components/schemas/DurationSeconds",
        },],
    }
    self.assertEqual(
        expected_field_durationseconds,
        self._GetParamSchema("/metadata_test/method10", "get",
                             "fieldDurationseconds"))
    self.assertEqual(
        expected_field_durationseconds,
        self._GetParamSchema("/metadata_test/method10", "post",
                             "fieldDurationseconds"))

    expected_field_rdfbytes = {
        "description": "RDF type is `RDFBytes` and it represents a buffer of "
                       "bytes.",
        "allOf": [{
            "$ref": "#/components/schemas/RDFBytes",
        },],
    }
    self.assertEqual(
        expected_field_rdfbytes,
        self._GetParamSchema("/metadata_test/method10", "get", "fieldRdfbytes"))
    self.assertEqual(
        expected_field_rdfbytes,
        self._GetParamSchema("/metadata_test/method10", "post",
                             "fieldRdfbytes"))

    expected_field_hashdigest = {
        "description":
            "RDF type is `HashDigest` and it represents a binary hash "
            "digest with hex string representation.",
        "allOf": [{
            "$ref": "#/components/schemas/HashDigest",
        },],
    }
    self.assertEqual(
        expected_field_hashdigest,
        self._GetParamSchema("/metadata_test/method10", "get",
                             "fieldHashdigest"))
    self.assertEqual(
        expected_field_hashdigest,
        self._GetParamSchema("/metadata_test/method10", "post",
                             "fieldHashdigest"))

    expected_field_globexpression = {
        "description": "RDF type is `GlobExpression` and it represents a glob "
                       "expression for a client path.",
        "allOf": [{
            "$ref": "#/components/schemas/GlobExpression",
        },],
    }
    self.assertEqual(
        expected_field_globexpression,
        self._GetParamSchema("/metadata_test/method10", "get",
                             "fieldGlobexpression"))
    self.assertEqual(
        expected_field_globexpression,
        self._GetParamSchema("/metadata_test/method10", "post",
                             "fieldGlobexpression"))

    expected_field_bytesize = {
        "description": "RDF type is `ByteSize` and it represents a size for "
                       "bytes allowing standard unit prefixes.",
        "allOf": [{
            "$ref": "#/components/schemas/ByteSize",
        },],
    }
    self.assertEqual(
        expected_field_bytesize,
        self._GetParamSchema("/metadata_test/method10", "get", "fieldBytesize"))
    self.assertEqual(
        expected_field_bytesize,
        self._GetParamSchema("/metadata_test/method10", "post",
                             "fieldBytesize"))

    expected_field_rdfurn = {
        "description": "RDF type is `RDFURN` and it represents an object to "
                       "abstract URL manipulation.",
        "allOf": [{
            "$ref": "#/components/schemas/RDFURN",
        },],
    }
    self.assertEqual(
        expected_field_rdfurn,
        self._GetParamSchema("/metadata_test/method10", "get", "fieldRdfurn"))
    self.assertEqual(
        expected_field_rdfurn,
        self._GetParamSchema("/metadata_test/method10", "post", "fieldRdfurn"))

    expected_field_sessionid = {
        "description": "RDF type is `SessionID` and it represents an rdfvalue "
                       "object that represents a session_id.",
        "allOf": [{
            "$ref": "#/components/schemas/SessionID",
        },],
    }
    self.assertEqual(
        expected_field_sessionid,
        self._GetParamSchema("/metadata_test/method10", "get",
                             "fieldSessionid"))
    self.assertEqual(
        expected_field_sessionid,
        self._GetParamSchema("/metadata_test/method10", "post",
                             "fieldSessionid"))

    # Check that descriptions get concatenated in the case we have a
    # `protobuf.oneof` field that has an RDF type `sem_type.type`.
    expected_oneof_datetime = {
        "description":
            "This field is part of the \"semtype_oneof\" oneof. Only "
            "one field per oneof should be present. RDF type is "
            "`RDFDatetime` and it represents the number of "
            "microseconds since epoch to a timestamp.",
        "allOf": [{
            "$ref": "#/components/schemas/RDFDatetime",
        },],
    }
    self.assertEqual(
        expected_oneof_datetime,
        self._GetParamSchema("/metadata_test/method10", "get", "oneofDatetime"))
    self.assertEqual(
        expected_oneof_datetime,
        self._GetParamSchema("/metadata_test/method10", "post",
                             "oneofDatetime"))

    # Now check that the RDF types have their schemas correctly described.
    component_schemas = self.openapi_desc_dict["components"]["schemas"]

    self.assertEqual(
        {
            "type":
                "string",
            "format":
                "uint64",
            "description":
                "the number of microseconds since epoch to a timestamp",
        }, component_schemas["RDFDatetime"])
    self.assertEqual(
        {
            "type": "string",
            "format": "uint64",
            "description": "the number of seconds since epoch to a timestamp",
        }, component_schemas["RDFDatetimeSeconds"])
    self.assertEqual(
        {
            "type": "string",
            "format": "uint64",
            "description": "the number of microseconds between two timestamps",
        }, component_schemas["Duration"])
    self.assertEqual(
        {
            "type": "string",
            "format": "uint64",
            "description": "the number of seconds between two timestamps",
        }, component_schemas["DurationSeconds"])
    self.assertEqual(
        {
            "type": "string",
            "format": "byte",
            "description": "a buffer of bytes",
        }, component_schemas["RDFBytes"])
    self.assertEqual(
        {
            "type":
                "string",
            "format":
                "byte",
            "description":
                "a binary hash digest with hex string representation",
        }, component_schemas["HashDigest"])
    self.assertEqual(
        {
            "type": "string",
            "description": "a glob expression for a client path",
        }, component_schemas["GlobExpression"])
    self.assertEqual(
        {
            "type": "string",
            "format": "uint64",
            "description": "a size for bytes allowing standard unit prefixes",
        }, component_schemas["ByteSize"])
    self.assertEqual(
        {
            "type": "string",
            "description": "an object to abstract URL manipulation",
        }, component_schemas["RDFURN"])
    self.assertEqual(
        {
            "type": "string",
            "description": "an rdfvalue object that represents a session_id",
        }, component_schemas["SessionID"])


def main(argv):
  test_lib.main(argv)


if __name__ == "__main__":
  app.run(main)
