From 676d044df29596f2bd3850b39b9d4f8fd9b8e9be Mon Sep 17 00:00:00 2001 From: stepan TeSt Date: Mon, 6 Apr 2026 14:06:27 +0300 Subject: [PATCH] initial commit --- features/FirstBahave.feature | 11 +++ features/PlaceInfoKVS.feature | 10 ++ features/steps/placeInfokvs_Steps.py | 40 ++++++++ features/steps/place_steps.py | 31 +++++++ features/steps/place_steps_info.py | 26 ++++++ main.py | 10 ++ pyrightconfig.json | 7 ++ pytest.ini | 10 ++ requirements.txt | 3 + scripts/generate_allure_report.ps1 | 27 ++++++ tests/TesFindPlaceInfo.py | 113 +++++++++++++++++++++++ worklib/QueryData.py | 58 ++++++++++++ worklib/__init__.py | 6 ++ worklib/admin_data.py | 76 +++++++++++++++ worklib/auth_as_employer.py | 56 +++++++++++ worklib/findplaceinfo/find_place_data.py | 62 +++++++++++++ 16 files changed, 546 insertions(+) create mode 100644 features/FirstBahave.feature create mode 100644 features/PlaceInfoKVS.feature create mode 100644 features/steps/placeInfokvs_Steps.py create mode 100644 features/steps/place_steps.py create mode 100644 features/steps/place_steps_info.py create mode 100644 main.py create mode 100644 pyrightconfig.json create mode 100644 pytest.ini create mode 100644 requirements.txt create mode 100644 scripts/generate_allure_report.ps1 create mode 100644 tests/TesFindPlaceInfo.py create mode 100644 worklib/QueryData.py create mode 100644 worklib/__init__.py create mode 100644 worklib/admin_data.py create mode 100644 worklib/auth_as_employer.py create mode 100644 worklib/findplaceinfo/find_place_data.py diff --git a/features/FirstBahave.feature b/features/FirstBahave.feature new file mode 100644 index 0000000..c972ab1 --- /dev/null +++ b/features/FirstBahave.feature @@ -0,0 +1,11 @@ +Feature: Place info (REST/GraphQL/WebSocket) + + + Scenario: Authorize as employer + When get access token + Then access token is valid + + Scenario: Get place info + When get place info + Then place info is valid for query data + diff --git a/features/PlaceInfoKVS.feature b/features/PlaceInfoKVS.feature new file mode 100644 index 0000000..9872d8b --- /dev/null +++ b/features/PlaceInfoKVS.feature @@ -0,0 +1,10 @@ +Feature: Place info KVS(REST/GraphQL/WebSocket) + + + Scenario: Authorize as employer + When get access token + Then access token is valid + + Scenario: Get place info + When get place info kvs + Then place info is valid \ No newline at end of file diff --git a/features/steps/placeInfokvs_Steps.py b/features/steps/placeInfokvs_Steps.py new file mode 100644 index 0000000..cc4d10f --- /dev/null +++ b/features/steps/placeInfokvs_Steps.py @@ -0,0 +1,40 @@ +# pyright: reportCallIssue=false + +import os +from typing import Any, Final +from behave import given, when, then +from worklib.QueryData import kvs_query_data, kvs_query_data_place_id +from worklib import admin_data +from worklib.auth_as_employer import get_access_token +from worklib.findplaceinfo.find_place_data import fetch_place_members + +expected_result = { + "members": [ + { + "id": "bb368ee9-c15f-40ef-acb0-c466df47d096", + "parent_id": None, + "user": { + "username": "+79999956657" + } + } + ] +} + + + + +@when("get place info kvs") # pyright: ignore[reportGeneralTypeIssues] +def step_get_place_info(context): + token = getattr(context, "access_token", None) or admin_data.get_or_create_user("tester").access_token + data = fetch_place_members(access_token=token, query=kvs_query_data()["query"], variables=kvs_query_data_place_id()["variables"]) + context.place_info = data + + +@then("place info is valid") # pyright: ignore[reportGeneralTypeIssues] +def step_place_info_valid(context): + data = getattr(context, "place_info", None) + assert isinstance(data, dict), "Ответ GraphQL не dict" + assert "data" in data or "place" in str(data), f"Не похоже на успешный GraphQL ответ: {data}" + assert data["data"]["place"]["results"][0]["members"][0]["id"] == expected_result["members"][0]["id"] + assert data["data"]["place"]["results"][0]["members"][0]["parent_id"] == expected_result["members"][0]["parent_id"] + assert data["data"]["place"]["results"][0]["members"][0]["user"]["username"] == expected_result["members"][0]["user"]["username"] \ No newline at end of file diff --git a/features/steps/place_steps.py b/features/steps/place_steps.py new file mode 100644 index 0000000..c23a87d --- /dev/null +++ b/features/steps/place_steps.py @@ -0,0 +1,31 @@ +# pyright: reportCallIssue=false + +import os +from typing import Any, Final +from behave import given, when, then +from worklib.QueryData import query_data, query_data_place_id_variables +from worklib import admin_data +from worklib.auth_as_employer import get_access_token +from worklib.findplaceinfo.find_place_data import fetch_place_members + + + + # pyright: ignore[reportGeneralTypeIssues] + + + +@when("get access token") # pyright: ignore[reportGeneralTypeIssues] +def step_get_access_token(context): + if not getattr(context, "access_token", None): + token = admin_data.get_access_token_from_env() + context.access_token = token + admin_data.get_or_create_user("tester").access_token = token + + +@then("access token is valid") # pyright: ignore[reportGeneralTypeIssues] +def step_token_is_valid(context): + token = getattr(context, "access_token", None) + assert isinstance(token, str) and token.strip(), f"access_token пустой/не строка: {token}" + + + diff --git a/features/steps/place_steps_info.py b/features/steps/place_steps_info.py new file mode 100644 index 0000000..01d2383 --- /dev/null +++ b/features/steps/place_steps_info.py @@ -0,0 +1,26 @@ +from behave import given, when, then +from typing import Final +from worklib import admin_data +from worklib.findplaceinfo.find_place_data import fetch_place_members +from worklib.QueryData import query_data, query_data_place_id_variables +# pyright: ignore[reportGeneralTypeIssues] + +_EXPECTED_RESULT: Final[dict[str, str]] = { + "id": "682b071a163ac2a0995355be", + "place_type": "street", + "name": "ул. Мебельная", +} +@when("get place info") # pyright: ignore[reportGeneralTypeIssues] +def step_get_place_info(context): + token = getattr(context, "access_token", None) or admin_data.get_or_create_user("tester").access_token + data = fetch_place_members(access_token=token, query=query_data()["query"], variables=query_data_place_id_variables()["variables"]) + context.place_info = data + +@then("place info is valid for query data") # pyright: ignore[reportGeneralTypeIssues] +def step_place_info_valid(context): + data = getattr(context, "place_info", None) + assert isinstance(data, dict), "Ответ GraphQL не dict" + assert "data" in data or "place" in str(data), f"Не похоже на успешный GraphQL ответ: {data}" + assert data["data"]["place"]["results"][0]["id"] == _EXPECTED_RESULT["id"] + assert data["data"]["place"]["results"][0]["place_type"] == _EXPECTED_RESULT["place_type"] + assert data["data"]["place"]["results"][0]["name"] == _EXPECTED_RESULT["name"] diff --git a/main.py b/main.py new file mode 100644 index 0000000..9982858 --- /dev/null +++ b/main.py @@ -0,0 +1,10 @@ + +import worklib.auth_as_employer +from tests.TesFindPlaceInfo import TestFindPlaceInfo +test = TestFindPlaceInfo() + + + + + + diff --git a/pyrightconfig.json b/pyrightconfig.json new file mode 100644 index 0000000..2a90b9b --- /dev/null +++ b/pyrightconfig.json @@ -0,0 +1,7 @@ +{ + "venvPath": ".", + "venv": ".venv", + "typeCheckingMode": "basic", + "reportMissingImports": true, + "reportMissingTypeStubs": "none" +} diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..a66bda2 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,10 @@ +[pytest] +testpaths = tests +python_files = *.py +python_classes = Test* +python_functions = test_* +addopts = + --alluredir=allure-results + --clean-alluredir +markers = + integration: тесты с реальными HTTP-запросами (auth + GraphQL) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..eaa552f --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +behave +pytest>=7.0 +allure-pytest>=2.13.0 diff --git a/scripts/generate_allure_report.ps1 b/scripts/generate_allure_report.ps1 new file mode 100644 index 0000000..6c2a1fe --- /dev/null +++ b/scripts/generate_allure_report.ps1 @@ -0,0 +1,27 @@ +# Запуск тестов с записью результатов Allure и генерация HTML-отчёта (нужен Allure CLI в PATH). +# Установка Allure 2: https://github.com/allure-framework/allure2/releases +# Добавьте bin каталог разархивированного allure в переменную PATH. + +$ErrorActionPreference = "Stop" +$Root = Split-Path -Parent $PSScriptRoot +Set-Location $Root + +$venvPy = Join-Path $Root ".venv\Scripts\python.exe" +if (-not (Test-Path $venvPy)) { + Write-Error "Нет виртуального окружения .venv. Выполните: python -m venv .venv && .\.venv\Scripts\python -m pip install -r requirements.txt" +} + +& $venvPy -m pytest tests\TesFindPlaceInfo.py -v + +$allure = Get-Command allure -ErrorAction SilentlyContinue +if (-not $allure) { + Write-Host "" + Write-Host "Allure CLI не найден в PATH. Установите Allure 2 и добавьте bin в PATH." + Write-Host "Сырые результаты уже в папке: allure-results" + Write-Host "После установки CLI выполните: allure generate allure-results -o allure-report --clean" + exit 0 +} + +& allure generate allure-results -o allure-report --clean +Write-Host "" +Write-Host "Отчёт: file:///$((Join-Path $Root 'allure-report\index.html') -replace '\\', '/')" diff --git a/tests/TesFindPlaceInfo.py b/tests/TesFindPlaceInfo.py new file mode 100644 index 0000000..d61e8a0 --- /dev/null +++ b/tests/TesFindPlaceInfo.py @@ -0,0 +1,113 @@ +"""Проверка ответа GraphQL place: структура и значения полей results[ ] + Allure-отчёт.""" + +from __future__ import annotations + +import json +from typing import Any, Final + +import allure # pyright: ignore[reportMissingImports] +import pytest +from allure_commons.types import AttachmentType # pyright: ignore[reportMissingImports] + +from worklib.findplaceinfo.find_place_data import fetch_place_members + + +# Ожидаемый фрагмент ответа (ключи и значения должны совпадать хотя бы в одной записи results). +_EXPECTED_RESULT: Final[dict[str, str]] = { + "id": "682b071a163ac2a0995355be", + "place_type": "street", + "name": "ул. Мебельная", +} + +ALLURE_EPIC = "Place API" +ALLURE_FEATURE = "GraphQL place" +ALLURE_STORY = "Формат ответа и значения полей" + + +@allure.epic(ALLURE_EPIC) +@allure.feature(ALLURE_FEATURE) +@allure.story(ALLURE_STORY) +@pytest.mark.integration +class TestFindPlaceInfo: + """Интеграционный тест: реальный запрос и сверка формата/значений.""" + + @pytest.fixture(scope="class") + def place_response(self) -> dict[str, Any]: + with allure.step("Запрос place через GraphQL (fetch_place_members)"): + payload = fetch_place_members() + allure.attach( + json.dumps(payload, ensure_ascii=False, indent=2), + name="Ответ GraphQL (raw JSON)", + attachment_type=AttachmentType.JSON, + ) + return payload + + @allure.title("Структура: data → place → results[] с полями id, place_type, name") + @allure.severity(allure.severity_level.NORMAL) + def test_response_has_place_shape(self, place_response: dict[str, Any]) -> None: + with allure.step("Проверка вложенности data.place.results"): + results = _extract_results(place_response) + allure.attach( + json.dumps(results, ensure_ascii=False, indent=2), + name="results (извлечено)", + attachment_type=AttachmentType.JSON, + ) + assert len(results) >= 1, "results должен содержать хотя бы один элемент" + for key in ("id", "place_type", "name"): + assert key in results[0], f'В элементе results[0] должен быть ключ "{key}"' + + @allure.title("Значения id, place_type, name совпадают с эталоном") + @allure.severity(allure.severity_level.CRITICAL) + def test_expected_id_place_type_name_match(self, place_response: dict[str, Any]) -> None: + allure.attach( + json.dumps(_EXPECTED_RESULT, ensure_ascii=False, indent=2), + name="Ожидаемый объект в results", + attachment_type=AttachmentType.JSON, + ) + with allure.step("Поиск записи с полным совпадением id, place_type, name"): + results = _extract_results(place_response) + match = _find_matching_result(results, _EXPECTED_RESULT) + if match is None: + allure.attach( + json.dumps(results, ensure_ascii=False, indent=2), + name="Фактический results (для отладки)", + attachment_type=AttachmentType.JSON, + ) + assert match is not None, ( + f"Ни одна запись в results не совпадает с ожидаемым набором {_EXPECTED_RESULT}. " + f"Получено: {json.dumps(results, ensure_ascii=False, indent=2)[:2000]}" + ) + allure.attach( + json.dumps(match, ensure_ascii=False, indent=2), + name="Найденная запись", + attachment_type=AttachmentType.JSON, + ) + assert match["id"] == _EXPECTED_RESULT["id"] + assert match["place_type"] == _EXPECTED_RESULT["place_type"] + assert match["name"] == _EXPECTED_RESULT["name"] + + +def _extract_results(payload: dict[str, Any]) -> list[dict[str, Any]]: + assert "data" in payload, f'В ответе нет ключа "data": {json.dumps(payload, ensure_ascii=False)[:500]}' + data = payload["data"] + assert isinstance(data, dict), '"data" должен быть объектом' + + assert "place" in data, f'В data нет "place": keys={list(data.keys())}' + place = data["place"] + assert place is not None, '"place" не должен быть null' + assert isinstance(place, dict), '"place" должен быть объектом' + + assert "results" in place, f'В place нет "results": keys={list(place.keys())}' + results = place["results"] + assert isinstance(results, list), '"results" должен быть массивом' + + for i, item in enumerate(results): + assert isinstance(item, dict), f'results[{i}] должен быть объектом, получено {type(item)}' + return results + + +def _find_matching_result(results: list[dict[str, Any]], expected: dict[str, str]) -> dict[str, Any] | None: + for item in results: + if all(item.get(k) == v for k, v in expected.items()): + return item + return None diff --git a/worklib/QueryData.py b/worklib/QueryData.py new file mode 100644 index 0000000..ff14c0c --- /dev/null +++ b/worklib/QueryData.py @@ -0,0 +1,58 @@ +import re +from typing import Optional + + + +class QuaryData: + def __init__(self, tag): + self.tag = tag + self.username: Optional[str] = None + + + +def kvs_query_data(): + return { + "query": """ +query ($id: String!){ + place(id: $id) { + results { + + members { + id + parent_id + user { + username + } + } + } + } +} +""".strip(), + } + +def kvs_query_data_place_id(): + return { + "variables": { + "id": "6915dc03462d5aea0adc8cbd" + } + } +def query_data(): + return { + "query": """ +query ($id: String!) { + place(id: $id) { + results { + id + name + place_type + } + } +} +""".strip() + } +def query_data_place_id_variables(): + return { + "variables": { + "id": "682b071a163ac2a0995355be" + } + } \ No newline at end of file diff --git a/worklib/__init__.py b/worklib/__init__.py new file mode 100644 index 0000000..f1c4b9c --- /dev/null +++ b/worklib/__init__.py @@ -0,0 +1,6 @@ +__all__ = [ + "admin_data", + "auth_as_employer", + "find_place_data", +] + diff --git a/worklib/admin_data.py b/worklib/admin_data.py new file mode 100644 index 0000000..673d813 --- /dev/null +++ b/worklib/admin_data.py @@ -0,0 +1,76 @@ +import re +from typing import Optional +from worklib.auth_as_employer import get_access_token +import os +class NoTokenException(Exception): + def __init__(self, tag): + super().__init__(f"User {tag} does not have access token") + + +class AdminData: + def __init__(self, tag): + self.tag = tag + self.username: Optional[str] = None + self.access_token: Optional[str] = None + self.refresh_token: Optional[str] = None + self.profile_id: Optional[str] = None + self.phone_number: Optional[str] = None + self.account_id: Optional[str] = None + + +class UserManager: + def __init__(self): + self._users = {} + self.last_user_tag = 0 + + def get_or_create_user(self, user_tag) -> AdminData: + user_id = self.str_to_user_tag(user_tag) + if user_id not in self._users: + self._users[user_id] = AdminData(user_id) + return self._users[user_id] + + def str_to_user_tag(self, user): + if user == "tester": + return 0 + if user == "he": + return self.last_user_tag or 0 + user_id = int(re.sub("[^0-9]", "", str(user)) or 0) + self.last_user_tag = user_id + return user_id + + def get_user_tags(self): + return [user.tag for user in self._users.values()] + + def delete(self, user_tag): + user_id = self.str_to_user_tag(user_tag) + del self._users[user_id] + + def delete_user_by_username(self, username): + for user_tag in list(self._users.keys()): + if self._users[user_tag].username == username: + del self._users[user_tag] + return + + @property + def users(self): + return self._users.copy() + + +user_manager = UserManager() + + +def get_or_create_user(user_tag) -> AdminData: + return user_manager.get_or_create_user(user_tag) + +def get_access_token_from_env(): + # Берём значения из env, чтобы не хардкодить в тестах. + username = os.getenv("AUTH_USERNAME", "+79214400842") + password = os.getenv("AUTH_PASSWORD", "stepan") + grant_type = os.getenv("AUTH_GRANT_TYPE", "password") + + token = get_access_token(username=username, password=password, grant_type=grant_type) + access_token = token + return access_token + + # Храним в простом менеджере, чтобы можно было расширять сценарии. + get_or_create_user("tester").access_token = access_token \ No newline at end of file diff --git a/worklib/auth_as_employer.py b/worklib/auth_as_employer.py new file mode 100644 index 0000000..dc3f08c --- /dev/null +++ b/worklib/auth_as_employer.py @@ -0,0 +1,56 @@ +import json +import os +import urllib.error +import urllib.request +from typing import Optional + + +DEFAULT_USERNAME = "+79214400842" +DEFAULT_PASSWORD = "stepan" +DEFAULT_GRANT_TYPE = "password" + + +def get_access_token( + auth_url: Optional[str] = None, + *, + username: str = DEFAULT_USERNAME, + password: str = DEFAULT_PASSWORD, + grant_type: str = DEFAULT_GRANT_TYPE, + timeout_s: int = 30, +) -> str: + """ + POST авторизация, возвращает access_token. + + Можно переопределить AUTH_URL через env. + """ + auth_url = auth_url or os.getenv("AUTH_URL") or "https://auth.dev.dipal.ru/api/v1/auth/login" + if not auth_url: + raise ValueError("Не задан AUTH_URL (передайте auth_url или установите env AUTH_URL)") + + payload = {"username": username, "password": password, "grant_type": grant_type} + req = urllib.request.Request( + auth_url, + data=json.dumps(payload).encode("utf-8"), + headers={"content-type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=timeout_s) as resp: + raw = resp.read().decode("utf-8") + except urllib.error.HTTPError as e: + body = e.read().decode("utf-8", errors="replace") + raise RuntimeError(f"Auth HTTP {e.code}: {body}") from e + + data = json.loads(raw) if raw else {} + if "data" in data and isinstance(data["data"], dict): + token = data["data"].get("access_token") + else: + token = data.get("access_token") + if not token: + raise ValueError(f"В ответе нет access_token. Ключи ответа: {list(data.keys())}") + return token + + +if __name__ == "__main__": + print(get_access_token()) + diff --git a/worklib/findplaceinfo/find_place_data.py b/worklib/findplaceinfo/find_place_data.py new file mode 100644 index 0000000..c8ada99 --- /dev/null +++ b/worklib/findplaceinfo/find_place_data.py @@ -0,0 +1,62 @@ +import json +import os +import urllib.error +import urllib.request +from typing import Any, Dict, Optional +from worklib.QueryData import kvs_query_data, kvs_query_data_place_id +from worklib.auth_as_employer import get_access_token + + +DEFAULT_COMPANY_ID = "65437401ae3af6f8ffcdbaf8" + + + +def build_headers(access_token: str, *, company_id: str = DEFAULT_COMPANY_ID) -> Dict[str, str]: + return { + "authorization": f"bearer {access_token}", + "x-company-id": company_id, + } + + +def fetch_place_members( + *, + query: str, + variables: dict[str, str], + graphql_url: Optional[str] = None, + + company_id: str = DEFAULT_COMPANY_ID, + access_token: Optional[str] = None, + timeout_s: int = 30, +) -> Dict[str, Any]: + graphql_url = graphql_url or os.getenv("GRAPHQL_URL") or "https://admin.dev.dipal.ru/graphql" + if not graphql_url: + raise ValueError("Не задан GRAPHQL_URL (передайте graphql_url или установите env GRAPHQL_URL)") + + token = access_token or get_access_token() + headers = build_headers(token, company_id=company_id) + + + + payload = {"query": query, "variables": variables} + req = urllib.request.Request( + graphql_url, + data=json.dumps(payload).encode("utf-8"), + headers={"content-type": "application/json", **headers}, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=timeout_s) as resp: + raw = resp.read().decode("utf-8") + except urllib.error.HTTPError as e: + body = e.read().decode("utf-8", errors="replace") + raise RuntimeError(f"GraphQL HTTP {e.code}: {body}") from e + + data = json.loads(raw) if raw else {} + if isinstance(data, dict) and data.get("errors"): + raise RuntimeError(f"GraphQL errors: {data['errors']}") + return data + + +if __name__ == "__main__": + print() +