218 lines
8.5 KiB
Python
218 lines
8.5 KiB
Python
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)
|