203 lines
7.8 KiB
Python
203 lines
7.8 KiB
Python
"""
|
||
In-process GraphQL mocks for Subscribe_to_bundle behave tests.
|
||
|
||
Включается env SUBSCRIBE_BUNDLE_MOCKS=1 (часто вместе с USE_WIREMOCK=1 в environment пакета).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from typing import Any, Optional
|
||
|
||
_SUBSCRIBE_BUNDLE_STATE: dict[str, Any] = {}
|
||
|
||
|
||
def reset_subscribe_bundle_mock_state() -> None:
|
||
_SUBSCRIBE_BUNDLE_STATE.clear()
|
||
|
||
|
||
def _new_id(prefix: str) -> str:
|
||
return f"{prefix}_{uuid.uuid4().hex[:12]}"
|
||
|
||
|
||
def execute_subscribe_bundle_mock_graphql(
|
||
*,
|
||
query: str,
|
||
variables: Optional[dict[str, Any]],
|
||
company_id: str,
|
||
access_token: str,
|
||
) -> dict[str, Any]:
|
||
q = query or ""
|
||
vars_ = variables or {}
|
||
st = _SUBSCRIBE_BUNDLE_STATE.setdefault(
|
||
company_id,
|
||
{
|
||
"places": [],
|
||
"services": [],
|
||
"plans": [],
|
||
"subscriptions": [],
|
||
"subscriber_id": None,
|
||
"links": [], # {service_id, place_id}
|
||
},
|
||
)
|
||
|
||
if "createPlaceMultiple" in q:
|
||
names = vars_.get("names")
|
||
if not isinstance(names, list):
|
||
dto = vars_.get("dto")
|
||
if isinstance(dto, dict):
|
||
names = dto.get("names")
|
||
if not isinstance(names, list):
|
||
names = ["p1"]
|
||
created: list[dict[str, Any]] = []
|
||
for _n in names:
|
||
pid = _new_id("place")
|
||
st["places"].append(pid)
|
||
created.append({"id": pid, "__typename": "Place"})
|
||
return {"data": {"createPlaceMultiple": created}}
|
||
|
||
if "createService" in q:
|
||
sid = _new_id("svc")
|
||
title = (vars_.get("title") or "svc") if isinstance(vars_, dict) else "svc"
|
||
stype = (vars_.get("type") or "access") if isinstance(vars_, dict) else "access"
|
||
st["services"].append({"id": sid, "title": title, "type": stype})
|
||
return {"data": {"createService": {"id": sid, "title": title, "type": stype}}}
|
||
|
||
if "addPlaceToService" in q:
|
||
dto = vars_.get("dto") or {}
|
||
if isinstance(dto, dict):
|
||
st["links"].append({"service_id": dto.get("service_id"), "place_id": dto.get("place_id")})
|
||
return {"data": {"addPlaceToService": {"id": "ok"}}}
|
||
|
||
if "createPlan" in q:
|
||
inp = vars_.get("input") or {}
|
||
services = inp.get("services") if isinstance(inp, dict) else None
|
||
if not isinstance(services, list):
|
||
services = []
|
||
plan_id = _new_id("plan")
|
||
place_ids = inp.get("place_ids") if isinstance(inp, dict) else []
|
||
if not isinstance(place_ids, list):
|
||
place_ids = []
|
||
st["plans"].append({"id": plan_id, "service_ids": services, "place_ids": place_ids, "title": inp.get("title")})
|
||
return {
|
||
"data": {
|
||
"createPlan": {
|
||
"id": plan_id,
|
||
"service_ids": services,
|
||
"bundle_ids": [],
|
||
"place_id": place_ids[0] if place_ids else None,
|
||
"place_ids": place_ids,
|
||
"price": inp.get("price") if isinstance(inp, dict) else 0,
|
||
"title": inp.get("title") if isinstance(inp, dict) else "plan",
|
||
"discount": 0,
|
||
"payment_interval": 1,
|
||
"price_without_discount": inp.get("price") if isinstance(inp, dict) else 0,
|
||
}
|
||
}
|
||
}
|
||
|
||
if "addUserToPlace" in q:
|
||
inp = vars_.get("input") or vars_.get("dto") or {}
|
||
aid = inp.get("account_id") if isinstance(inp, dict) else None
|
||
if isinstance(aid, str) and aid:
|
||
st["subscriber_id"] = aid
|
||
return {"data": {"addUserToPlace": {"place_id": "mock", "member_id": _new_id("member")}}}
|
||
|
||
if "createUser" in q:
|
||
uid = _new_id("user")
|
||
return {"data": {"createUser": uid}}
|
||
|
||
if "createSubscription" in q:
|
||
dto = vars_.get("dto") or {}
|
||
plan_id = dto.get("plan_id") if isinstance(dto, dict) else None
|
||
plan = next((p for p in st["plans"] if p.get("id") == plan_id), None) or {}
|
||
svc_ids = plan.get("service_ids") or []
|
||
services = [{"id": x, "title": f"svc-{x}"} for x in svc_ids if isinstance(x, str)]
|
||
sub_id = _new_id("sub")
|
||
st["subscriptions"].append(
|
||
{
|
||
"id": sub_id,
|
||
"place_id": dto.get("place_id") if isinstance(dto, dict) else None,
|
||
"subscriber_id": dto.get("subscriber_id") if isinstance(dto, dict) else None,
|
||
"service_ids": svc_ids,
|
||
}
|
||
)
|
||
return {
|
||
"data": {
|
||
"createSubscription": {
|
||
"id": sub_id,
|
||
"services": services,
|
||
"user": {"id": dto.get("subscriber_id"), "data": {"first_name": "t", "last_name": "t"}},
|
||
"plan": {"id": plan_id, "title": "mock-plan"},
|
||
"place_id": dto.get("place_id") if isinstance(dto, dict) else None,
|
||
}
|
||
}
|
||
}
|
||
|
||
if "bundleScope" in q or ("members(" in q and "place(" in q and "services" in q):
|
||
place_id = vars_.get("place_id") or vars_.get("pid")
|
||
pid = vars_.get("pid") or place_id
|
||
subscriber = st.get("subscriber_id")
|
||
svc_at_place: list[dict[str, str]] = []
|
||
for link in st.get("links", []):
|
||
if isinstance(link, dict) and link.get("place_id") == pid:
|
||
sid = link.get("service_id")
|
||
if isinstance(sid, str):
|
||
meta = next((s for s in st["services"] if s.get("id") == sid), {})
|
||
svc_at_place.append({"id": sid, "title": str(meta.get("title") or sid)})
|
||
members_results: list[dict[str, Any]] = []
|
||
if (
|
||
isinstance(subscriber, str)
|
||
and isinstance(place_id, str)
|
||
and st.get("places")
|
||
and len(st["places"]) >= 1
|
||
and place_id == st["places"][0]
|
||
):
|
||
members_results = [{"id": _new_id("mem"), "user": {"id": subscriber}}]
|
||
place_results: list[dict[str, Any]] = []
|
||
if isinstance(pid, str):
|
||
place_results.append({"id": pid, "services": svc_at_place})
|
||
return {
|
||
"data": {
|
||
"members": {"results": members_results},
|
||
"place": {"results": place_results},
|
||
}
|
||
}
|
||
|
||
if "members(" in q and "members(filters" in q:
|
||
place_id = vars_.get("place_id")
|
||
subscriber = st.get("subscriber_id")
|
||
results: list[dict[str, Any]] = []
|
||
if (
|
||
isinstance(place_id, str)
|
||
and isinstance(subscriber, str)
|
||
and st.get("places")
|
||
and len(st["places"]) >= 1
|
||
and place_id == st["places"][0]
|
||
):
|
||
results = [{"id": _new_id("mem"), "user": {"id": subscriber}}]
|
||
return {"data": {"members": {"results": results}}}
|
||
|
||
if "place(" in q and "place(id" in q.replace(" ", ""):
|
||
pid = vars_.get("id") or vars_.get("place_id")
|
||
svc_at_place = []
|
||
for link in st.get("links", []):
|
||
if isinstance(link, dict) and link.get("place_id") == pid:
|
||
sid = link.get("service_id")
|
||
if isinstance(sid, str):
|
||
meta = next((s for s in st["services"] if s.get("id") == sid), {})
|
||
svc_at_place.append({"id": sid, "title": meta.get("title") or sid})
|
||
return {"data": {"place": {"results": [{"id": pid, "services": svc_at_place}]}}}
|
||
|
||
if "deleteSubscription" in q:
|
||
return {"data": {"deleteSubscription": True}}
|
||
if "deletePlan" in q:
|
||
return {"data": {"deletePlan": True}}
|
||
if "deleteService" in q:
|
||
return {"data": {"deleteService": True}}
|
||
if "deletePlace" in q:
|
||
return {"data": {"deletePlace": True}}
|
||
if "deleteUser" in q:
|
||
return {"data": {"deleteUser": True}}
|
||
|
||
return {"data": {}, "errors": [{"message": f"Subscribe bundle mock: unsupported query snippet: {q[:120]!r}"}]}
|