51 lines
2.1 KiB
Python
51 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import traceback
|
|
import urllib.request
|
|
from typing import Any, Callable
|
|
|
|
import allure # pyright: ignore[reportMissingImports]
|
|
from allure_commons.types import AttachmentType # pyright: ignore[reportMissingImports]
|
|
|
|
from worklib.subscribe_bundle_graphql_mock import reset_subscribe_bundle_mock_state
|
|
|
|
|
|
def before_scenario(context: Any, scenario: Any) -> None: # noqa: ARG001
|
|
context._cleanup_fns = [] # type: ignore[attr-defined]
|
|
reset_subscribe_bundle_mock_state()
|
|
|
|
# USE_WIREMOCK=1 — как в Pass_request: локальный GraphQL (или in-process моки).
|
|
# SUBSCRIBE_BUNDLE_MOCKS=1 — ответы из worklib.subscribe_bundle_graphql_mock (без полной схемы WireMock).
|
|
if os.getenv("USE_WIREMOCK") in {"1", "true", "True"}:
|
|
os.environ.setdefault("SUBSCRIBE_BUNDLE_MOCKS", "1")
|
|
os.environ.setdefault("GRAPHQL_URL", "http://localhost:8080/graphql")
|
|
os.environ.setdefault("AUTH_URL", "http://localhost:8080/api/v1/auth/login")
|
|
try:
|
|
for url in (
|
|
"http://localhost:8080/__admin/reset",
|
|
"http://localhost:8080/__admin/scenarios/reset",
|
|
"http://localhost:8080/__admin/requests/reset",
|
|
):
|
|
req = urllib.request.Request(url, data=b"", method="POST")
|
|
urllib.request.urlopen(req, timeout=2).read()
|
|
except Exception as e: # noqa: BLE001
|
|
allure.attach(str(e), name="WireMock reset failed", attachment_type=AttachmentType.TEXT)
|
|
|
|
context.graphql_url = os.getenv("GRAPHQL_URL")
|
|
|
|
|
|
def after_scenario(context: Any, scenario: Any) -> None: # noqa: ARG001
|
|
cleanup_fns: list[Callable[[], None]] = getattr(context, "_cleanup_fns", [])
|
|
while cleanup_fns:
|
|
fn = cleanup_fns.pop()
|
|
try:
|
|
with allure.step(f"Cleanup: {getattr(fn, '__name__', 'cleanup')}"):
|
|
fn()
|
|
except Exception:
|
|
allure.attach(
|
|
traceback.format_exc(),
|
|
name="Cleanup error",
|
|
attachment_type=AttachmentType.TEXT,
|
|
)
|