diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 731ebc4..3eed5b9 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -52,7 +52,7 @@ jobs: env: COVERAGE_FILE: coverage-unit.dat run: > - pytest -q tests/test_mellophone.py tests/test_structures.py + pytest -q tests/unit --cov=src/mellophone --cov-report= @@ -132,7 +132,7 @@ jobs: env: COVERAGE_FILE: coverage-integration.dat run: > - pytest -q tests/test_integration_mellophone.py + pytest -q tests/integration --cov=src/mellophone --cov-report= diff --git a/README.md b/README.md index 95071ac..a08da02 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,8 @@ uv add "mellophone-valve[httpx,requests] @ git+https://github.com/CourseOrchestr ## Быстрый старт +### Синхронный + ```python from mellophone import Mellophone @@ -85,6 +87,23 @@ print(client.is_authenticated(session_id)) client.logout(session_id) ``` +### Асинхронный + +```python +import asyncio +from mellophone import Mellophone + + +async def async_call(): + client = Mellophone(base_url="http://localhost:8082/mellophone") + session_id = await client.login_async("user", "password") + print(await client.is_authenticated_async(session_id)) + await client.logout_async(session_id) + + +asyncio.run(async_call()) +``` + ## API клиента Класс `Mellophone` поддерживает пары методов `sync/async`. @@ -110,14 +129,14 @@ client.logout(session_id) - `import_gp/import_gp_async` - импортирует groups/providers, возвращает список строк из ответа API. - `get_provider_list/get_provider_list_async` - возвращает список или структуру провайдеров по учетным данным. -- `get_user_list/get_user_list_async` - возвращает список пользователей по `gp` (опционально `ip`, `pid`); токен берется из `self.user_manage_token`. +- `get_user_list/get_user_list_async` - возвращает список пользователей по `gp` (опционально `ip`, `pid`); токен берется из `self.token_user_manage`. Настройки и user management: -- `set_settings/set_settings_async` - обновляет настройки (`lockout_time`, `login_attempts_allowed`); токен берется из `self.set_settings_token`. -- `create_user/create_user_async` - создает пользователя (`POST /user/create`, XML payload); токен берется из `self.user_manage_token`. -- `update_user/update_user_async` - обновляет пользователя по `sid` (`POST /user/{sid}`, XML payload); токен берется из `self.user_manage_token`. -- `delete_user/delete_user_async` - удаляет пользователя по `sid` (`DELETE /user/{sid}`); токен берется из `self.user_manage_token`. +- `set_settings/set_settings_async` - обновляет настройки (`lockout_time`, `login_attempts_allowed`); токен берется из `self.token_set_settings`. +- `create_user/create_user_async` - создает пользователя (`POST /user/create`, XML payload); токен берется из `self.token_user_manage`. +- `update_user/update_user_async` - обновляет пользователя по `sid` (`POST /user/{sid}`, XML payload); токен берется из `self.token_user_manage`. +- `delete_user/delete_user_async` - удаляет пользователя по `sid` (`DELETE /user/{sid}`); токен берется из `self.token_user_manage`. Состояние сессии: @@ -135,5 +154,9 @@ client.logout(session_id) - `TransportError` - транспортная ошибка HTTP-клиента (сеть/соединение). - `RequestTimeoutError` - превышен таймаут запроса. - `ResponseParseError` - не удалось распарсить XML-ответ API. -- `MissingTokenError` - в клиенте не задан обязательный токен (`set_settings_token` или `user_manage_token`). +- `MissingTokenError` - в клиенте не задан обязательный токен (`token_set_settings` или `token_user_manage`). + +Совместимость: + +- Старые имена `set_settings_token` и `user_manage_token` сохранены как deprecated-алиасы и вызывают `DeprecationWarning`. - `AsyncClientUnavailableError` - вызваны `async`-методы без установленного `httpx`. diff --git a/pyproject.toml b/pyproject.toml index c68b35a..facb064 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "mellophone-valve" -version = "3.1.1" +version = "3.2.0" description = "Python-клиент для Mellophone" readme = "README.md" requires-python = ">=3.7" diff --git a/src/mellophone/client.py b/src/mellophone/client.py index 0f16feb..6adcb30 100644 --- a/src/mellophone/client.py +++ b/src/mellophone/client.py @@ -1,7 +1,8 @@ from __future__ import annotations +import json +import warnings import xml.etree.ElementTree as ET -from dataclasses import dataclass from http import HTTPStatus from typing import Any, Dict, List, Optional, Union from urllib.parse import urlencode @@ -34,17 +35,91 @@ requests = None # type: ignore[assignment] -@dataclass class Mellophone: - """Unified Mellophone client with sync and async methods.""" + """Единый клиент Mellophone с синхронными и асинхронными методами.""" base_url: str - set_settings_token: Optional[str] = None - user_manage_token: Optional[str] = None - session_id: Optional[str] = None - timeout: float = 10.0 + token_set_settings: Optional[str] + token_user_manage: Optional[str] + session_id: Optional[str] + timeout: float + + def __init__( + self, + base_url: str, + token_set_settings: Optional[str] = None, + token_user_manage: Optional[str] = None, + session_id: Optional[str] = None, + timeout: float = 10.0, + set_settings_token: Optional[str] = None, + user_manage_token: Optional[str] = None, + ) -> None: + """Инициализирует клиент и поддерживает устаревшие имена токенов.""" + if set_settings_token is not None: + warnings.warn( + "`set_settings_token` is deprecated, use `token_set_settings`.", + DeprecationWarning, + stacklevel=2, + ) + if token_set_settings is None: + token_set_settings = set_settings_token + if user_manage_token is not None: + warnings.warn( + "`user_manage_token` is deprecated, use `token_user_manage`.", + DeprecationWarning, + stacklevel=2, + ) + if token_user_manage is None: + token_user_manage = user_manage_token + + self.base_url = base_url + self.token_set_settings = token_set_settings + self.token_user_manage = token_user_manage + self.session_id = session_id + self.timeout = timeout + + @property + def set_settings_token(self) -> Optional[str]: + """Устаревший алиас для token_set_settings.""" + warnings.warn( + "`set_settings_token` is deprecated, use `token_set_settings`.", + DeprecationWarning, + stacklevel=2, + ) + return self.token_set_settings + + @set_settings_token.setter + def set_settings_token(self, value: Optional[str]) -> None: + """Устаревший сеттер для token_set_settings.""" + warnings.warn( + "`set_settings_token` is deprecated, use `token_set_settings`.", + DeprecationWarning, + stacklevel=2, + ) + self.token_set_settings = value + + @property + def user_manage_token(self) -> Optional[str]: + """Устаревший алиас для token_user_manage.""" + warnings.warn( + "`user_manage_token` is deprecated, use `token_user_manage`.", + DeprecationWarning, + stacklevel=2, + ) + return self.token_user_manage + + @user_manage_token.setter + def user_manage_token(self, value: Optional[str]) -> None: + """Устаревший сеттер для token_user_manage.""" + warnings.warn( + "`user_manage_token` is deprecated, use `token_user_manage`.", + DeprecationWarning, + stacklevel=2, + ) + self.token_user_manage = value def _build_url(self, path: str, params: Optional[Dict[str, Any]] = None) -> str: + """Собирает полный URL запроса с query-параметрами.""" if not isinstance(path, str): raise TypeError("path must be str") path = "/" + path.strip("/") @@ -55,6 +130,7 @@ def _build_url(self, path: str, params: Optional[Dict[str, Any]] = None) -> str: @staticmethod def _raise_for_status(response: Any) -> None: + """Выбрасывает доменное исключение для неуспешных HTTP-статусов.""" status_code = int(response.status_code) response_text = response.text @@ -73,6 +149,7 @@ def _raise_for_status(response: Any) -> None: @staticmethod def _ensure_sync_backend() -> None: + """Проверяет, что установлен хотя бы один sync HTTP-бэкенд.""" if httpx is None and requests is None: raise RuntimeError( "No HTTP client is installed. Install mellophone-valve[httpx] or mellophone-valve[requests]." @@ -87,6 +164,7 @@ def _request_text( data: Optional[str] = None, headers: Optional[Dict[str, str]] = None, ) -> str: + """Выполняет sync HTTP-запрос и возвращает текст ответа.""" self._ensure_sync_backend() url = self._build_url(path, params) @@ -119,6 +197,7 @@ async def _request_text_async( data: Optional[str] = None, headers: Optional[Dict[str, str]] = None, ) -> str: + """Выполняет async HTTP-запрос и возвращает текст ответа.""" if httpx is None: raise AsyncClientUnavailableError("Async methods require httpx. Install mellophone-valve[httpx].") @@ -136,6 +215,7 @@ async def _request_text_async( @staticmethod def _as_json(response_text: str) -> Dict[str, Any]: + """Парсит XML-текст ответа в словарь.""" if not response_text.strip(): return {} try: @@ -145,19 +225,34 @@ def _as_json(response_text: str) -> Dict[str, Any]: @staticmethod def _require_user(user: Dict[str, Any]) -> Dict[str, Any]: + """Проверяет и нормализует payload пользователя для API-вызовов.""" if not user: raise ValueError("user data cannot be empty") - payload = dict(user) - if "password" in payload: - payload["pwd"] = payload.pop("password") + payload: Dict[str, Any] = {} + for key, val in list(user.items())[:]: + if key == "password": + key = "pwd" + elif isinstance(val, (list, dict)): + val = json.dumps(val, ensure_ascii=False) + elif val is None: + val = "" + payload[key] = val return payload @staticmethod def _require_token(token: Optional[str], *, field_name: str) -> str: + """Возвращает обязательный токен или выбрасывает MissingTokenError.""" if token: return token raise MissingTokenError(f"{field_name} is required on Mellophone client.") + def _require_session_id(self, ses_id: Optional[str] = None) -> str: + """Определяет session id из аргумента или состояния клиента.""" + resolved_ses_id = ses_id or self.session_id + if resolved_ses_id: + return resolved_ses_id + raise ValueError("ses_id is required. Pass ses_id explicitly or call login first.") + @staticmethod def _login_props( login: str, @@ -166,6 +261,7 @@ def _login_props( gp: Optional[str] = None, ip: Optional[str] = None, ) -> RequestParams: + """Формирует параметры запроса для endpoint `login`.""" return RequestParams( path="login", params={ @@ -185,6 +281,7 @@ def login( gp: Optional[str] = None, ip: Optional[str] = None, ) -> str: + """Аутентифицирует пользователя и сохраняет session id в клиенте.""" ses_id = ses_id or self.session_id or str(uuid4()) self._request_text(**self._login_props(login, password, ses_id, gp, ip)) self.session_id = ses_id @@ -198,6 +295,7 @@ async def login_async( gp: Optional[str] = None, ip: Optional[str] = None, ) -> str: + """Асинхронно аутентифицирует пользователя и сохраняет session id.""" ses_id = ses_id or self.session_id or str(uuid4()) await self._request_text_async(**self._login_props(login, password, ses_id, gp, ip)) self.session_id = ses_id @@ -205,28 +303,38 @@ async def login_async( @staticmethod def _logout_props(ses_id: Optional[str] = None) -> RequestParams: + """Формирует параметры запроса для endpoint `logout`.""" return RequestParams(path="logout", params={"sesid": ses_id}) def logout(self, ses_id: Optional[str] = None) -> None: - self._request_text(**self._logout_props(ses_id or self.session_id)) + """Завершает текущую или явно переданную сессию.""" + resolved_ses_id = self._require_session_id(ses_id) + self._request_text(**self._logout_props(resolved_ses_id)) async def logout_async(self, ses_id: Optional[str] = None) -> None: - await self._request_text_async(**self._logout_props(ses_id or self.session_id)) + """Асинхронно завершает текущую или явно переданную сессию.""" + resolved_ses_id = self._require_session_id(ses_id) + await self._request_text_async(**self._logout_props(resolved_ses_id)) @staticmethod def _is_authenticated_props(ses_id: Optional[str]) -> RequestParams: + """Формирует параметры запроса для проверки аутентификации.""" return RequestParams(path="isauthenticated", params={"sesid": ses_id}) def is_authenticated(self, ses_id: Optional[str] = None) -> Union[Dict[str, Any], bool]: + """Проверяет, аутентифицирована ли сессия.""" + resolved_ses_id = self._require_session_id(ses_id) try: - response = self._request_text(**self._is_authenticated_props(ses_id or self.session_id)) + response = self._request_text(**self._is_authenticated_props(resolved_ses_id)) except ForbiddenError: return False return self._as_json(response).get("user", {}) async def is_authenticated_async(self, ses_id: Optional[str] = None) -> Union[Dict[str, Any], bool]: + """Асинхронно проверяет, аутентифицирована ли сессия.""" + resolved_ses_id = self._require_session_id(ses_id) try: - response = await self._request_text_async(**self._is_authenticated_props(ses_id or self.session_id)) + response = await self._request_text_async(**self._is_authenticated_props(resolved_ses_id)) except ForbiddenError: return False return self._as_json(response).get("user", {}) @@ -238,6 +346,7 @@ def _check_credentials_props( gp: Optional[str] = None, ip: Optional[str] = None, ) -> RequestParams: + """Формирует параметры запроса для проверки учетных данных.""" return RequestParams( path="checkcredentials", params={"login": login, "pwd": password, "gp": gp, "ip": ip}, @@ -250,6 +359,7 @@ def check_credentials( gp: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: + """Проверяет учетные данные и возвращает данные пользователя.""" response = self._request_text(**self._check_credentials_props(login, password, gp, ip)) return self._as_json(response).get("user", {}) @@ -260,23 +370,30 @@ async def check_credentials_async( gp: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: + """Асинхронно проверяет учетные данные и возвращает данные пользователя.""" response = await self._request_text_async(**self._check_credentials_props(login, password, gp, ip)) return self._as_json(response).get("user", {}) @staticmethod def _check_name_props(name: str, ses_id: Optional[str]) -> RequestParams: + """Формирует параметры запроса для проверки доступности имени.""" return RequestParams(path="checkname", params={"sesid": ses_id, "name": name}) def check_name(self, name: str, ses_id: Optional[str] = None) -> Dict[str, Any]: - response = self._request_text(**self._check_name_props(name, ses_id or self.session_id)) + """Проверяет доступность имени пользователя для сессии.""" + resolved_ses_id = self._require_session_id(ses_id) + response = self._request_text(**self._check_name_props(name, resolved_ses_id)) return self._as_json(response).get("user", {}) async def check_name_async(self, name: str, ses_id: Optional[str] = None) -> Dict[str, Any]: - response = await self._request_text_async(**self._check_name_props(name, ses_id or self.session_id)) + """Асинхронно проверяет доступность имени пользователя для сессии.""" + resolved_ses_id = self._require_session_id(ses_id) + response = await self._request_text_async(**self._check_name_props(name, resolved_ses_id)) return self._as_json(response).get("user", {}) @staticmethod def _change_pwd_props(old_pwd: str, new_pwd: str, ses_id: Optional[str]) -> RequestParams: + """Формирует параметры запроса для смены пароля текущего пользователя.""" return RequestParams( path="changepwd", params={ @@ -287,10 +404,14 @@ def _change_pwd_props(old_pwd: str, new_pwd: str, ses_id: Optional[str]) -> Requ ) def change_pwd(self, old_pwd: str, new_pwd: str, ses_id: Optional[str] = None) -> None: - self._request_text(**self._change_pwd_props(old_pwd, new_pwd, ses_id or self.session_id)) + """Меняет пароль текущего аутентифицированного пользователя.""" + resolved_ses_id = self._require_session_id(ses_id) + self._request_text(**self._change_pwd_props(old_pwd, new_pwd, resolved_ses_id)) async def change_pwd_async(self, old_pwd: str, new_pwd: str, ses_id: Optional[str] = None) -> None: - await self._request_text_async(**self._change_pwd_props(old_pwd, new_pwd, ses_id or self.session_id)) + """Асинхронно меняет пароль текущего пользователя.""" + resolved_ses_id = self._require_session_id(ses_id) + await self._request_text_async(**self._change_pwd_props(old_pwd, new_pwd, resolved_ses_id)) @staticmethod def _change_user_pwd_props( @@ -298,6 +419,7 @@ def _change_user_pwd_props( new_pwd: str, ses_id: Optional[str], ) -> RequestParams: + """Формирует параметры запроса для смены пароля пользователя по имени.""" return RequestParams( path="changeuserpwd", params={ @@ -308,36 +430,48 @@ def _change_user_pwd_props( ) def change_user_pwd(self, username: str, new_pwd: str, ses_id: Optional[str] = None) -> None: - self._request_text(**self._change_user_pwd_props(username, new_pwd, ses_id or self.session_id)) + """Меняет пароль пользователя, указанного по имени.""" + resolved_ses_id = self._require_session_id(ses_id) + self._request_text(**self._change_user_pwd_props(username, new_pwd, resolved_ses_id)) async def change_user_pwd_async(self, username: str, new_pwd: str, ses_id: Optional[str] = None) -> None: - await self._request_text_async(**self._change_user_pwd_props(username, new_pwd, ses_id or self.session_id)) + """Асинхронно меняет пароль пользователя по имени.""" + resolved_ses_id = self._require_session_id(ses_id) + await self._request_text_async(**self._change_user_pwd_props(username, new_pwd, resolved_ses_id)) @staticmethod def _change_app_ses_id_props(new_ses_id: str, ses_id: Optional[str]) -> RequestParams: + """Формирует параметры запроса для смены session id.""" return RequestParams( path="changeappsesid", params={"oldsesid": ses_id, "newsesid": new_ses_id}, ) def change_app_ses_id(self, new_ses_id: str, ses_id: Optional[str] = None) -> None: - self._request_text(**self._change_app_ses_id_props(new_ses_id, ses_id or self.session_id)) + """Заменяет существующий session id на новый.""" + resolved_ses_id = self._require_session_id(ses_id) + self._request_text(**self._change_app_ses_id_props(new_ses_id, resolved_ses_id)) if ses_id is None: self.session_id = new_ses_id async def change_app_ses_id_async(self, new_ses_id: str, ses_id: Optional[str] = None) -> None: - await self._request_text_async(**self._change_app_ses_id_props(new_ses_id, ses_id or self.session_id)) + """Асинхронно заменяет существующий session id на новый.""" + resolved_ses_id = self._require_session_id(ses_id) + await self._request_text_async(**self._change_app_ses_id_props(new_ses_id, resolved_ses_id)) if ses_id is None: self.session_id = new_ses_id @staticmethod def _import_gp_props() -> RequestParams: + """Формирует параметры запроса для импорта групп и провайдеров.""" return RequestParams(path="importgroupsproviders", params={}) def import_gp(self) -> List[str]: + """Импортирует группы/провайдеров и возвращает их идентификаторы.""" return self._request_text(**self._import_gp_props()).split() async def import_gp_async(self) -> List[str]: + """Асинхронно импортирует группы/провайдеров и возвращает идентификаторы.""" response = await self._request_text_async(**self._import_gp_props()) return response.split() @@ -348,6 +482,7 @@ def _get_provider_list_props( gp: Optional[str] = None, ip: Optional[str] = None, ) -> RequestParams: + """Формирует параметры запроса для получения списка провайдеров.""" return RequestParams( path="getproviderlist", params={"login": login, "pwd": password, "gp": gp, "ip": ip}, @@ -360,6 +495,7 @@ def get_provider_list( gp: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: + """Возвращает доступных провайдеров для заданных учетных данных и контекста.""" response = self._request_text(**self._get_provider_list_props(login, password, gp, ip)) return self._as_json(response).get("providers", {}) @@ -370,23 +506,27 @@ async def get_provider_list_async( gp: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: + """Асинхронно возвращает провайдеров для заданных данных и контекста.""" response = await self._request_text_async(**self._get_provider_list_props(login, password, gp, ip)) return self._as_json(response).get("providers", {}) @staticmethod def _get_user_list_props(token: str, gp: str, ip: Optional[str] = None, pid: Optional[str] = None) -> RequestParams: + """Формирует параметры запроса для получения списка пользователей.""" return RequestParams( path="getuserlist", params={"token": token, "gp": gp, "ip": ip, "pid": pid}, ) def get_user_list(self, gp: str, ip: Optional[str] = None, pid: Optional[str] = None) -> Dict[str, Any]: - token = self._require_token(self.user_manage_token, field_name="user_manage_token") + """Возвращает список пользователей для провайдера и дополнительных фильтров.""" + token = self._require_token(self.token_user_manage, field_name="token_user_manage") response = self._request_text(**self._get_user_list_props(token, gp, ip, pid)) return self._as_json(response) async def get_user_list_async(self, gp: str, ip: Optional[str] = None, pid: Optional[str] = None) -> Dict[str, Any]: - token = self._require_token(self.user_manage_token, field_name="user_manage_token") + """Асинхронно возвращает список пользователей для провайдера и фильтров.""" + token = self._require_token(self.token_user_manage, field_name="token_user_manage") response = await self._request_text_async(**self._get_user_list_props(token, gp, ip, pid)) return self._as_json(response) @@ -395,7 +535,8 @@ def _set_settings_props( lockout_time: Optional[int] = None, login_attempts_allowed: Optional[int] = None, ) -> RequestParams: - token = self._require_token(self.set_settings_token, field_name="set_settings_token") + """Формирует параметры запроса для обновления настроек сервиса.""" + token = self._require_token(self.token_set_settings, field_name="token_set_settings") return RequestParams( path="setsettings", params={ @@ -410,6 +551,7 @@ def set_settings( lockout_time: Optional[int] = None, login_attempts_allowed: Optional[int] = None, ) -> None: + """Обновляет настройки сервиса.""" self._request_text(**self._set_settings_props(lockout_time, login_attempts_allowed)) async def set_settings_async( @@ -417,10 +559,12 @@ async def set_settings_async( lockout_time: Optional[int] = None, login_attempts_allowed: Optional[int] = None, ) -> None: + """Асинхронно обновляет настройки сервиса.""" await self._request_text_async(**self._set_settings_props(lockout_time, login_attempts_allowed)) def _create_user_props(self, payload: Dict[str, Any]) -> RequestArgs: - token = self._require_token(self.user_manage_token, field_name="user_manage_token") + """Формирует аргументы запроса для создания пользователя.""" + token = self._require_token(self.token_user_manage, field_name="token_user_manage") return RequestArgs( path="user/create", method="POST", @@ -430,15 +574,18 @@ def _create_user_props(self, payload: Dict[str, Any]) -> RequestArgs: ) def create_user(self, user: Dict[str, Any]) -> None: + """Создает пользователя из переданного payload.""" payload = self._require_user(user) self._request_text(**self._create_user_props(payload)) async def create_user_async(self, user: Dict[str, Any]) -> None: + """Асинхронно создает пользователя из переданного payload.""" payload = self._require_user(user) await self._request_text_async(**self._create_user_props(payload)) def _update_user_props(self, sid: str, user: Dict[str, Any]) -> RequestArgs: - token = self._require_token(self.user_manage_token, field_name="user_manage_token") + """Формирует аргументы запроса для обновления пользователя.""" + token = self._require_token(self.token_user_manage, field_name="token_user_manage") return RequestArgs( path=f"/user/{sid}", method="POST", @@ -448,58 +595,53 @@ def _update_user_props(self, sid: str, user: Dict[str, Any]) -> RequestArgs: ) def update_user(self, sid: str, user: Dict[str, Any]) -> None: + """Обновляет данные пользователя по sid.""" self._request_text(**self._update_user_props(sid, user)) async def update_user_async(self, sid: str, user: Dict[str, Any]) -> None: + """Асинхронно обновляет данные пользователя по sid.""" await self._request_text_async(**self._update_user_props(sid, user)) def _delete_user_props(self, sid: str) -> RequestArgs: - token = self._require_token(self.user_manage_token, field_name="user_manage_token") + """Формирует аргументы запроса для удаления пользователя.""" + token = self._require_token(self.token_user_manage, field_name="token_user_manage") return RequestArgs(path=f"/user/{sid}", method="DELETE", params={"token": token}) def delete_user(self, sid: str) -> None: + """Удаляет пользователя по sid.""" self._request_text(**self._delete_user_props(sid)) async def delete_user_async(self, sid: str) -> None: + """Асинхронно удаляет пользователя по sid.""" await self._request_text_async(**self._delete_user_props(sid)) @staticmethod def _set_state_props(ses_id: str, state: str) -> RequestArgs: + """Формирует аргументы запроса для обновления состояния.""" return RequestArgs(path="setstate", method="POST", params={"sesid": ses_id}, data=state) def set_state(self, ses_id: str, state: str) -> None: + """Устанавливает произвольное состояние в рамках сессии.""" self._request_text(**self._set_state_props(ses_id, state)) async def set_state_async(self, ses_id: str, state: str) -> None: + """Асинхронно устанавливает произвольное состояние в рамках сессии.""" await self._request_text_async(**self._set_state_props(ses_id, state)) @staticmethod def _get_state_props(ses_id: str) -> RequestParams: + """Формирует параметры запроса для получения состояния.""" return RequestParams(path="getstate", params={"sesid": ses_id}) def get_state(self, ses_id: str) -> str: + """Возвращает состояние для указанного session id.""" return self._request_text(**self._get_state_props(ses_id)) async def get_state_async(self, ses_id: str) -> str: + """Асинхронно возвращает состояние для указанного session id.""" return await self._request_text_async(**self._get_state_props(ses_id)) __all__ = [ "Mellophone", - "RequestParams", - "RequestArgs", - "xml_to_json", - "httpx", - "requests", - "HttpError", - "MissingTokenError", - "BadRequestError", - "UnauthorizedError", - "ForbiddenError", - "NotFoundError", - "ServerError", - "AsyncClientUnavailableError", - "TransportError", - "RequestTimeoutError", - "ResponseParseError", ] diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 0000000..45b82eb --- /dev/null +++ b/tests/integration/conftest.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import threading +import time +import xml.etree.ElementTree as ET +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any, Dict, Iterator, Tuple +from uuid import uuid4 + +import httpx +import pytest +import yaml + +from mellophone import Mellophone + +ROOT = Path(__file__).resolve().parents[2] +DOCKER_COMPOSE_PATH = ROOT / "docker-compose.yml" +MELLOPHONE_CONFIG_PATH = ROOT / "docker-config" / "config.xml" + + +def _extract_base_url_from_compose() -> str: + compose_text = DOCKER_COMPOSE_PATH.read_text(encoding="utf-8") + data = yaml.safe_load(compose_text) + ports = data.get("services", {}).get("mellophone", {}).get("ports", []) + if ports: + port = ports[0].split(":")[0] + return f"http://localhost:{port}/mellophone" + raise RuntimeError("Port mapping not found in docker-compose.yml") + + +def _extract_tokens_from_config() -> Tuple[str, str]: + root = ET.parse(MELLOPHONE_CONFIG_PATH).getroot() + set_token = None + user_token = None + for element in root.iter(): + tag = element.tag.split("}")[-1] + if tag == "setsettingstoken": + set_token = (element.text or "").strip() + elif tag == "getuserlisttoken": + user_token = (element.text or "").strip() + if not set_token or not user_token: + raise RuntimeError("Tokens not found in docker-config/config.xml") + return set_token, user_token + + +def _ensure_service_reachable(base_url: str) -> None: + try: + httpx.get( + f"{base_url}/isauthenticated", + params={"sesid": "integration-smoke"}, + timeout=3.0, + ) + except (httpx.HTTPError, OSError) as exc: + pytest.skip(f"Mellophone is not available at {base_url}: {exc}") + + +@pytest.fixture(scope="session") +def integration_base_url() -> str: + base_url = _extract_base_url_from_compose() + _ensure_service_reachable(base_url) + return base_url + + +@pytest.fixture(scope="session") +def integration_tokens() -> Tuple[str, str]: + return _extract_tokens_from_config() + + +@pytest.fixture +def integration_client(integration_base_url: str, integration_tokens: Tuple[str, str]) -> Mellophone: + set_token, user_token = integration_tokens + return Mellophone( + base_url=integration_base_url, + token_set_settings=set_token, + token_user_manage=user_token, + ) + + +@pytest.fixture +def integration_user(integration_client: Mellophone) -> Dict[str, str]: + unique = uuid4().hex[:8] + sid = f"it-real-{unique}" + login = f"it_real_{unique}" + password = "pwd_1" + integration_client.create_user({"sid": sid, "login": login, "password": password}) + return {"sid": sid, "login": login, "password": password} + + +@pytest.fixture +def local_error_server() -> Iterator[str]: + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 + if self.path.startswith("/unauthorized"): + self.send_response(401) + self.end_headers() + self.wfile.write(b"unauthorized") + return + if self.path.startswith("/not-found"): + self.send_response(404) + self.end_headers() + self.wfile.write(b"not found") + return + if self.path.startswith("/teapot"): + self.send_response(418) + self.end_headers() + self.wfile.write(b"teapot") + return + if self.path.startswith("/slow"): + time.sleep(0.2) + self.send_response(200) + self.end_headers() + self.wfile.write(b"slow-ok") + return + self.send_response(200) + self.end_headers() + self.wfile.write(b"ok") + + def log_message(self, format: str, *args: Any) -> None: # noqa: A003 + return + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + host, port = server.server_address # ty:ignore[invalid-assignment] + yield f"http://{host}:{port}" + finally: + server.shutdown() + thread.join(timeout=1) diff --git a/tests/integration/helpers.py b/tests/integration/helpers.py new file mode 100644 index 0000000..27a226f --- /dev/null +++ b/tests/integration/helpers.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import asyncio +from typing import Any, Dict, List + +import pytest + +from mellophone import ForbiddenError, Mellophone + + +def users_from_list(payload: Dict[str, Any]) -> List[Dict[str, str]]: + users = payload.get("users", {}).get("user", []) + if isinstance(users, dict): + return [users] + return users + + +def invoke(client: Mellophone, mode: str, method: str, *args: Any, **kwargs: Any) -> Any: + callable_obj = getattr(client, method if mode == "sync" else f"{method}_async") + result = callable_obj(*args, **kwargs) + return asyncio.run(result) if mode == "async" else result + + +def assert_credentials_valid(client: Mellophone, mode: str, login: str, password: str, sid: str) -> None: + result = invoke(client, mode, "check_credentials", login, password) + assert result.get("sid") == sid + assert result.get("login") == login + + +def assert_credentials_invalid(client: Mellophone, mode: str, login: str, password: str) -> None: + with pytest.raises(ForbiddenError): + invoke(client, mode, "check_credentials", login, password) diff --git a/tests/integration/test_client_flows.py b/tests/integration/test_client_flows.py new file mode 100644 index 0000000..3b7674e --- /dev/null +++ b/tests/integration/test_client_flows.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import json +from typing import Dict +from uuid import uuid4 + +import pytest + +from mellophone import Mellophone +from tests.integration.helpers import assert_credentials_invalid, assert_credentials_valid, invoke, users_from_list + + +@pytest.mark.parametrize("mode", ["sync", "async"], ids=["sync", "async"]) +def test_it_user_lifecycle(mode: str, integration_client: Mellophone, integration_user: Dict[str, str]) -> None: + sid = integration_user["sid"] + login = integration_user["login"] + password = integration_user["password"] + + users_payload = invoke(integration_client, mode, "get_user_list", gp="not_defined") + users = users_from_list(users_payload) + assert any(user.get("sid") == sid and user.get("login") == login for user in users) + + providers = invoke(integration_client, mode, "import_gp") + assert providers + + provider_list = invoke(integration_client, mode, "get_provider_list", login, password, gp="not_defined") + assert provider_list + + assert_credentials_valid(integration_client, mode, login, password, sid) + + session_id = invoke(integration_client, mode, "login", login, password) + auth = invoke(integration_client, mode, "is_authenticated", session_id) + assert isinstance(auth, dict) + assert auth.get("sid") == sid + + check_name_exists = invoke(integration_client, mode, "check_name", login, session_id) + assert check_name_exists.get("sid") == sid + + check_name_missing = invoke(integration_client, mode, "check_name", f"{login}_missing", session_id) + assert check_name_missing == {} + + invoke(integration_client, mode, "logout", session_id) + assert invoke(integration_client, mode, "is_authenticated", session_id) is False + + +@pytest.mark.parametrize("mode", ["sync", "async"], ids=["sync", "async"]) +def test_it_password_changes(mode: str, integration_client: Mellophone, integration_user: Dict[str, str]) -> None: + sid = integration_user["sid"] + login = integration_user["login"] + pwd_before = integration_user["password"] + pwd_2 = "pwd_2" + pwd_3 = "pwd_3" + pwd_after_update = "pwd_4" + + session_id = invoke(integration_client, mode, "login", login, pwd_before) + + invoke(integration_client, mode, "change_pwd", pwd_before, pwd_2, session_id) + assert_credentials_invalid(integration_client, mode, login, pwd_before) + assert_credentials_valid(integration_client, mode, login, pwd_2, sid) + + invoke(integration_client, mode, "change_user_pwd", login, pwd_3, session_id) + assert_credentials_invalid(integration_client, mode, login, pwd_2) + assert_credentials_valid(integration_client, mode, login, pwd_3, sid) + + invoke(integration_client, mode, "update_user", sid, {"sid": sid, "login": login, "pwd": pwd_after_update}) + assert_credentials_invalid(integration_client, mode, login, pwd_3) + assert_credentials_valid(integration_client, mode, login, pwd_after_update, sid) + + invoke(integration_client, mode, "logout", session_id) + assert invoke(integration_client, mode, "is_authenticated", session_id) is False + + +@pytest.mark.parametrize("mode", ["sync", "async"], ids=["sync", "async"]) +def test_it_state_session_settings_and_delete(mode: str, integration_client: Mellophone) -> None: + unique = uuid4().hex[:8] + sid = f"it-real-{mode}-{unique}" + login = f"it_real_{mode}_{unique}" + pwd_1 = "pwd_1" + pwd_2 = "pwd_2" + state_value = f"state_{mode}_{unique}" + + invoke(integration_client, mode, "create_user", {"sid": sid, "login": login, "password": pwd_1}) + assert_credentials_valid(integration_client, mode, login, pwd_1, sid) + + session_id = invoke(integration_client, mode, "login", login, pwd_1) + invoke(integration_client, mode, "set_state", session_id, state_value) + assert invoke(integration_client, mode, "get_state", session_id) == state_value + + invoke(integration_client, mode, "set_settings", lockout_time=30, login_attempts_allowed=5) + + new_session_id = f"{session_id}-moved" + invoke(integration_client, mode, "change_app_ses_id", new_session_id, session_id) + auth_after_change = invoke(integration_client, mode, "is_authenticated", new_session_id) + assert isinstance(auth_after_change, dict) + assert auth_after_change.get("sid") == sid + assert invoke(integration_client, mode, "is_authenticated", session_id) is False + + invoke(integration_client, mode, "change_user_pwd", login, pwd_2, new_session_id) + assert_credentials_invalid(integration_client, mode, login, pwd_1) + assert_credentials_valid(integration_client, mode, login, pwd_2, sid) + + invoke(integration_client, mode, "delete_user", sid) + assert_credentials_invalid(integration_client, mode, login, pwd_2) + + +@pytest.mark.parametrize("mode", ["sync", "async"], ids=["sync", "async"]) +def test_it_user_additional_fields(mode: str, integration_client: Mellophone) -> None: + unique = uuid4().hex[:8] + sid = f"it-extra-{mode}-{unique}" + login = f"it_extra_{mode}_{unique}" + pwd_1 = "pwd_1" + pwd_2 = "pwd_2" + + create_payload = { + "sid": sid, + "login": login, + "password": pwd_1, + "field_str": f"{login}@example.com", + "field_int": 20, + "field_float": 4.5, + "field_bool": False, + "field_none": None, + "field_str_empty": "", + "field_list_empty": [], + "field_list": ["org1"], + "field_dict": {"org1": {"name": "google", "age": 15}}, + "field_dict_empty": {}, + } + update_payload = { + "sid": sid, + "login": login, + "pwd": pwd_2, + "field_str": f"{login}+updated@example.com", + "field_int": 30, + "field_float": 9.75, + "field_bool": True, + "field_list": ["org1", "org2"], + } + + try: + invoke(integration_client, mode, "create_user", create_payload) + assert_credentials_valid(integration_client, mode, login, pwd_1, sid) + + users_after_create = users_from_list(invoke(integration_client, mode, "get_user_list", gp="not_defined")) + created_user = next((user for user in users_after_create if user.get("sid") == sid), None) + assert created_user is not None + for key in ("field_str", "field_int", "field_float", "field_bool", "field_str_empty"): + assert created_user.get(key) == str(create_payload[key]) + + for key in ("field_list", "field_list_empty", "field_dict", "field_dict_empty"): + assert created_user.get(key) == json.dumps(create_payload[key], ensure_ascii=False) + assert json.loads(created_user.get(key)) == create_payload[key] + + assert created_user.get("field_none") == "" + + invoke(integration_client, mode, "update_user", sid, update_payload) + assert_credentials_invalid(integration_client, mode, login, pwd_1) + assert_credentials_valid(integration_client, mode, login, pwd_2, sid) + + users_after_update = users_from_list(invoke(integration_client, mode, "get_user_list", gp="not_defined")) + updated_user = next((user for user in users_after_update if user.get("sid") == sid), None) + assert updated_user is not None + for key in ("field_str", "field_int", "field_float", "field_bool"): + assert updated_user.get(key) == str(update_payload[key]), key + finally: + invoke(integration_client, mode, "delete_user", sid) diff --git a/tests/integration/test_transport_errors.py b/tests/integration/test_transport_errors.py new file mode 100644 index 0000000..08bce6a --- /dev/null +++ b/tests/integration/test_transport_errors.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import socket + +import pytest + +from mellophone import HttpError, Mellophone, NotFoundError, RequestTimeoutError, TransportError, UnauthorizedError + + +def _free_tcp_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +@pytest.mark.parametrize( + ("path", "expected_exc", "expected_status"), + [ + ("unauthorized", UnauthorizedError, 401), + ("not-found", NotFoundError, 404), + ("teapot", HttpError, 418), + ], +) +def test_it_maps_http_errors( + local_error_server: str, path: str, expected_exc: type[Exception], expected_status: int +) -> None: + client = Mellophone(local_error_server) + with pytest.raises(expected_exc) as exc: + client._request_text(path) + assert getattr(exc.value, "status_code", None) == expected_status + + +def test_it_maps_request_timeout_error(local_error_server: str) -> None: + client = Mellophone(local_error_server, timeout=0.01) + with pytest.raises(RequestTimeoutError): + client._request_text("slow") + + +def test_it_maps_transport_error() -> None: + port = _free_tcp_port() + client = Mellophone(f"http://127.0.0.1:{port}") + with pytest.raises(TransportError): + client._request_text("down") diff --git a/tests/test_integration_mellophone.py b/tests/test_integration_mellophone.py deleted file mode 100644 index 22efd11..0000000 --- a/tests/test_integration_mellophone.py +++ /dev/null @@ -1,423 +0,0 @@ -from __future__ import annotations - -import asyncio -import socket -import threading -import time -import xml.etree.ElementTree as ET -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from pathlib import Path -from typing import Any, Dict, List, Tuple -from uuid import uuid4 - -import httpx -import pytest -import yaml - -from mellophone import ( - ForbiddenError, - HttpError, - Mellophone, - NotFoundError, - RequestTimeoutError, - TransportError, - UnauthorizedError, -) - -ROOT = Path(__file__).resolve().parents[1] -DOCKER_COMPOSE_PATH = ROOT / "docker-compose.yml" -MELLOPHONE_CONFIG_PATH = ROOT / "docker-config" / "config.xml" - - -@pytest.fixture(scope="session") -def integration_base_url() -> str: - base_url = _extract_base_url_from_compose() - _ensure_service_reachable(base_url) - return base_url - - -@pytest.fixture(scope="session") -def integration_tokens() -> Tuple[str, str]: - return _extract_tokens_from_config() - - -@pytest.fixture -def integration_client( - integration_base_url: str, - integration_tokens: Tuple[str, str], -) -> Mellophone: - set_token, user_token = integration_tokens - return Mellophone( - base_url=integration_base_url, - set_settings_token=set_token, - user_manage_token=user_token, - ) - - -@pytest.fixture -def integration_user( - integration_client: Mellophone, -) -> Dict[str, str]: - unique = uuid4().hex[:8] - sid = f"it-real-{unique}" - login = f"it_real_{unique}" - password = "pwd_1" - integration_client.create_user({"sid": sid, "login": login, "password": password}) - return {"sid": sid, "login": login, "password": password} - - -def _extract_base_url_from_compose() -> str: - compose_text = DOCKER_COMPOSE_PATH.read_text(encoding="utf-8") - data = yaml.safe_load(compose_text) - ports = data.get("services", {}).get("mellophone", {}).get("ports", []) - if ports: - port = ports[0].split(":")[0] - return f"http://localhost:{port}/mellophone" - raise RuntimeError("Port mapping not found in docker-compose.yml") - - -def _extract_tokens_from_config() -> Tuple[str, str]: - root = ET.parse(MELLOPHONE_CONFIG_PATH).getroot() - set_token = None - user_token = None - for element in root.iter(): - tag = element.tag.split("}")[-1] - if tag == "setsettingstoken": - set_token = (element.text or "").strip() - elif tag == "getuserlisttoken": - user_token = (element.text or "").strip() - if not set_token or not user_token: - raise RuntimeError("Tokens not found in docker-config/config.xml") - return set_token, user_token - - -def _ensure_service_reachable(base_url: str) -> None: - try: - httpx.get( - f"{base_url}/isauthenticated", - params={"sesid": "integration-smoke"}, - timeout=3.0, - ) - except (httpx.HTTPError, OSError) as exc: - pytest.skip(f"Mellophone is not available at {base_url}: {exc}") - - -def _users_from_list(payload: Dict[str, Any]) -> List[Dict[str, str]]: - users = payload.get("users", {}).get("user", []) - if isinstance(users, dict): - return [users] - return users - - -def _assert_credentials_valid( - client: Mellophone, - login: str, - password: str, - sid: str, -) -> None: - result = client.check_credentials(login, password) - assert result.get("sid") == sid - assert result.get("login") == login - - -def _assert_credentials_invalid(client: Mellophone, login: str, password: str) -> None: - with pytest.raises(ForbiddenError): - client.check_credentials(login, password) - - -def _free_tcp_port() -> int: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.bind(("127.0.0.1", 0)) - return int(sock.getsockname()[1]) - - -@pytest.fixture -def local_error_server() -> str: - class Handler(BaseHTTPRequestHandler): - def do_GET(self) -> None: # noqa: N802 - if self.path.startswith("/unauthorized"): - self.send_response(401) - self.end_headers() - self.wfile.write(b"unauthorized") - return - if self.path.startswith("/not-found"): - self.send_response(404) - self.end_headers() - self.wfile.write(b"not found") - return - if self.path.startswith("/teapot"): - self.send_response(418) - self.end_headers() - self.wfile.write(b"teapot") - return - if self.path.startswith("/slow"): - time.sleep(0.2) - self.send_response(200) - self.end_headers() - self.wfile.write(b"slow-ok") - return - self.send_response(200) - self.end_headers() - self.wfile.write(b"ok") - - def log_message(self, format: str, *args: Any) -> None: # noqa: A003 - return - - server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() - try: - host, port = server.server_address - yield f"http://{host}:{port}" - finally: - server.shutdown() - thread.join(timeout=1) - - -def test_it_sync_user_lifecycle( - integration_client: Mellophone, - integration_user: Dict[str, str], -) -> None: - sid = integration_user["sid"] - login = integration_user["login"] - password = integration_user["password"] - - users_payload = integration_client.get_user_list(gp="not_defined") - users = _users_from_list(users_payload) - assert any(user.get("sid") == sid and user.get("login") == login for user in users) - - providers = integration_client.import_gp() - assert providers - - provider_list = integration_client.get_provider_list(login, password, gp="not_defined") - assert provider_list - - _assert_credentials_valid(integration_client, login, password, sid) - - session_id = integration_client.login(login, password) - auth = integration_client.is_authenticated(session_id) - assert isinstance(auth, dict) - assert auth.get("sid") == sid - - check_name_exists = integration_client.check_name(login, session_id) - assert check_name_exists.get("sid") == sid - - check_name_missing = integration_client.check_name(f"{login}_missing", session_id) - assert check_name_missing == {} - - integration_client.logout(session_id) - assert integration_client.is_authenticated(session_id) is False - - -def test_it_sync_password_changes( - integration_client: Mellophone, - integration_user: Dict[str, str], -) -> None: - sid = integration_user["sid"] - login = integration_user["login"] - pwd_1 = integration_user["password"] - pwd_2 = "pwd_2" - pwd_3 = "pwd_3" - pwd_4 = "pwd_4" - - session_id = integration_client.login(login, pwd_1) - - integration_client.change_pwd(pwd_1, pwd_2, session_id) - _assert_credentials_invalid(integration_client, login, pwd_1) - _assert_credentials_valid(integration_client, login, pwd_2, sid) - - integration_client.change_user_pwd(login, pwd_3, session_id) - _assert_credentials_invalid(integration_client, login, pwd_2) - _assert_credentials_valid(integration_client, login, pwd_3, sid) - - integration_client.update_user(sid, {"sid": sid, "login": login, "pwd": pwd_4}) - _assert_credentials_invalid(integration_client, login, pwd_3) - _assert_credentials_valid(integration_client, login, pwd_4, sid) - - integration_client.logout(session_id) - assert integration_client.is_authenticated(session_id) is False - - -def test_it_sync_state_session_settings_and_delete( - integration_client: Mellophone, -) -> None: - unique = uuid4().hex[:8] - sid = f"it-real-sync-{unique}" - login = f"it_real_sync_{unique}" - pwd_1 = "pwd_1" - pwd_2 = "pwd_2" - - integration_client.create_user({"sid": sid, "login": login, "password": pwd_1}) - _assert_credentials_valid(integration_client, login, pwd_1, sid) - - session_id = integration_client.login(login, pwd_1) - state_value = f"state_sync_{unique}" - integration_client.set_state(session_id, state_value) - assert integration_client.get_state(session_id) == state_value - - integration_client.set_settings(lockout_time=30, login_attempts_allowed=5) - - new_session_id = f"{session_id}-moved" - integration_client.change_app_ses_id(new_session_id, session_id) - auth_after_change = integration_client.is_authenticated(new_session_id) - assert isinstance(auth_after_change, dict) - assert auth_after_change.get("sid") == sid - assert integration_client.is_authenticated(session_id) is False - - integration_client.change_user_pwd(login, pwd_2, new_session_id) - _assert_credentials_invalid(integration_client, login, pwd_1) - _assert_credentials_valid(integration_client, login, pwd_2, sid) - - integration_client.delete_user(sid) - _assert_credentials_invalid(integration_client, login, pwd_2) - - -def test_it_async_user_lifecycle( - integration_client: Mellophone, - integration_user: Dict[str, str], -) -> None: - sid = integration_user["sid"] - login = integration_user["login"] - password = integration_user["password"] - - async def _run() -> None: - providers_async = await integration_client.import_gp_async() - assert providers_async - - provider_list_async = await integration_client.get_provider_list_async( - login, - password, - gp="not_defined", - ) - assert provider_list_async - - users_payload_async = await integration_client.get_user_list_async(gp="not_defined") - users_async = _users_from_list(users_payload_async) - assert any(user.get("sid") == sid and user.get("login") == login for user in users_async) - - check_credentials_async = await integration_client.check_credentials_async( - login, - password, - ) - assert check_credentials_async.get("sid") == sid - - session_async = await integration_client.login_async(login, password) - auth_async = await integration_client.is_authenticated_async(session_async) - assert isinstance(auth_async, dict) - assert auth_async.get("sid") == sid - - check_name_async = await integration_client.check_name_async(login, session_async) - assert check_name_async.get("sid") == sid - - await integration_client.logout_async(session_async) - assert await integration_client.is_authenticated_async(session_async) is False - - asyncio.run(_run()) - - -def test_it_async_password_changes( - integration_client: Mellophone, - integration_user: Dict[str, str], -) -> None: - sid = integration_user["sid"] - login = integration_user["login"] - pwd_4 = integration_user["password"] - pwd_5 = "pwd_5" - - async def _run() -> None: - session_async = await integration_client.login_async(login, pwd_4) - - await integration_client.change_pwd_async(pwd_4, pwd_5, session_async) - with pytest.raises(ForbiddenError): - await integration_client.check_credentials_async(login, pwd_4) - - valid_after_change = await integration_client.check_credentials_async(login, pwd_5) - assert valid_after_change.get("sid") == sid - - await integration_client.update_user_async(sid, {"sid": sid, "login": login, "pwd": pwd_4}) - with pytest.raises(ForbiddenError): - await integration_client.check_credentials_async(login, pwd_5) - - valid_after_update = await integration_client.check_credentials_async(login, pwd_4) - assert valid_after_update.get("sid") == sid - - await integration_client.logout_async(session_async) - assert await integration_client.is_authenticated_async(session_async) is False - - asyncio.run(_run()) - - -def test_it_async_state_session_settings_and_delete( - integration_client: Mellophone, -) -> None: - unique = uuid4().hex[:8] - sid = f"it-real-async-{unique}" - login = f"it_real_async_{unique}" - pwd_1 = "pwd_1" - pwd_2 = "pwd_2" - - async def _run() -> None: - await integration_client.create_user_async({"sid": sid, "login": login, "password": pwd_1}) - - valid_before = await integration_client.check_credentials_async(login, pwd_1) - assert valid_before.get("sid") == sid - - session_id = await integration_client.login_async(login, pwd_1) - state_value = f"state_async_{unique}" - await integration_client.set_state_async(session_id, state_value) - assert await integration_client.get_state_async(session_id) == state_value - - await integration_client.set_settings_async(lockout_time=30, login_attempts_allowed=5) - - new_session_id = f"{session_id}-moved" - await integration_client.change_app_ses_id_async(new_session_id, session_id) - auth_after_change = await integration_client.is_authenticated_async(new_session_id) - assert isinstance(auth_after_change, dict) - assert auth_after_change.get("sid") == sid - assert await integration_client.is_authenticated_async(session_id) is False - - await integration_client.change_user_pwd_async(login, pwd_2, new_session_id) - with pytest.raises(ForbiddenError): - await integration_client.check_credentials_async(login, pwd_1) - - valid_after_change = await integration_client.check_credentials_async(login, pwd_2) - assert valid_after_change.get("sid") == sid - - await integration_client.delete_user_async(sid) - with pytest.raises(ForbiddenError): - await integration_client.check_credentials_async(login, pwd_2) - - asyncio.run(_run()) - - -@pytest.mark.parametrize( - ("path", "expected_exc", "expected_status"), - [ - ("unauthorized", UnauthorizedError, 401), - ("not-found", NotFoundError, 404), - ("teapot", HttpError, 418), - ], -) -def test_it_maps_http_errors( - local_error_server: str, - path: str, - expected_exc: type[Exception], - expected_status: int, -) -> None: - client = Mellophone(local_error_server) - with pytest.raises(expected_exc) as exc: - client._request_text(path) - assert getattr(exc.value, "status_code", None) == expected_status - - -def test_it_maps_request_timeout_error(local_error_server: str) -> None: - client = Mellophone(local_error_server, timeout=0.01) - with pytest.raises(RequestTimeoutError): - client._request_text("slow") - - -def test_it_maps_transport_error() -> None: - port = _free_tcp_port() - client = Mellophone(f"http://127.0.0.1:{port}") - with pytest.raises(TransportError): - client._request_text("down") diff --git a/tests/test_mellophone.py b/tests/test_mellophone.py deleted file mode 100644 index e3d932c..0000000 --- a/tests/test_mellophone.py +++ /dev/null @@ -1,461 +0,0 @@ -import asyncio -import xml.etree.ElementTree as ET -from typing import Dict, List - -import httpx -import pytest - -import mellophone -from mellophone import client as mellophone_client -from mellophone.utils import xml_to_json - - -class SyncClientStub: - def __init__(self, response: httpx.Response, calls: List[Dict]): - self.response = response - self.calls = calls - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc, tb): - return False - - def request(self, method, url, content=None, headers=None): - self.calls.append({"method": method, "url": url, "content": content, "headers": headers}) - return self.response - - -class AsyncClientStub: - def __init__(self, response: httpx.Response, calls: List[Dict]): - self.response = response - self.calls = calls - - async def __aenter__(self): - return self - - async def __aexit__(self, exc_type, exc, tb): - return False - - async def request(self, method, url, content=None, headers=None): - self.calls.append({"method": method, "url": url, "content": content, "headers": headers}) - return self.response - - -class SyncTimeoutClientStub: - def __enter__(self): - return self - - def __exit__(self, exc_type, exc, tb): - return False - - def request(self, method, url, content=None, headers=None): - raise httpx.TimeoutException("timeout") - - -class AsyncTimeoutClientStub: - async def __aenter__(self): - return self - - async def __aexit__(self, exc_type, exc, tb): - return False - - async def request(self, method, url, content=None, headers=None): - raise httpx.TimeoutException("timeout") - - -class RequestsSessionStub: - def __init__(self, response, calls: List[Dict]): - self.response = response - self.calls = calls - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc, tb): - return False - - def request(self, method, url, data=None, headers=None, timeout=None): - self.calls.append( - { - "method": method, - "url": url, - "content": data, - "headers": headers, - "timeout": timeout, - } - ) - return self.response - - -def _response(status_code: int, text: str = "", url: str = "http://test.local/"): - return httpx.Response(status_code, text=text, request=httpx.Request("GET", url)) - - -@pytest.fixture -def mock_sync_client(monkeypatch): - def _install(response: httpx.Response) -> List[Dict]: - calls: List[Dict] = [] - - def client_factory(*_args, **_kwargs): - return SyncClientStub(response, calls) - - monkeypatch.setattr(mellophone_client.httpx, "Client", client_factory) - return calls - - return _install - - -@pytest.fixture -def mock_async_client(monkeypatch): - def _install(response: httpx.Response) -> List[Dict]: - calls: List[Dict] = [] - - def async_client_factory(*_args, **_kwargs): - return AsyncClientStub(response, calls) - - monkeypatch.setattr(mellophone_client.httpx, "AsyncClient", async_client_factory) - return calls - - return _install - - -@pytest.fixture -def mock_requests_session(monkeypatch): - def _install(response) -> List[Dict]: - if mellophone_client.requests is None: - pytest.skip("requests is not installed") - calls: List[Dict] = [] - - def session_factory(*_args, **_kwargs): - return RequestsSessionStub(response, calls) - - monkeypatch.setattr(mellophone_client.requests, "Session", session_factory) - return calls - - return _install - - -def test_xml_to_json_with_repeated_tags(): - xml = "" - result = xml_to_json(xml) - assert result == {"users": {"user": [{"login": "a"}, {"login": "b"}]}} - - -def test_client_options_are_stored(): - client = mellophone.Mellophone( - "http://example.com", - set_settings_token="set-token", - user_manage_token="user-token", - session_id="ses-1", - timeout=7.5, - ) - - assert client.base_url == "http://example.com" - assert client.set_settings_token == "set-token" - assert client.user_manage_token == "user-token" - assert client.session_id == "ses-1" - assert client.timeout == 7.5 - - -def test_build_url_skips_none_params(): - client = mellophone.Mellophone("http://example.com") - - url = client._build_url("path", {"a": 1, "b": None, "c": "x"}) - - assert url == "http://example.com/path?a=1&c=x" - - -def test_login_sync_sets_session_and_sends_request(mock_sync_client): - calls = mock_sync_client(_response(200)) - - client = mellophone.Mellophone("http://example.com", session_id=None) - ses_id = client.login("john", "secret", ses_id="ses-1", gp="grp", ip="127.0.0.1") - - assert ses_id == "ses-1" - assert client.session_id == "ses-1" - assert calls - assert calls[0]["method"] == "GET" - assert "/login?" in calls[0]["url"] - assert "sesid=ses-1" in calls[0]["url"] - assert "login=john" in calls[0]["url"] - - -def test_is_authenticated_returns_false_on_forbidden(mock_sync_client): - mock_sync_client(_response(403, "forbidden")) - - client = mellophone.Mellophone("http://example.com", session_id="ses-1") - assert client.is_authenticated() is False - - -def test_bad_request_maps_to_domain_error(mock_sync_client): - mock_sync_client(_response(400, "bad request")) - - client = mellophone.Mellophone("http://example.com") - with pytest.raises(mellophone.BadRequestError) as exc: - client.login("john", "secret", ses_id="ses-1") - - assert exc.value.status_code == 400 - - -def test_server_error_maps_to_domain_error(mock_sync_client): - mock_sync_client(_response(500, "server error")) - - client = mellophone.Mellophone("http://example.com") - with pytest.raises(mellophone.ServerError) as exc: - client.login("john", "secret", ses_id="ses-1") - - assert exc.value.status_code == 500 - - -def test_create_user_converts_password_to_pwd_and_sends_xml(mock_sync_client): - calls = mock_sync_client(_response(200, "")) - - client = mellophone.Mellophone( - "http://example.com", - user_manage_token="token-1", - ) - client.create_user({"sid": "1", "login": "neo", "password": "1234"}) - assert calls - assert calls[0]["method"] == "POST" - assert "token=token-1" in calls[0]["url"] - assert calls[0]["headers"] == {"Content-Type": "application/xml"} - xml = ET.fromstring(calls[0]["content"]) - assert xml.tag == "user" - assert xml.attrib["pwd"] == "1234" - assert "password" not in xml.attrib - - -def test_create_user_empty_payload_raises(): - client = mellophone.Mellophone("http://example.com") - with pytest.raises(ValueError): - client.create_user({}) - - -def test_set_settings_uses_client_token(mock_sync_client): - calls = mock_sync_client(_response(200)) - - client = mellophone.Mellophone("http://example.com", set_settings_token="set-token") - client.set_settings(lockout_time=30, login_attempts_allowed=5) - - assert calls - assert calls[0]["method"] == "GET" - assert "/setsettings?" in calls[0]["url"] - assert "token=set-token" in calls[0]["url"] - assert "lockouttime=30" in calls[0]["url"] - assert "loginattemptsallowed=5" in calls[0]["url"] - - -def test_update_user_uses_client_user_manage_token(mock_sync_client): - calls = mock_sync_client(_response(200)) - - client = mellophone.Mellophone("http://example.com", user_manage_token="token-1") - client.update_user("u-1", {"sid": "u-1", "login": "neo", "pwd": "1234"}) - - assert calls - assert calls[0]["method"] == "POST" - assert "/user/u-1?" in calls[0]["url"] - assert "token=token-1" in calls[0]["url"] - assert calls[0]["headers"] == {"Content-Type": "application/xml"} - - -def test_change_user_pwd_sends_expected_params(mock_sync_client): - calls = mock_sync_client(_response(200)) - - client = mellophone.Mellophone("http://example.com", session_id="ses-1") - client.change_user_pwd("neo", "new-secret") - - assert calls - assert calls[0]["method"] == "GET" - assert "/changeuserpwd?" in calls[0]["url"] - assert "sesid=ses-1" in calls[0]["url"] - assert "username=neo" in calls[0]["url"] - assert "newpwd=new-secret" in calls[0]["url"] - assert "oldpwd=" not in calls[0]["url"] - - -def test_change_app_ses_id(mock_sync_client): - calls = mock_sync_client(_response(200)) - - client = mellophone.Mellophone("http://example.com", session_id="ses-self") - client.change_app_ses_id("ses-new") - client.change_app_ses_id("ses-foreign-new", ses_id="ses-foreign-old") - - assert client.session_id == "ses-new" - assert len(calls) == 2 - assert "oldsesid=ses-self" in calls[0]["url"] - assert "newsesid=ses-new" in calls[0]["url"] - assert "oldsesid=ses-foreign-old" in calls[1]["url"] - assert "newsesid=ses-foreign-new" in calls[1]["url"] - - -def test_delete_user(mock_sync_client): - calls = mock_sync_client(_response(200)) - - client = mellophone.Mellophone("http://example.com", user_manage_token="token-1") - client.delete_user("u-1") - - assert calls - assert calls[0]["method"] == "DELETE" - assert "/user/u-1?" in calls[0]["url"] - assert "token=token-1" in calls[0]["url"] - - -def test_set_state_and_get_state_sync(mock_sync_client): - calls = mock_sync_client(_response(200, "state-value")) - - client = mellophone.Mellophone("http://example.com") - client.set_state("ses-1", "new-state") - state = client.get_state("ses-1") - - assert state == "state-value" - assert len(calls) == 2 - assert calls[0]["method"] == "POST" - assert "/setstate?" in calls[0]["url"] - assert "sesid=ses-1" in calls[0]["url"] - assert calls[0]["content"] == "new-state" - assert calls[1]["method"] == "GET" - assert "/getstate?" in calls[1]["url"] - assert "sesid=ses-1" in calls[1]["url"] - - -def test_login_async_sets_session_and_sends_request(mock_async_client): - calls = mock_async_client(_response(200)) - - client = mellophone.Mellophone("http://example.com") - ses_id = asyncio.run(client.login_async("john", "secret", ses_id="ses-async")) - - assert ses_id == "ses-async" - assert client.session_id == "ses-async" - assert calls - assert calls[0]["method"] == "GET" - assert "sesid=ses-async" in calls[0]["url"] - - -def test_check_credentials_async_parses_xml(mock_async_client): - calls = mock_async_client(_response(200, "")) - - client = mellophone.Mellophone("http://example.com") - result = asyncio.run(client.check_credentials_async("neo", "1234")) - - assert result == {"sid": "1", "login": "neo"} - assert calls - - -def test_delete_user_async(mock_async_client): - calls = mock_async_client(_response(200)) - - client = mellophone.Mellophone("http://example.com", user_manage_token="token-2") - asyncio.run(client.delete_user_async("u-2")) - - assert calls - assert calls[0]["method"] == "DELETE" - assert "/user/u-2?" in calls[0]["url"] - assert "token=token-2" in calls[0]["url"] - - -def test_change_app_ses_id_async(mock_async_client): - calls = mock_async_client(_response(200)) - - client = mellophone.Mellophone("http://example.com", session_id="ses-self") - asyncio.run(client.change_app_ses_id_async("ses-new")) - asyncio.run(client.change_app_ses_id_async("ses-foreign-new", ses_id="ses-foreign-old")) - - assert client.session_id == "ses-new" - assert len(calls) == 2 - assert "oldsesid=ses-self" in calls[0]["url"] - assert "newsesid=ses-new" in calls[0]["url"] - assert "oldsesid=ses-foreign-old" in calls[1]["url"] - assert "newsesid=ses-foreign-new" in calls[1]["url"] - - -def test_set_state_and_get_state_async(mock_async_client): - calls = mock_async_client(_response(200, "async-state")) - - client = mellophone.Mellophone("http://example.com") - asyncio.run(client.set_state_async("ses-async", "payload")) - state = asyncio.run(client.get_state_async("ses-async")) - - assert state == "async-state" - assert len(calls) == 2 - assert calls[0]["method"] == "POST" - assert "/setstate?" in calls[0]["url"] - assert calls[0]["content"] == "payload" - assert calls[1]["method"] == "GET" - assert "/getstate?" in calls[1]["url"] - - -def test_parse_error_maps_to_response_parse_error(mock_sync_client): - mock_sync_client(_response(200, " index + call = calls[index] + if method is not None: + assert call["method"] == method + for part in url_contains: + assert part in call["url"] + for part in url_not_contains: + assert part not in call["url"] + if headers is not None: + assert call["headers"] == headers + if content is not None: + assert call["content"] == content + return call + + +@pytest.fixture(params=["sync", "async"], ids=["sync", "async"]) +def mode_and_mock(request, mock_sync_client, mock_async_client): + mode = request.param + install = mock_sync_client if mode == "sync" else mock_async_client + return mode, install + + +def test_login_sets_session_and_sends_request(mode_and_mock, response_factory): + mode, install = mode_and_mock + calls = install(response_factory(200)) + ses_id = "ses-1" if mode == "sync" else "ses-async" + + client = mellophone.Mellophone("http://example.com", session_id=None) + result = _call(client, mode, "login", "john", "secret", ses_id=ses_id, gp="grp", ip="127.0.0.1") + + assert result == ses_id + assert client.session_id == ses_id + _assert_call(calls, method="GET", url_contains=("/login?", f"sesid={ses_id}", "login=john")) + + +def test_is_authenticated_returns_false_on_forbidden(mode_and_mock, response_factory): + mode, install = mode_and_mock + install(response_factory(403, "forbidden")) + + client = mellophone.Mellophone("http://example.com", session_id="ses-1") + assert _call(client, mode, "is_authenticated") is False + + +def test_bad_request_maps_to_domain_error(mode_and_mock, response_factory): + mode, install = mode_and_mock + install(response_factory(400, "bad request")) + + client = mellophone.Mellophone("http://example.com") + with pytest.raises(mellophone.BadRequestError) as exc: + _call(client, mode, "login", "john", "secret", ses_id="ses-1") + + assert exc.value.status_code == 400 + + +def test_server_error_maps_to_domain_error(mode_and_mock, response_factory): + mode, install = mode_and_mock + install(response_factory(500, "server error")) + + client = mellophone.Mellophone("http://example.com") + with pytest.raises(mellophone.ServerError) as exc: + _call(client, mode, "login", "john", "secret", ses_id="ses-1") + + assert exc.value.status_code == 500 + + +def test_check_credentials_parses_xml(mode_and_mock, response_factory): + mode, install = mode_and_mock + calls = install(response_factory(200, "")) + + client = mellophone.Mellophone("http://example.com") + result = _call(client, mode, "check_credentials", "neo", "1234") + + assert result == {"sid": "1", "login": "neo"} + assert calls + + +def test_create_user_converts_password_to_pwd_and_sends_xml(mode_and_mock, response_factory): + mode, install = mode_and_mock + calls = install(response_factory(200, "")) + + client = mellophone.Mellophone("http://example.com", token_user_manage="token-1") + _call(client, mode, "create_user", {"sid": "1", "login": "neo", "password": "1234"}) + + call = _assert_call( + calls, + method="POST", + url_contains=("token=token-1",), + headers={"Content-Type": "application/xml"}, + ) + xml = ET.fromstring(call["content"]) + assert xml.tag == "user" + assert xml.attrib["pwd"] == "1234" + assert "password" not in xml.attrib + + +def test_create_user_empty_payload_raises(mode_and_mock): + mode, _ = mode_and_mock + client = mellophone.Mellophone("http://example.com") + with pytest.raises(ValueError): + _call(client, mode, "create_user", {}) + + +def test_set_settings_uses_client_token(mode_and_mock, response_factory): + mode, install = mode_and_mock + calls = install(response_factory(200)) + + client = mellophone.Mellophone("http://example.com", token_set_settings="set-token") + _call(client, mode, "set_settings", lockout_time=30, login_attempts_allowed=5) + + _assert_call( + calls, + method="GET", + url_contains=("/setsettings?", "token=set-token", "lockouttime=30", "loginattemptsallowed=5"), + ) + + +def test_update_user_uses_client_user_manage_token(mode_and_mock, response_factory): + mode, install = mode_and_mock + calls = install(response_factory(200)) + + client = mellophone.Mellophone("http://example.com", token_user_manage="token-1") + _call(client, mode, "update_user", "u-1", {"sid": "u-1", "login": "neo", "pwd": "1234"}) + + _assert_call( + calls, + method="POST", + url_contains=("/user/u-1?", "token=token-1"), + headers={"Content-Type": "application/xml"}, + ) + + +def test_change_user_pwd_sends_expected_params(mode_and_mock, response_factory): + mode, install = mode_and_mock + calls = install(response_factory(200)) + + client = mellophone.Mellophone("http://example.com", session_id="ses-1") + _call(client, mode, "change_user_pwd", "neo", "new-secret") + + _assert_call( + calls, + method="GET", + url_contains=("/changeuserpwd?", "sesid=ses-1", "username=neo", "newpwd=new-secret"), + url_not_contains=("oldpwd=",), + ) + + +def test_change_app_ses_id(mode_and_mock, response_factory): + mode, install = mode_and_mock + calls = install(response_factory(200)) + + client = mellophone.Mellophone("http://example.com", session_id="ses-self") + _call(client, mode, "change_app_ses_id", "ses-new") + _call(client, mode, "change_app_ses_id", "ses-foreign-new", ses_id="ses-foreign-old") + + assert client.session_id == "ses-new" + assert len(calls) == 2 + _assert_call(calls, index=0, url_contains=("oldsesid=ses-self", "newsesid=ses-new")) + _assert_call(calls, index=1, url_contains=("oldsesid=ses-foreign-old", "newsesid=ses-foreign-new")) + + +def test_delete_user(mode_and_mock, response_factory): + mode, install = mode_and_mock + calls = install(response_factory(200)) + + client = mellophone.Mellophone("http://example.com", token_user_manage="token-1") + _call(client, mode, "delete_user", "u-1") + + _assert_call(calls, method="DELETE", url_contains=("/user/u-1?", "token=token-1")) + + +def test_set_state_and_get_state(mode_and_mock, response_factory): + mode, install = mode_and_mock + calls = install(response_factory(200, "state-value")) + + client = mellophone.Mellophone("http://example.com") + ses_id = "ses-1" if mode == "sync" else "ses-async" + payload = "new-state" if mode == "sync" else "payload" + _call(client, mode, "set_state", ses_id, payload) + state = _call(client, mode, "get_state", ses_id) + + assert state == "state-value" + assert len(calls) == 2 + _assert_call(calls, index=0, method="POST", url_contains=("/setstate?", f"sesid={ses_id}"), content=payload) + _assert_call(calls, index=1, method="GET", url_contains=("/getstate?", f"sesid={ses_id}")) + + +def test_parse_error_maps_to_response_parse_error(mode_and_mock, response_factory): + mode, install = mode_and_mock + install(response_factory(200, "