from __future__ import annotations import json import random import time from dataclasses import dataclass from dataclasses import field from typing import Any, Callable, Optional import allure # pyright: ignore[reportMissingImports] from allure_commons.types import AttachmentType # pyright: ignore[reportMissingImports] from worklib import admin_data from worklib.graphql_client import DEFAULT_COMPANY_ID, execute_graphql from KVSTest.testdata.query_data import add_user_to_place_mutation, kvs_place_members_query def _attach_json(name: str, payload: Any) -> None: allure.attach( json.dumps(payload, ensure_ascii=False, indent=2), name=name, attachment_type=AttachmentType.JSON, ) def _exec_or_fail(*, op_name: str, token: str, query: str, variables: dict[str, Any] | None = None, company_id: str) -> dict[str, Any]: try: return execute_graphql( query=query, variables=variables, company_id=company_id, access_token=token, ) except PermissionError as e: allure.attach(str(e), name=f"Forbidden: {op_name}", attachment_type=AttachmentType.TEXT) raise AssertionError(f"Forbidden на операции: {op_name}") from e except RuntimeError as e: if "Forbidden" in str(e): allure.attach(str(e), name=f"Forbidden (runtime): {op_name}", attachment_type=AttachmentType.TEXT) raise AssertionError(f"Forbidden на операции: {op_name}") from e raise @dataclass class KVSTestData: """ Хранилище/фабрика тестовых данных для KVS GraphQL. - Создаёт сущности (place/user) по мере необходимости - Кеширует id в полях - Регистрирует cleanup в behave context (если передан) """ company_id: str = DEFAULT_COMPANY_ID parent_place_id: str = "6915dc03462d5aea0adc8cbd" default_user_first_name: str = "kvstest1" default_user_last_name: str = "kvstest2" access_token: Optional[str] = None place_id: Optional[str] = None account_id: Optional[str] = None username: Optional[str] = None _cleanup_fns: Optional[list[Callable[[], None]]] = None _last_add_user_to_place_response: dict[str, Any] | None = None @classmethod def from_behave_context(cls, context: Any, *, company_id: str = DEFAULT_COMPANY_ID) -> "KVSTestData": td: KVSTestData | None = getattr(context, "kvs_test_data", None) if isinstance(td, cls): if not td.access_token and getattr(context, "access_token", None): td.access_token = context.access_token if not td._cleanup_fns: td._cleanup_fns = getattr(context, "_cleanup_fns", None) return td td = cls(company_id=company_id) td.access_token = getattr(context, "access_token", None) or None td._cleanup_fns = getattr(context, "_cleanup_fns", None) setattr(context, "kvs_test_data", td) return td def _register_cleanup(self, fn: Callable[[], None]) -> None: if self._cleanup_fns is not None: self._cleanup_fns.append(fn) def ensure_token(self) -> str: if self.access_token: return self.access_token token = admin_data.get_or_create_user("tester").access_token assert token, "Нет access_token (ни в context, ни в admin_data.get_or_create_user('tester'))." self.access_token = token return token def ensure_place(self) -> str: if self.place_id: return self.place_id token = self.ensure_token() mutation = """ mutation ($place_type: PlaceType!, $names: [String!]!, $parent_id: String, $address_id: String) { createPlaceMultiple( dto: {place_type: $place_type, names: $names, parent_id: $parent_id, address_id: $address_id} ) { id __typename } } """.strip() suffix = str(int(time.time())) variables = {"names": [f"kvs-test-{suffix}"], "parent_id": self.parent_place_id, "place_type": "flat"} with allure.step("GraphQL: createPlaceMultiple (KVS)"): resp = _exec_or_fail(op_name="createPlaceMultiple(mutation)", token=token, query=mutation, variables=variables, company_id=self.company_id) _attach_json("createPlaceMultiple response", resp) created = resp.get("data", {}).get("createPlaceMultiple") if isinstance(created, list): assert created, f"createPlaceMultiple вернул пустой список. Ответ: {resp}" created0 = created[0] else: created0 = created assert isinstance(created0, dict), f"createPlaceMultiple вернул не объект: {created0!r}. Ответ: {resp}" place_id = created0.get("id") assert place_id, f"createPlaceMultiple не вернул id. Ответ: {resp}" self.place_id = place_id def _cleanup_delete_place() -> None: delete_mutation = """mutation deleteplace($id: String!) { deletePlace(id: $id) }""".strip() _exec_or_fail(op_name="deletePlace(mutation)", token=token, query=delete_mutation, variables={"id": place_id}, company_id=self.company_id) self._register_cleanup(_cleanup_delete_place) return place_id def create_user(self) -> str: if self.account_id: return self.account_id token = self.ensure_token() username = f"+7999{random.randint(1000000, 9999999)}" mutation = """ mutation createUser($input: CreateAccountDTO!) { createUser(dto: $input) } """.strip() variables = { "input": { "username": username, "first_name": self.default_user_first_name, "last_name": self.default_user_last_name, "is_demo": True, "password": "", } } with allure.step("GraphQL: createUser (KVS)"): resp = _exec_or_fail(op_name="createUser(mutation)", token=token, query=mutation, variables=variables, company_id=self.company_id) _attach_json("createUser response", resp) created = resp.get("data", {}).get("createUser") if isinstance(created, dict): account_id = created.get("id") returned_username = created.get("username") else: account_id = created returned_username = None assert isinstance(account_id, str) and account_id, f"createUser не вернул id. Ответ: {resp}" self.account_id = account_id self.username = returned_username or username def _cleanup_delete_user() -> None: delete_mutation = """ mutation deleteUser($account_id: String!) { deleteUser(account_id: $account_id) } """.strip() _exec_or_fail( op_name="deleteUser(mutation)", token=token, query=delete_mutation, variables={"account_id": account_id}, company_id=self.company_id, ) self._register_cleanup(_cleanup_delete_user) return account_id def add_user_to_place(self, *, account_id: str | None = None, place_id: str | None = None) -> dict[str, Any]: token = self.ensure_token() aid = account_id or self.create_user() pid = place_id or self.ensure_place() mutation = add_user_to_place_mutation() variables = {"account_id": aid, "place_id": pid} with allure.step("GraphQL: addUserToPlace (KVS)"): resp = _exec_or_fail(op_name="addUserToPlace(mutation)", token=token, query=mutation, variables=variables, company_id=self.company_id) _attach_json("addUserToPlace response", resp) self._last_add_user_to_place_response = resp return resp def query_place_members(self, *, place_id: str | None = None) -> dict[str, Any]: token = self.ensure_token() pid = place_id or self.ensure_place() query = kvs_place_members_query() variables = {"id": pid} with allure.step("GraphQL: place members (KVS)"): resp = _exec_or_fail(op_name="place(query)", token=token, query=query, variables=variables, company_id=self.company_id) _attach_json("place members response", resp) return resp