213 lines
8.2 KiB
Python
213 lines
8.2 KiB
Python
import json
|
||
import requests
|
||
import allure
|
||
import logging
|
||
from typing import Optional, Dict, Any
|
||
from urllib.parse import urljoin
|
||
from config.api_config import api_config
|
||
from config.session_config import session_config
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class APIClient:
|
||
def __init__(self, base_url: Optional[str] = None):
|
||
self.base_url = base_url or api_config.API_BASE_URL
|
||
self.session = requests.Session()
|
||
self.session.cookies.clear()
|
||
self.logged_in = False
|
||
logger.info("New API created.")
|
||
|
||
# Настройка сессии
|
||
self._configure_session()
|
||
|
||
def _configure_session(self):
|
||
"""Настройка HTTP сессии"""
|
||
# Установка заголовков по умолчанию
|
||
self.session.headers.update(api_config.DEFAULT_HEADERS)
|
||
self.session.headers.update(session_config.get_session_headers())
|
||
|
||
logger.debug(f"API Client configured for {self.base_url}")
|
||
|
||
def login(self, username: str, password: str) -> bool:
|
||
"""Логин пользователя (создание сессии)"""
|
||
login_data = {
|
||
"username": username,
|
||
"password": password
|
||
}
|
||
|
||
endpoint = api_config.get_endpoint('auth_login')
|
||
response = self.post(endpoint, json=login_data)
|
||
|
||
if response.status_code == 200:
|
||
self.logged_in = True
|
||
logger.info(f"Successfully logged in as {username}")
|
||
|
||
# Логирование cookies для отладки
|
||
if self.session.cookies:
|
||
cookies_info = dict(self.session.cookies)
|
||
logger.info(f"Session cookies: {cookies_info}")
|
||
|
||
# Проверяем наличие сессионного cookie
|
||
session_cookie = session_config.SESSION_COOKIE_NAME
|
||
if session_cookie in cookies_info:
|
||
logger.info(f"Session cookie '{session_cookie}' is set")
|
||
|
||
return True
|
||
else:
|
||
logger.error(f"Login failed for {username}: {response.status_code}")
|
||
logger.debug(f"Response: {response.text}")
|
||
return False
|
||
|
||
def logout(self) -> bool:
|
||
"""Выход из системы (закрытие сессии)"""
|
||
if not self.logged_in:
|
||
logger.warning("Attempted logout without being logged in")
|
||
return True
|
||
|
||
endpoint = api_config.get_endpoint('auth_logout')
|
||
response = self.get(endpoint)
|
||
|
||
if response.status_code == 200:
|
||
self.logged_in = False
|
||
|
||
# Очищаем cookies
|
||
self.session.cookies.clear()
|
||
|
||
logger.info("Successfully logged out")
|
||
return True
|
||
else:
|
||
logger.error(f"Logout failed: {response.status_code}")
|
||
return False
|
||
|
||
def ensure_logged_in(self, username: str, password: str):
|
||
"""Убедиться, что пользователь залогинен"""
|
||
if not self.logged_in:
|
||
return self.login(username, password)
|
||
return True
|
||
|
||
def _send_request(
|
||
self,
|
||
method: str,
|
||
endpoint: str,
|
||
**kwargs
|
||
) -> requests.Response:
|
||
"""Отправка HTTP запроса с поддержкой сессий"""
|
||
url = urljoin(self.base_url, endpoint)
|
||
|
||
with allure.step(f"{method.upper()} {endpoint}"):
|
||
# Добавляем логирование
|
||
self._log_request(method, url, kwargs)
|
||
|
||
try:
|
||
response = self.session.request(method, url, headers=api_config.DEFAULT_HEADERS, **kwargs)
|
||
except requests.exceptions.RequestException as e:
|
||
logger.error(f"Request failed: {e}")
|
||
raise
|
||
|
||
# Логирование ответа
|
||
self._log_response(response)
|
||
|
||
return response
|
||
|
||
def _log_request(self, method: str, url: str, kwargs: Dict[str, Any]):
|
||
"""Логирование деталей запроса"""
|
||
request_details = (
|
||
f"Request: {method} {url}\n"
|
||
f"Headers: {dict(self.session.headers)}\n"
|
||
f"Cookies: {dict(self.session.cookies)}\n"
|
||
f"Body: {kwargs.get('json', 'N/A')}"
|
||
)
|
||
|
||
allure.attach(
|
||
request_details,
|
||
name="Request Details",
|
||
attachment_type=allure.attachment_type.TEXT
|
||
)
|
||
|
||
logger.debug(f"{method} {url}")
|
||
|
||
def _log_response(self, response: requests.Response):
|
||
"""Логирование деталей ответа"""
|
||
response_details = (
|
||
f"Status Code: {response.status_code}\n"
|
||
f"Response Headers: {dict(response.headers)}\n"
|
||
f"Response Cookies: {dict(response.cookies)}\n"
|
||
f"Response Body: {response.text[:500]}..."
|
||
)
|
||
|
||
allure.attach(
|
||
response_details,
|
||
name="Response Details",
|
||
attachment_type=allure.attachment_type.TEXT
|
||
)
|
||
|
||
logger.debug(f"Response: {response.status_code}")
|
||
|
||
# Логирование ошибок
|
||
if response.status_code >= 400:
|
||
logger.warning(f"Request failed with status {response.status_code}")
|
||
logger.debug(f"Error response: {response.text}")
|
||
|
||
# Методы HTTP запросов
|
||
def get(self, endpoint: str, **kwargs) -> requests.Response:
|
||
return self._send_request("GET", endpoint, **kwargs)
|
||
|
||
def post(self, endpoint: str, **kwargs) -> requests.Response:
|
||
return self._send_request("POST", endpoint, **kwargs)
|
||
|
||
def put(self, endpoint: str, **kwargs) -> requests.Response:
|
||
return self._send_request("PUT", endpoint, **kwargs)
|
||
|
||
def delete(self, endpoint: str, **kwargs) -> requests.Response:
|
||
return self._send_request("DELETE", endpoint, **kwargs)
|
||
|
||
def patch(self, endpoint: str, **kwargs) -> requests.Response:
|
||
return self._send_request("PATCH", endpoint, **kwargs)
|
||
|
||
# Специальные методы для endpoints
|
||
def create_user(self, user_data: Dict[str, Any]) -> requests.Response:
|
||
endpoint = api_config.get_endpoint('users_create')
|
||
return self.post(endpoint, json=user_data)
|
||
|
||
def get_user(self, user_id: str) -> requests.Response:
|
||
endpoint = api_config.get_endpoint('users_get_by_id', id=user_id)
|
||
return self.get(endpoint)
|
||
|
||
def get_all_users(self) -> requests.Response:
|
||
endpoint = api_config.get_endpoint('users_get_all')
|
||
return self.get(endpoint)
|
||
|
||
def update_user(self, user_id: str, user_data: Dict[str, Any]) -> requests.Response:
|
||
endpoint = api_config.get_endpoint('users_update', id=user_id)
|
||
return self.put(endpoint, json=user_data)
|
||
|
||
def delete_user(self, user_id: str) -> requests.Response:
|
||
endpoint = api_config.get_endpoint('users_delete', id=user_id)
|
||
return self.delete(endpoint)
|
||
|
||
def create_post(self, post_data: Dict[str, Any]) -> requests.Response:
|
||
endpoint = api_config.get_endpoint('posts_create')
|
||
return self.post(endpoint, json=post_data)
|
||
|
||
def get_post(self, post_id: str) -> requests.Response:
|
||
endpoint = api_config.get_endpoint('posts_get_by_id', id=post_id)
|
||
return self.get(endpoint)
|
||
|
||
def get_all_posts(self) -> requests.Response:
|
||
endpoint = api_config.get_endpoint('posts_get_all')
|
||
return self.get(endpoint)
|
||
|
||
def update_post(self, post_id: str, post_data: Dict[str, Any]) -> requests.Response:
|
||
endpoint = api_config.get_endpoint('posts_update', id=post_id)
|
||
return self.put(endpoint, json=post_data)
|
||
|
||
def delete_post(self, post_id: str) -> requests.Response:
|
||
endpoint = api_config.get_endpoint("posts_delete", id=post_id)
|
||
return self.delete(endpoint)
|
||
|
||
def upload_image(self, file_path: str, field_name: str = "file") -> requests.Response:
|
||
endpoint = api_config.get_endpoint('images_upload')
|
||
|
||
with open(file_path, 'rb') as f:
|
||
files = {field_name: (file_path, f)}
|
||
return self.post(endpoint, files=files)
|