264 lines
9.6 KiB
Python
264 lines
9.6 KiB
Python
import re
|
|
import requests
|
|
from http import HTTPStatus
|
|
import json as JSON
|
|
|
|
from infra.config import BaseUri, Settings
|
|
from utils.logger import logger, bcolors
|
|
from utils.singleton import SingletonMeta
|
|
|
|
swagger_api_json_endpoint = '/api-json'
|
|
api_info_urls = {
|
|
BaseUri.Iguana: 'Iguana',
|
|
BaseUri.Pyrador: 'Pyrador',
|
|
BaseUri.Zoo: 'Zoo'
|
|
}
|
|
excluded_endpoints = [
|
|
('POST', BaseUri.Iguana + '/api/v1/test/controller'),
|
|
('GET', BaseUri.Iguana + '/api/v1/test/controller'),
|
|
('DELETE', BaseUri.Iguana + '/api/v1/test/controller'),
|
|
('GET', BaseUri.Iguana + '/api/v1/health'),
|
|
('GET', BaseUri.Iguana + '/api/v1/metrics'),
|
|
('GET', BaseUri.Iguana + '/api/v1/settings'),
|
|
('POST', BaseUri.Iguana + '/api/v1/settings'),
|
|
('PUT', BaseUri.Iguana + '/api/v1/settings'),
|
|
('GET', BaseUri.Iguana + '/api/v1/activity'),
|
|
('GET', BaseUri.Iguana + '/api/v1/activity/{activity_id}'),
|
|
('POST', BaseUri.Iguana + '/api/v1/doorlock'),
|
|
('PUT', BaseUri.Iguana + '/api/v1/profile/set-account-number'),
|
|
('PUT', BaseUri.Iguana + '/api/v1/profile/address'),
|
|
('PUT', BaseUri.Iguana + '/api/v1/profile/contact'),
|
|
('POST', BaseUri.Iguana + '/api/v1/profile/set-firebase-token'),
|
|
('PUT', BaseUri.Iguana + '/api/v1/profile/balance'),
|
|
('GET', BaseUri.Iguana + '/api/v1/providable-service/{place_id}'),
|
|
('POST', BaseUri.Iguana + '/api/v1/light-device/toggle'),
|
|
('GET', BaseUri.Iguana + '/api/v1/light-device/state/{device_id}'),
|
|
('PUT', BaseUri.Iguana + '/api/v1/user-place/{place_id}'),
|
|
('GET', BaseUri.Iguana + '/api/v1/user-place/{place_id}/services'),
|
|
('PUT', BaseUri.Iguana + '/api/v1/user-place/set/status'),
|
|
('POST', BaseUri.Iguana + '/api/v1/profile/device/to/service'),
|
|
('DELETE', BaseUri.Iguana + '/api/v1/profile/device/from/service'),
|
|
('GET', BaseUri.Iguana + '/api/v1/profile/place/{place_id}/service/devices/{device_category}'),
|
|
('POST', BaseUri.Iguana + '/api/v1/room'),
|
|
('GET', BaseUri.Iguana + '/api/v1/room/by/place/{parent_id}'),
|
|
('PUT', BaseUri.Iguana + '/api/v1/room/{id}'),
|
|
('DELETE', BaseUri.Iguana + '/api/v1/room/{id}'),
|
|
('GET', BaseUri.Iguana + '/api/v1/device/list/{type}'),
|
|
('DELETE', BaseUri.Iguana + '/api/v1/user-place/qrcode'),
|
|
('POST', BaseUri.Iguana + '/api/v1/billing'),
|
|
('POST', BaseUri.Iguana + '/api/v1/intercom/acceptCall'), # TODO: test it with notifications
|
|
('POST', BaseUri.Iguana + '/api/v1/upload/avatar'), # TODO: unable to test
|
|
('POST', BaseUri.Zoo + '/api/v1/notifications/send-notification'),
|
|
('POST', BaseUri.Zoo + '/api/v1/notifications/send-sms'),
|
|
('DELETE', BaseUri.Zoo + '/api/v1/place/available_services')
|
|
]
|
|
|
|
class APICoverageTracker(metaclass=SingletonMeta):
|
|
def __init__(self):
|
|
self.called_endpoints = {}
|
|
self.api_info = self.request_api_info(api_info_urls)
|
|
self.build_called_endpoints()
|
|
|
|
def request_api_info(self, urls):
|
|
api_info = {}
|
|
|
|
for url in urls:
|
|
res = requests.get(url + swagger_api_json_endpoint)
|
|
api_info[url] = res.json()
|
|
|
|
return api_info
|
|
|
|
def build_called_endpoints(self):
|
|
for url, info in self.api_info.items():
|
|
try:
|
|
paths = info.get('paths')
|
|
if not url in self.called_endpoints:
|
|
self.called_endpoints[url] = {}
|
|
for path, methods in paths.items():
|
|
endpoint = url + path
|
|
self.called_endpoints[url][path] = {}
|
|
for method, method_info in methods.items():
|
|
if (method.upper(), endpoint) in excluded_endpoints:
|
|
continue
|
|
self.called_endpoints[url][path][method] = 0
|
|
except Exception as e:
|
|
logger.error('Error happened while getting api info:', e)
|
|
|
|
def endpoint_is_called(self, called_endpoint, method):
|
|
if not Settings.EnableCoverageStatistics:
|
|
return
|
|
|
|
for url, paths in self.called_endpoints.items():
|
|
for path, methods in paths.items():
|
|
endpoint = url + path
|
|
pattern = re.sub(r'{.+?}', r'[^/]+', endpoint) + '$'
|
|
if re.match(pattern, called_endpoint) and method.lower() in methods:
|
|
self.called_endpoints[url][path][method.lower()] += 1
|
|
return
|
|
|
|
def print_coverage(self):
|
|
def calculate_coverage_statistics(total_urls, covered_urls):
|
|
if total_urls == 0:
|
|
return 0
|
|
coverage_percentage = int(covered_urls / total_urls * 100)
|
|
|
|
if coverage_percentage < 50:
|
|
color = bcolors.FAIL
|
|
elif coverage_percentage < 75:
|
|
color = bcolors.WARNING
|
|
else:
|
|
color = bcolors.OKGREEN
|
|
|
|
statistics = f'{coverage_percentage}% ({covered_urls} / {total_urls})'
|
|
return f'{color}{statistics}{bcolors.ENDC}'
|
|
|
|
def count_urls(gateway_url):
|
|
urls_num = 0
|
|
covered_urls_num = 0
|
|
|
|
for url, paths in self.called_endpoints.items():
|
|
for path, methods in paths.items():
|
|
endpoint = url + path
|
|
if gateway_url in endpoint:
|
|
for method, num_of_calls in methods.items():
|
|
urls_num += 1
|
|
if num_of_calls > 0:
|
|
covered_urls_num += 1
|
|
else:
|
|
logger.warn(f'{method.upper()} {endpoint} is not covered')
|
|
|
|
return urls_num, covered_urls_num
|
|
|
|
if not Settings.EnableCoverageStatistics:
|
|
return
|
|
|
|
urls_num_sum = 0
|
|
covered_urls_num_sum = 0
|
|
|
|
urls_info = \
|
|
[(gateway_name, count_urls(gateway_url)) \
|
|
for gateway_url, gateway_name in api_info_urls.items()]
|
|
|
|
logger.info('Coverage statistics:')
|
|
logger.info()
|
|
|
|
for gateway_name, (urls_num, covered_urls_num) in urls_info:
|
|
coverage_statistics = calculate_coverage_statistics(urls_num, covered_urls_num)
|
|
message = f' {gateway_name}: {coverage_statistics}'
|
|
logger.info(message)
|
|
|
|
urls_num_sum += urls_num
|
|
covered_urls_num_sum += covered_urls_num
|
|
|
|
coverage_statistics = \
|
|
calculate_coverage_statistics(urls_num_sum, covered_urls_num_sum)
|
|
logger.info()
|
|
logger.info(f' Total: {coverage_statistics}\n')
|
|
|
|
class Response(requests.Response):
|
|
def __init__(self, status_code=HTTPStatus.OK):
|
|
super().__init__()
|
|
self.status_code = status_code
|
|
|
|
def log_req(method, url, params=None, data=None, json=None, headers=None):
|
|
logger.verbose(f'============================================================')
|
|
logger.verbose(f'[REQUEST] {method} {url}')
|
|
if params:
|
|
logger.verbose(f'params: {params}')
|
|
if data:
|
|
data = JSON.dumps(data, sort_keys=True, indent=4)
|
|
logger.verbose(f'data: {data}')
|
|
if json:
|
|
json = JSON.dumps(json, sort_keys=True, indent=4)
|
|
logger.verbose(f'json: {json}')
|
|
if headers:
|
|
headers = JSON.dumps(headers, sort_keys=True, indent=4)
|
|
logger.verbose(f'headers: {headers}')
|
|
|
|
def log_res(res: requests.Response):
|
|
req = res.request
|
|
logger.verbose(f'[RESPONSE] {req.method} {req.url} {res.status_code}')
|
|
try:
|
|
json = JSON.dumps(res.json(), sort_keys=True, indent=4).replace('\\"', '"')
|
|
lines_num = json.count('\n')
|
|
max_lines_num = Settings.LoggingResponseMaxLinesNum
|
|
if lines_num <= max_lines_num:
|
|
logger.verbose(f'json: {json}')
|
|
else:
|
|
stats = f'{lines_num}/{max_lines_num}'
|
|
logger.verbose(f'Maximum number of lines for response exceeded:', stats)
|
|
except ValueError:
|
|
logger.verbose('response:', res.content)
|
|
except Exception as e:
|
|
logger.verbose(e)
|
|
|
|
def request(method, url, headers=None, **kwargs):
|
|
APICoverageTracker().endpoint_is_called(url, method)
|
|
log_req(method, url, params=kwargs.get('params'), \
|
|
data=kwargs.get('data'), json=kwargs.get('json'), headers=headers)
|
|
res = requests.request(method, url, **kwargs)
|
|
log_res(res)
|
|
return res
|
|
|
|
def get(url, params=None, headers=None, **kwargs):
|
|
method = 'GET'
|
|
APICoverageTracker().endpoint_is_called(url, method)
|
|
log_req(method, url, params=params, \
|
|
data=kwargs.get('data'), json=kwargs.get('json'), headers=headers)
|
|
res = requests.get(url, params=params, headers=headers, **kwargs)
|
|
log_res(res)
|
|
return res
|
|
|
|
def options(url, headers=None, **kwargs):
|
|
method = 'OPTIONS'
|
|
APICoverageTracker().endpoint_is_called(url, method)
|
|
log_req(method, url, params=kwargs.get('params'), \
|
|
data=kwargs.get('data'), json=kwargs.get('json'), headers=headers)
|
|
res = requests.options(url, headers=headers, **kwargs)
|
|
log_res(res)
|
|
return res
|
|
|
|
def head(url, headers=None, **kwargs):
|
|
method = 'HEAD'
|
|
APICoverageTracker().endpoint_is_called(url, method)
|
|
log_req(method, url, params=kwargs.get('params'), \
|
|
data=kwargs.get('data'), json=kwargs.get('json'), headers=headers)
|
|
res = requests.head(url, headers=headers, **kwargs)
|
|
log_res(res)
|
|
return res
|
|
|
|
def post(url, data=None, json=None, headers=None, **kwargs):
|
|
method = 'POST'
|
|
APICoverageTracker().endpoint_is_called(url, method)
|
|
log_req(method, url, params=kwargs.get('params'), \
|
|
data=data, json=json, headers=headers)
|
|
res = requests.post(url, data=data, json=json, headers=headers, **kwargs)
|
|
log_res(res)
|
|
return res
|
|
|
|
def put(url, data=None, headers=None, **kwargs):
|
|
method = 'PUT'
|
|
APICoverageTracker().endpoint_is_called(url, method)
|
|
log_req(method, url, params=kwargs.get('params'), \
|
|
data=data, json=kwargs.get('json'), headers=headers),
|
|
res = requests.put(url, data=data, headers=headers, **kwargs)
|
|
log_res(res)
|
|
return res
|
|
|
|
def patch(url, data=None, headers=None, **kwargs):
|
|
method = 'PATCH'
|
|
APICoverageTracker().endpoint_is_called(url, method)
|
|
log_req(method, url, params=kwargs.get('params'), \
|
|
data=data, json=kwargs.get('json'), headers=headers)
|
|
res = requests.patch(url, data=data, headers=headers, **kwargs)
|
|
log_res(res)
|
|
return res
|
|
|
|
def delete(url, headers=None, **kwargs):
|
|
method = 'DELETE'
|
|
APICoverageTracker().endpoint_is_called(url, method)
|
|
log_req(method, url, params=kwargs.get('params'), \
|
|
data=kwargs.get('data'), json=kwargs.get('json'), headers=headers)
|
|
res = requests.delete(url, headers=headers, **kwargs)
|
|
log_res(res)
|
|
return res |