9906372057

Committed 12 Jul 2024 10:23AM UTC coverage: 81.219% (+0.3%) from 80.937%

Build # 9906372057

Build Type

push

github

Committed by

mvalik

Commit Message

New tests for _authorization_warning and verify_authorization

Run Details

312 of 410 branches covered (76.1%)

Branch coverage included in aggregate %.

3 of 3 new or added lines in 2 files covered. (100.0%)

46 existing lines in 5 files now uncovered.

981 of 1182 relevant lines covered (82.99%)

0.83 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.18

/waiverdb/utils.py

# SPDX-License-Identifier: GPL-2.0+

import functools
import stomp
from flask import request, url_for, jsonify, current_app, Flask
from flask_restx import marshal
from flask_pydantic.exceptions import ValidationError
from waiverdb.fields import waiver_fields
from werkzeug.exceptions import BadRequest, NotFound, HTTPException
from contextlib import contextmanager

VALIDATION_KEYS = frozenset({
    "input", "loc", "msg", "type", "url"
})


def json_collection(query, page=1, limit=10):
    """
    Helper function for Flask request handlers which want to return
    a collection of resources as JSON.
    """
    try:
        p = query.paginate(page=page, per_page=limit)
    except NotFound:
        return {'data': [], 'prev': None, 'next': None, 'first': None, 'last': None}
    pages = {'data': marshal(p.items, waiver_fields)}
    query_pairs = request.args.copy()
    if query_pairs:
        # remove the page number
        query_pairs.pop('page', default=None)
    if p.has_prev:
        pages['prev'] = url_for(request.endpoint, page=p.prev_num, _external=True,
                                **query_pairs)
    else:
        pages['prev'] = None
    if p.has_next:
        pages['next'] = url_for(request.endpoint, page=p.next_num, _external=True,
                                **query_pairs)
    else:
        pages['next'] = None
    pages['first'] = url_for(request.endpoint, page=1, _external=True, **query_pairs)
    pages['last'] = url_for(request.endpoint, page=p.pages, _external=True, **query_pairs)
    return pages


def json_error(error):
    """
    Return error responses in JSON.

    :param error: One of Exceptions. It could be HTTPException, ConnectionError, or
    Timeout.
    :return: JSON error response.

    """
    if not isinstance(error, (BadRequest, NotFound)):
        current_app.logger.warning("%s: %s", type(error).__name__, error)

    if isinstance(error, HTTPException):
        response = jsonify(message=error.description)
        if error.code is not None:
            response.status_code = error.code
    else:
        # Could be ConnectionError or Timeout
        response = jsonify(message=str(error))
        response.status_code = 500

    return response


def handle_validation_error(error: ValidationError):
    errors = (
        error.body_params
        or error.form_params
        or error.path_params
        or error.query_params
    )
    # Keep only interesting stuff and remove objects potentially
    # unserializable in JSON.
    err = [
        {k: v for k, v in e.items() if k in VALIDATION_KEYS}
        for e in errors
    ]
    response = jsonify({"validation_error": err})
    response.status_code = 400
    return response


def jsonp(func):
    """Wraps Jsonified output for JSONP requests."""
    @functools.wraps(func)
    def wrapped(*args, **kwargs):
        callback = request.args.get('callback', False)
        if callback:
            resp = jsonify(func(*args, **kwargs))
            resp.set_data('{}({})'.format(
                str(callback),
                resp.get_data()
            ))
            resp.mimetype = 'application/javascript'
            return resp
        else:
            return func(*args, **kwargs)
    return wrapped


@contextmanager
def stomp_connection():
    """
    Helper function for stomp connection.
    """
    if current_app.config.get('STOMP_CONFIGS'):
        configs = current_app.config.get('STOMP_CONFIGS')
        if 'destination' not in configs or not configs['destination']:
            raise RuntimeError('stomp was configured to publish messages, '
                               'but destination is not configured in STOMP_CONFIGS')
        if 'connection' not in configs or not configs['connection']:
            raise RuntimeError('stomp was configured to publish messages,, '
                               'but connection is not configured in STOMP_CONFIGS')

        conn_args = configs['connection'].copy()
        if 'use_ssl' in conn_args:
            use_ssl = conn_args['use_ssl']
            del conn_args['use_ssl']
        else:
            use_ssl = False

        ssl_args = {'for_hosts': conn_args['host_and_ports']}
        for attr in ('key_file', 'cert_file', 'ca_certs'):
            conn_attr = f'ssl_{attr}'
            if conn_attr in conn_args:
                ssl_args[attr] = conn_args[conn_attr]
                del conn_args[conn_attr]

        conn = stomp.connect.StompConnection11(**conn_args)

        if use_ssl:
            conn.set_ssl(**ssl_args)

        conn.connect(wait=True, **configs.get('credentials', {}))

        try:
            yield conn
        finally:
            conn.disconnect()
    else:
        raise RuntimeError('stomp was configured to publish messages, '
                           'but STOMP_CONFIGS is not configured')


def auth_methods(app: Flask) -> list[str]:
    methods = app.config.get('AUTH_METHODS')
    if methods:
        return methods

    method = app.config.get('AUTH_METHOD')
    if method:
        return [method]

    return []

1	# SPDX-License-Identifier: GPL-2.0+
2
3	import functools	1✔
4	import stomp	1✔
5	from flask import request, url_for, jsonify, current_app, Flask	1✔
6	from flask_restx import marshal	1✔
7	from flask_pydantic.exceptions import ValidationError	1✔
8	from waiverdb.fields import waiver_fields	1✔
9	from werkzeug.exceptions import BadRequest, NotFound, HTTPException	1✔
10	from contextlib import contextmanager	1✔
11
12	VALIDATION_KEYS = frozenset({	1✔
13	"input", "loc", "msg", "type", "url"
14	})
15
16
17	def json_collection(query, page=1, limit=10):	1✔
18	"""
19	Helper function for Flask request handlers which want to return
20	a collection of resources as JSON.
21	"""
22	try:	1✔
23	p = query.paginate(page=page, per_page=limit)	1✔
24	except NotFound:	×
25	return {'data': [], 'prev': None, 'next': None, 'first': None, 'last': None}	×
26	pages = {'data': marshal(p.items, waiver_fields)}	1✔
27	query_pairs = request.args.copy()	1✔
28	if query_pairs:	1✔
29	# remove the page number
30	query_pairs.pop('page', default=None)	1✔
31	if p.has_prev:	1✔
32	pages['prev'] = url_for(request.endpoint, page=p.prev_num, _external=True,	1✔
33	**query_pairs)
34	else:
35	pages['prev'] = None	1✔
36	if p.has_next:	1✔
37	pages['next'] = url_for(request.endpoint, page=p.next_num, _external=True,	1✔
38	**query_pairs)
39	else:
40	pages['next'] = None	1✔
41	pages['first'] = url_for(request.endpoint, page=1, _external=True, **query_pairs)	1✔
42	pages['last'] = url_for(request.endpoint, page=p.pages, _external=True, **query_pairs)	1✔
43	return pages	1✔
44
45
46	def json_error(error):	1✔
47	"""
48	Return error responses in JSON.
49
50	:param error: One of Exceptions. It could be HTTPException, ConnectionError, or
51	Timeout.
52	:return: JSON error response.
53
54	"""
55	if not isinstance(error, (BadRequest, NotFound)):	1✔
56	current_app.logger.warning("%s: %s", type(error).__name__, error)	1✔
57
58	if isinstance(error, HTTPException):	1✔
59	response = jsonify(message=error.description)	1✔
60	if error.code is not None:	1!
61	response.status_code = error.code	1✔
62	else:
63	# Could be ConnectionError or Timeout
64	response = jsonify(message=str(error))	1✔
65	response.status_code = 500	1✔
66
67	return response	1✔
68
69
70	def handle_validation_error(error: ValidationError):	1✔
71	errors = (	1✔
72	error.body_params
73	or error.form_params
74	or error.path_params
75	or error.query_params
76	)
77	# Keep only interesting stuff and remove objects potentially
78	# unserializable in JSON.
79	err = [	1✔
80	{k: v for k, v in e.items() if k in VALIDATION_KEYS}
81	for e in errors
82	]
83	response = jsonify({"validation_error": err})	1✔
84	response.status_code = 400	1✔
85	return response	1✔
86
87
88	def jsonp(func):	1✔
89	"""Wraps Jsonified output for JSONP requests."""
90	@functools.wraps(func)	1✔
91	def wrapped(args, *kwargs):	1✔
92	callback = request.args.get('callback', False)	1✔
93	if callback:	1✔
94	resp = jsonify(func(args, *kwargs))	1✔
95	resp.set_data('{}({})'.format(	1✔
96	str(callback),
97	resp.get_data()
98	))
99	resp.mimetype = 'application/javascript'	1✔
100	return resp	1✔
101	else:
102	return func(args, *kwargs)	1✔
103	return wrapped	1✔
104
105
106	@contextmanager	1✔
107	def stomp_connection():	1✔
108	"""
109	Helper function for stomp connection.
110	"""
111	if current_app.config.get('STOMP_CONFIGS'):	1!
112	configs = current_app.config.get('STOMP_CONFIGS')	1✔
113	if 'destination' not in configs or not configs['destination']:	1!
114	raise RuntimeError('stomp was configured to publish messages, '	×
115	'but destination is not configured in STOMP_CONFIGS')
116	if 'connection' not in configs or not configs['connection']:	1!
UNCOV 117	raise RuntimeError('stomp was configured to publish messages,, '	×
118	'but connection is not configured in STOMP_CONFIGS')
119
120	conn_args = configs['connection'].copy()	1✔
121	if 'use_ssl' in conn_args:	1!
UNCOV 122	use_ssl = conn_args['use_ssl']	×
UNCOV 123	del conn_args['use_ssl']	×
124	else:
125	use_ssl = False	1✔
126
127	ssl_args = {'for_hosts': conn_args['host_and_ports']}	1✔
128	for attr in ('key_file', 'cert_file', 'ca_certs'):	1✔
129	conn_attr = f'ssl_{attr}'	1✔
130	if conn_attr in conn_args:	1!
UNCOV 131	ssl_args[attr] = conn_args[conn_attr]	×
UNCOV 132	del conn_args[conn_attr]	×
133
134	conn = stomp.connect.StompConnection11(**conn_args)	1✔
135
136	if use_ssl:	1!
UNCOV 137	conn.set_ssl(**ssl_args)	×
138
139	conn.connect(wait=True, **configs.get('credentials', {}))	1✔
140
141	try:	1✔
142	yield conn	1✔
143	finally:
144	conn.disconnect()	1✔
145	else:
UNCOV 146	raise RuntimeError('stomp was configured to publish messages, '	×
147	'but STOMP_CONFIGS is not configured')
148
149
150	def auth_methods(app: Flask) -> list[str]:	1✔
151	methods = app.config.get('AUTH_METHODS')	1✔
152	if methods:	1✔
153	return methods	1✔
154
155	method = app.config.get('AUTH_METHOD')	1✔
156	if method:	1✔
157	return [method]	1✔
158
159	return []	1✔

release-engineering / waiverdb / 9906372057

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous