commit eff152ea4adebc92861f1e68683b3e1161b33830 Author: somilovanov Date: Mon Apr 27 16:51:09 2026 +0300 Initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a1b9cbb --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +.env +__pycache__/ \ No newline at end of file diff --git a/test_metric.py b/test_metric.py new file mode 100644 index 0000000..a74a8d7 --- /dev/null +++ b/test_metric.py @@ -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)