Initial commit
This commit is contained in:
somilovanov
2
.gitignore
vendored
Normal file
2
.gitignore
vendored
Normal file
@@ -0,0 +1,2 @@
|
||||
.env
|
||||
__pycache__/
|
||||
217
test_metric.py
Normal file
217
test_metric.py
Normal file
@@ -0,0 +1,217 @@
|
||||
import requests
|
||||
from urllib3.exceptions import InsecureRequestWarning
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||||
datefmt="%Y-%m-%d %H:%M:%S",
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_TIMEOUT = 20.0
|
||||
DEFAULT_VERIFY_TLS = False
|
||||
DEFAULT_HEADERS = {
|
||||
'accept': 'application/json',
|
||||
}
|
||||
|
||||
CREDS = {
|
||||
"User": os.getenv('TEST_ORCH_API_USER'),
|
||||
"Password": os.getenv('TEST_ORCH_API_PASS'),
|
||||
}
|
||||
BASE_URL = os.getenv('BASE_URL')
|
||||
URL_API_TOKEN = f"http://{BASE_URL}:5001/api/Account"
|
||||
URL_API_GENERAL = f"https://{BASE_URL}:44392"
|
||||
|
||||
|
||||
def send_http_request(
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
headers: dict | None = None,
|
||||
params: dict | None = None,
|
||||
data: dict | str | None = None,
|
||||
json_data: dict | list | None = None,
|
||||
timeout: float = DEFAULT_TIMEOUT,
|
||||
verify: bool = DEFAULT_VERIFY_TLS,
|
||||
) -> requests.Response:
|
||||
"""Унифицированный HTTP-запрос с дефолтными настройками и логированием."""
|
||||
merged_headers = {**DEFAULT_HEADERS, **(headers or {})}
|
||||
|
||||
response = requests.request(
|
||||
method=method.upper(),
|
||||
url=url,
|
||||
headers=merged_headers,
|
||||
params=params,
|
||||
data=data,
|
||||
json=json_data,
|
||||
timeout=timeout,
|
||||
verify=verify,
|
||||
)
|
||||
logger.debug("%s %s -> %s", method.upper(), response.request.url, response.status_code)
|
||||
return response
|
||||
|
||||
|
||||
def get_token():
|
||||
logger.info("Получаем токен для пользователя %s", CREDS['User'])
|
||||
try:
|
||||
data = {
|
||||
"userName": CREDS["User"],
|
||||
"password": CREDS["Password"],
|
||||
}
|
||||
response = send_http_request(
|
||||
'POST',
|
||||
URL_API_TOKEN,
|
||||
headers={'accept': '*/*'},
|
||||
json_data=data,
|
||||
)
|
||||
response.raise_for_status()
|
||||
response_data = response.json()
|
||||
token = response_data.get('token')
|
||||
if not token:
|
||||
logger.error('Ответ получения токена не содержит поле token')
|
||||
logger.debug('response json: %s', response_data)
|
||||
raise SystemExit(1)
|
||||
return token
|
||||
except requests.HTTPError as error:
|
||||
response = error.response
|
||||
logger.error('Произошла ошибка во время получения токена: %s', error)
|
||||
if response is not None:
|
||||
logger.debug('status=%s url=%s', response.status_code, response.url)
|
||||
logger.debug('content-type=%s', response.headers.get('Content-Type'))
|
||||
logger.debug('response(first 1500): %s', response.text[:1500])
|
||||
raise SystemExit(1)
|
||||
except requests.RequestException as error:
|
||||
logger.error('Сетевая ошибка во время получения токена: %s', error)
|
||||
raise SystemExit(1)
|
||||
except ValueError as error:
|
||||
logger.error('Некорректный JSON при получении токена: %s', error)
|
||||
raise SystemExit(1)
|
||||
except Exception as error:
|
||||
logger.error('Неожиданная ошибка во время получения токена: %s', error)
|
||||
raise SystemExit(1)
|
||||
|
||||
def get_queue(auth_token: str, event_date: str | None = None):
|
||||
postfix = "/api/RpaProjectQueue/All"
|
||||
URL = URL_API_GENERAL + postfix
|
||||
try:
|
||||
params = {'eventDate': event_date} if event_date else None
|
||||
response = send_http_request(
|
||||
'GET',
|
||||
URL,
|
||||
headers={'Authorization': f'Bearer {auth_token}'},
|
||||
params=params,
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
if 'application/json' in response.headers.get('Content-Type', '').lower():
|
||||
response_data = response.json()
|
||||
else:
|
||||
response_data = response.text
|
||||
|
||||
return response_data
|
||||
except requests.HTTPError as error:
|
||||
response = error.response
|
||||
logger.error('Произошла ошибка во время получения очередей: %s', error)
|
||||
if response is not None:
|
||||
logger.debug("status=%s url=%s", response.status_code, response.url)
|
||||
logger.debug("content-type=%s", response.headers.get('Content-Type'))
|
||||
logger.debug("response(first 1500): %s", response.text[:1500])
|
||||
return None
|
||||
except Exception as error:
|
||||
logger.exception('Произошла ошибка во время получения очередей: %s', error)
|
||||
return None
|
||||
|
||||
|
||||
def get_queue_by_project_id(auth_token: str, project_id: int = 150):
|
||||
"""GET /api/RpaProjectQueue с параметром ProjectId (по умолчанию 150)."""
|
||||
url = URL_API_GENERAL + '/api/RpaProjectQueue'
|
||||
try:
|
||||
response = send_http_request(
|
||||
'GET',
|
||||
url,
|
||||
headers={'Authorization': f'Bearer {auth_token}'},
|
||||
params={'ProjectId': project_id},
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
if 'application/json' in response.headers.get('Content-Type', '').lower():
|
||||
return response.json()
|
||||
return response.text
|
||||
except requests.HTTPError as error:
|
||||
response = error.response
|
||||
logger.error('Произошла ошибка во время GET /api/RpaProjectQueue: %s', error)
|
||||
if response is not None:
|
||||
logger.debug("status=%s url=%s", response.status_code, response.url)
|
||||
logger.debug("content-type=%s", response.headers.get('Content-Type'))
|
||||
logger.debug("response(first 1500): %s", response.text[:1500])
|
||||
return None
|
||||
except Exception as error:
|
||||
logger.exception('Произошла ошибка во время GET /api/RpaProjectQueue: %s', error)
|
||||
return None
|
||||
|
||||
|
||||
def get_test_endpoint(auth_token: str):
|
||||
"""Контрольный GET /api/test для проверки базового доступа с токеном."""
|
||||
url = URL_API_GENERAL + '/api/test'
|
||||
params = None
|
||||
try:
|
||||
response = send_http_request(
|
||||
'GET',
|
||||
url,
|
||||
headers={'Authorization': f'Bearer {auth_token}'},
|
||||
params=params,
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
if 'application/json' in response.headers.get('Content-Type', '').lower():
|
||||
return response.json()
|
||||
return response.text
|
||||
except requests.HTTPError as error:
|
||||
response = error.response
|
||||
logger.error('Произошла ошибка во время контрольного GET /api/test: %s', error)
|
||||
if response is not None:
|
||||
logger.debug("status=%s url=%s", response.status_code, response.url)
|
||||
logger.debug("content-type=%s", response.headers.get('Content-Type'))
|
||||
logger.debug("response(first 1500): %s", response.text[:1500])
|
||||
return None
|
||||
except Exception as error:
|
||||
logger.exception('Произошла ошибка во время контрольного GET /api/test: %s', error)
|
||||
return None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
token = get_token()
|
||||
logger.info("Длинна токена: %s", len(token))
|
||||
|
||||
logger.info('Проверяем контрольный GET /api/test ...')
|
||||
control_result = get_test_endpoint(token)
|
||||
if control_result is not None:
|
||||
logger.info('Контрольный GET успешен')
|
||||
else:
|
||||
logger.info('Контрольный GET неуспешен')
|
||||
logger.debug("Результат: %s", control_result)
|
||||
|
||||
logger.info('Проверяем GET /api/RpaProjectQueue?ProjectId=150 ...')
|
||||
queue_by_project_result = get_queue_by_project_id(token)
|
||||
if queue_by_project_result is not None:
|
||||
logger.info('GET /api/RpaProjectQueue?ProjectId=150 успешен')
|
||||
else:
|
||||
logger.info('GET /api/RpaProjectQueue?ProjectId=150 неуспешен')
|
||||
logger.debug('Результат /api/RpaProjectQueue?ProjectId=150: %s', queue_by_project_result)
|
||||
|
||||
logger.info('Проверяем GET /api/RpaProjectQueue/All без eventDate ...')
|
||||
queue_result_no_date = get_queue(token)
|
||||
logger.debug('Результат без eventDate: %s', queue_result_no_date)
|
||||
|
||||
event_date = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
|
||||
logger.info('Проверяем GET /api/RpaProjectQueue/All с eventDate=%s ...', event_date)
|
||||
queue_result_with_date = get_queue(token, event_date=event_date)
|
||||
logger.debug('Результат с eventDate: %s', queue_result_with_date)
|
||||
Reference in New Issue
Block a user