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)