248 lines
9.4 KiB
Python
248 lines
9.4 KiB
Python
"""Базовый класс страницы для работы с Playwright.
|
||
|
||
Содержит общие методы для взаимодействия со страницей и API.
|
||
"""
|
||
import time
|
||
import json
|
||
from typing import Dict, Any
|
||
from playwright.sync_api import Page, Response, APIRequestContext, expect
|
||
from data.environment import host
|
||
from tools.logger import get_logger
|
||
|
||
logger = get_logger("BASE_PAGE")
|
||
|
||
|
||
class BasePage:
|
||
"""Базовый класс для работы со страницами через Playwright.
|
||
|
||
Атрибуты:
|
||
page (Page): Экземпляр страницы Playwright.
|
||
"""
|
||
|
||
def __init__(self, page: Page):
|
||
"""Инициализирует базовую страницу.
|
||
|
||
Args:
|
||
page (Page): Экземпляр страницы Playwright.
|
||
"""
|
||
self.page = page
|
||
|
||
# Действия:
|
||
def current_url(self) -> str:
|
||
"""Возвращает текущий URL страницы.
|
||
|
||
Returns:
|
||
str: Текущий URL страницы.
|
||
"""
|
||
return self.page.url
|
||
|
||
def open(self, uri) -> Response | None:
|
||
"""Открывает указанный URI в браузере.
|
||
|
||
Args:
|
||
uri (str): URI для открытия (без базового URL).
|
||
|
||
Returns:
|
||
Response | None: Ответ сервера или None в случае ошибки.
|
||
"""
|
||
return self.page.goto(f"{host.get_base_url()}{uri}", wait_until='domcontentloaded')
|
||
|
||
def page_reload(self) -> None:
|
||
"""Перезагружает текущую страницу."""
|
||
self.page.reload()
|
||
|
||
def wait_for_timeout(self, timeout: int) -> None:
|
||
"""Ожидает указанное количество миллисекунд.
|
||
|
||
Args:
|
||
timeout (int): Время ожидания в миллисекундах.
|
||
"""
|
||
self.page.wait_for_timeout(timeout)
|
||
|
||
def get_api_request_context(self) -> APIRequestContext:
|
||
"""Возвращает контекст API-запросов.
|
||
|
||
Returns:
|
||
APIRequestContext: Контекст для выполнения API-запросов.
|
||
"""
|
||
return self.page.context.request
|
||
|
||
def send_get_api_request(self, uri: str) -> Response:
|
||
"""Отправляет GET-запрос к API.
|
||
|
||
Args:
|
||
uri (str): URI API-эндпоинта (без базового URL).
|
||
|
||
Returns:
|
||
Response: Ответ сервера.
|
||
"""
|
||
api_request_context = self.get_api_request_context()
|
||
token = host.get_access_token()
|
||
|
||
# Проверяем что токен получен
|
||
if not token:
|
||
logger.error("Failed to get access token: token is None or empty")
|
||
# Возвращаем заглушечный response или бросаем исключение
|
||
# В Playwright можно создать mock response если нужно
|
||
return None
|
||
|
||
headers = {"Accept": "application/json", "Authorization": f"Bearer {token}"}
|
||
full_url = f"{host.get_request_url()}{uri}"
|
||
|
||
logger.debug("Sending GET request to: %s", full_url)
|
||
response = api_request_context.get(full_url, headers=headers)
|
||
|
||
# Логируем статус ответа
|
||
logger.debug("GET response status: %s", response.status)
|
||
|
||
return response
|
||
|
||
def send_post_api_request(self, uri: str, payload: Dict) -> Response:
|
||
"""Отправляет POST-запрос к API."""
|
||
api_request_context = self.get_api_request_context()
|
||
token = host.get_access_token()
|
||
|
||
if not token:
|
||
logger.error("Failed to get access token: token is None or empty")
|
||
return None
|
||
|
||
headers = {
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {token}"
|
||
}
|
||
full_url = f"{host.get_request_url()}{uri}"
|
||
|
||
logger.debug("Sending POST request to: %s", full_url)
|
||
|
||
# Сериализуем payload в JSON
|
||
json_data = json.dumps(payload)
|
||
|
||
# Проверяем что сериализация прошла успешно
|
||
if json_data is None:
|
||
logger.error("Failed to serialize payload to JSON: result is None")
|
||
return None
|
||
|
||
if not isinstance(json_data, str):
|
||
logger.error("Failed to serialize payload to JSON: expected string got %s", type(json_data))
|
||
return None
|
||
|
||
response = api_request_context.post(
|
||
full_url,
|
||
headers=headers,
|
||
data=json_data # Передаем сериализованный JSON как data
|
||
)
|
||
|
||
logger.debug("POST response status: %s", response.status)
|
||
|
||
return response
|
||
|
||
def get_response_body(self, response: Response) -> dict | list | None:
|
||
"""Извлекает тело ответа в format JSON.
|
||
|
||
Args:
|
||
response (Response): Ответ сервера.
|
||
|
||
Returns:
|
||
dict | list | None: Распарсенное тело ответа или None в случае ошибки.
|
||
"""
|
||
start_time = time.time()
|
||
|
||
# Проверяем что response не None
|
||
if response is None:
|
||
logger.error("Response object is None")
|
||
processing_time = time.time() - start_time
|
||
logger.debug("Response processing time1: %.3f seconds", processing_time)
|
||
return None
|
||
|
||
# Проверяем статус ответа
|
||
if response.status >= 400:
|
||
logger.error("API request failed with status %s", response.status)
|
||
processing_time = time.time() - start_time
|
||
logger.debug("Response processing time2: %.3f seconds", processing_time)
|
||
return None
|
||
|
||
# Пытаемся получить JSON
|
||
json_result = response.json()
|
||
|
||
# Проверяем что результат не None
|
||
if json_result is None:
|
||
logger.error("JSON parsing returned None")
|
||
processing_time = time.time() - start_time
|
||
logger.debug("Response processing time3: %.3f seconds", processing_time)
|
||
return None
|
||
|
||
# Принимаем как словари, так и списки
|
||
if not isinstance(json_result, (dict, list)):
|
||
logger.error("Expected dict or list but got %s", type(json_result))
|
||
processing_time = time.time() - start_time
|
||
logger.debug("Response processing time4: %.3f seconds", processing_time)
|
||
return None
|
||
|
||
processing_time = time.time() - start_time
|
||
logger.debug("Response processing time5: %.3f seconds", processing_time)
|
||
|
||
return json_result
|
||
|
||
# Проверки:
|
||
def check_URL(self, uri: str, msg: str) -> None:
|
||
"""Проверяет, что текущий URL соответствует ожидаемому.
|
||
|
||
Args:
|
||
uri (str): Ожидаемый URI (без базового URL).
|
||
msg (str): Сообщение об ошибке при несоответствии.
|
||
|
||
Raises:
|
||
AssertionError: Если URL не соответствует ожидаемому.
|
||
"""
|
||
expect(self.page).to_have_url( # pylint: disable=expression-not-assigned
|
||
f"{host.get_base_url()}{uri}",
|
||
timeout=60000
|
||
), msg
|
||
|
||
def check_equals(self, actual: Any, expected: Any, msg: str) -> None:
|
||
"""Проверяет равенство фактического и ожидаемого значений.
|
||
|
||
Args:
|
||
actual (Any): Фактическое значение.
|
||
expected (Any): Ожидаемое значение.
|
||
msg (str): Сообщение об ошибке при несоответствии.
|
||
|
||
Raises:
|
||
AssertionError: Если значения не равны.
|
||
"""
|
||
assert actual == expected, msg
|
||
|
||
def check_lists_equals(self, actual: list, expected: list, msg: str) -> None:
|
||
"""Рекурсивно проверяет равенство двух списков.
|
||
|
||
Args:
|
||
actual (list): Фактический список.
|
||
expected (list): Ожидаемый список.
|
||
msg (str): Сообщение об ошибке при несоответствии.
|
||
|
||
Raises:
|
||
AssertionError: Если списки не равны.
|
||
"""
|
||
def compare_lists(list1: list, list2: list) -> bool:
|
||
"""Вспомогательная функция для рекурсивного сравнения списков.
|
||
|
||
Args:
|
||
list1 (list): Первый список для сравнения.
|
||
list2 (list): Второй список для сравнения.
|
||
|
||
Returns:
|
||
bool: True если списки идентичны, иначе False.
|
||
"""
|
||
if len(list1) != len(list2):
|
||
return False
|
||
for item1, item2 in zip(list1, list2):
|
||
if isinstance(item1, list) and isinstance(item2, list):
|
||
if not compare_lists(item1, item2):
|
||
return False
|
||
elif item1 != item2:
|
||
return False
|
||
return True
|
||
|
||
assert compare_lists(actual, expected), msg
|