build.utils.validate() - Code Metrics - Inspection of "Merge pull request #129 from kytos-ng/fix/issue_12..." - ajoaoff/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( a63869...63c1d9 )

by Antonio

created 2022-01-12 15:39 UTC

build.utils.validate() B

↳ Parent: build.utils

Complexity

Conditions

Size

Total Lines	51
Code Lines	38

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	38
nop	1
dl	0
loc	51
rs	8.5013
c	0
b	0
f	0

How to fix Long Method

"""Utility functions."""
import functools
from pathlib import Path

from flask import request
from openapi_core import create_spec
from openapi_core.contrib.flask import FlaskOpenAPIRequest
from openapi_core.validation.request.validators import RequestValidator
from openapi_spec_validator import validate_spec
from openapi_spec_validator.readers import read_from_filename
from werkzeug.exceptions import BadRequest, UnsupportedMediaType

from kytos.core import log
from kytos.core.events import KytosEvent


def emit_event(controller, name, **kwargs):
    """Send an event when something happens with an EVC."""
    event_name = f"kytos/mef_eline.{name}"
    event = KytosEvent(name=event_name, content=kwargs)
    controller.buffers.app.put(event)


def compare_endpoint_trace(endpoint, vlan, trace):
    """Compare and endpoint with a trace step."""
    return (
        endpoint.switch.dpid == trace["dpid"]
        and endpoint.port_number == trace["port"]
        and vlan == trace["vlan"]
    )


def load_spec():
    """Validate openapi spec."""
    napp_dir = Path(__file__).parent
    yml_file = napp_dir / "openapi.yml"
    spec_dict, _ = read_from_filename(yml_file)

    validate_spec(spec_dict)

    return create_spec(spec_dict)


def validate(spec):
    """Decorator to validate a REST endpoint input.

    Uses the schema defined in the openapi.yml file
    to validate.
    """

    def validate_decorator(func):
        @functools.wraps(func)
        def wrapper_validate(*args, **kwargs):
            try:
                data = request.get_json()
            except BadRequest:
                result = "The request body is not a well-formed JSON."
                log.debug("create_circuit result %s %s", result, 400)
                raise BadRequest(result) from BadRequest
            if data is None:
                result = "The request body mimetype is not application/json."
                log.debug("update result %s %s", result, 415)
                raise UnsupportedMediaType(result)

            validator = RequestValidator(spec)
            openapi_request = FlaskOpenAPIRequest(request)
            result = validator.validate(openapi_request)
            if result.errors:
                errors = result.errors[0]
                if hasattr(errors, "schema_errors"):
                    schema_errors = errors.schema_errors[0]
                    error_log = {
                        "error_message": schema_errors.message,
                        "error_validator": schema_errors.validator,
                        "error_validator_value": schema_errors.validator_value,
                        "error_path": list(schema_errors.path),
                        "error_schema": schema_errors.schema,
                        "error_schema_path": list(schema_errors.schema_path),
                    }
                    log.debug("error response: %s", error_log)
                    error_response = f"{schema_errors.message} for field"
                    error_response += (
                        f" {'/'.join(map(str,schema_errors.path))}."
                    )
                else:
                    error_response = (
                        "The request body mimetype is not application/json."
                    )
                raise BadRequest(error_response) from BadRequest
            return func(*args, data=data, **kwargs)

        return wrapper_validate

    return validate_decorator


1			"""Utility functions."""
2			import functools
3			from pathlib import Path
4
5			from flask import request
6			from openapi_core import create_spec
7			from openapi_core.contrib.flask import FlaskOpenAPIRequest
8			from openapi_core.validation.request.validators import RequestValidator
9			from openapi_spec_validator import validate_spec
10			from openapi_spec_validator.readers import read_from_filename
11			from werkzeug.exceptions import BadRequest, UnsupportedMediaType
12
13			from kytos.core import log
14			from kytos.core.events import KytosEvent
15
16
17			def emit_event(controller, name, **kwargs):
18			"""Send an event when something happens with an EVC."""
19			event_name = f"kytos/mef_eline.{name}"
20			event = KytosEvent(name=event_name, content=kwargs)
21			controller.buffers.app.put(event)
22
23
24			def compare_endpoint_trace(endpoint, vlan, trace):
25			"""Compare and endpoint with a trace step."""
26			return (
27			endpoint.switch.dpid == trace["dpid"]
28			and endpoint.port_number == trace["port"]
29			and vlan == trace["vlan"]
30			)
31
32
33			def load_spec():
34			"""Validate openapi spec."""
35			napp_dir = Path(__file__).parent
36			yml_file = napp_dir / "openapi.yml"
37			spec_dict, _ = read_from_filename(yml_file)
38
39			validate_spec(spec_dict)
40
41			return create_spec(spec_dict)
42
43
44			def validate(spec):
45			"""Decorator to validate a REST endpoint input.
46
47			Uses the schema defined in the openapi.yml file
48			to validate.
49			"""
50
51			def validate_decorator(func):
52			@functools.wraps(func)
53			def wrapper_validate(args, *kwargs):
54			try:
55			data = request.get_json()
56			except BadRequest:
57			result = "The request body is not a well-formed JSON."
58			log.debug("create_circuit result %s %s", result, 400)
59			raise BadRequest(result) from BadRequest
60			if data is None:
61			result = "The request body mimetype is not application/json."
62			log.debug("update result %s %s", result, 415)
63			raise UnsupportedMediaType(result)
64
65			validator = RequestValidator(spec)
66			openapi_request = FlaskOpenAPIRequest(request)
67			result = validator.validate(openapi_request)
68			if result.errors:
69			errors = result.errors[0]
70			if hasattr(errors, "schema_errors"):
71			schema_errors = errors.schema_errors[0]
72			error_log = {
73			"error_message": schema_errors.message,
74			"error_validator": schema_errors.validator,
75			"error_validator_value": schema_errors.validator_value,
76			"error_path": list(schema_errors.path),
77			"error_schema": schema_errors.schema,
78			"error_schema_path": list(schema_errors.schema_path),
79			}
80			log.debug("error response: %s", error_log)
81			error_response = f"{schema_errors.message} for field"
82			error_response += (
83			f" {'/'.join(map(str,schema_errors.path))}."
84			)
85			else:
86			error_response = (
87			"The request body mimetype is not application/json."
88			)
89			raise BadRequest(error_response) from BadRequest
90			return func(args, data=data, *kwargs)
91
92			return wrapper_validate
93
94			return validate_decorator
95

ajoaoff / mef_eline

Push — master ( a63869...63c1d9 )

build.utils.validate() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like