196 lines
7.5 KiB
Python
196 lines
7.5 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import time
|
||
import uuid
|
||
import urllib.error
|
||
import urllib.request
|
||
from typing import Any, Optional
|
||
|
||
from worklib.auth_as_employer import get_access_token
|
||
|
||
|
||
DEFAULT_COMPANY_ID = "65437401ae3af6f8ffcdbaf8"
|
||
DEFAULT_GRAPHQL_URL = "https://admin.dev.dipal.ru/graphql"
|
||
|
||
_PASSREQUESTS_MOCK_STATE: dict[str, Any] = {}
|
||
|
||
|
||
def reset_passrequests_mock_state() -> None:
|
||
_PASSREQUESTS_MOCK_STATE.clear()
|
||
|
||
|
||
def _mock_passrequests_execute_graphql(*, query: str, variables: Optional[dict[str, Any]], company_id: str, access_token: str) -> dict[str, Any]:
|
||
state = _PASSREQUESTS_MOCK_STATE.setdefault(
|
||
company_id,
|
||
{"places": {}, "users": {}, "passes": {}, "pass_requests_by_pass_id": {}},
|
||
)
|
||
|
||
q = query or ""
|
||
vars_ = variables or {}
|
||
|
||
def _new_id(prefix: str) -> str:
|
||
return f"{prefix}_{uuid.uuid4().hex[:12]}"
|
||
|
||
# createPlaceMultiple -> [{id}]
|
||
if "createPlaceMultiple" in q:
|
||
place_id = _new_id("place")
|
||
state["places"][place_id] = {"id": place_id}
|
||
return {"data": {"createPlaceMultiple": [{"id": place_id, "__typename": "Place"}]}}
|
||
|
||
# createUser -> id
|
||
if "createUser" in q:
|
||
account_id = _new_id("user")
|
||
state["users"][account_id] = {"id": account_id}
|
||
return {"data": {"createUser": account_id}}
|
||
|
||
# createService -> {id,title,type}
|
||
if "createService" in q:
|
||
service_id = _new_id("service")
|
||
title = (vars_.get("title") or "mock-service") if isinstance(vars_, dict) else "mock-service"
|
||
stype = (vars_.get("type") or "access") if isinstance(vars_, dict) else "access"
|
||
return {"data": {"createService": {"id": service_id, "title": title, "type": stype}}}
|
||
|
||
# addPlaceToService/removePlaceFromService -> {id}
|
||
if "addPlaceToService" in q or "removePlaceFromService" in q:
|
||
return {"data": {"addPlaceToService": {"id": "ok"}, "removePlaceFromService": {"id": "ok"}}}
|
||
|
||
# addUserToPlace(dto: $input) -> {place_id, member_id}
|
||
if "addUserToPlace" in q:
|
||
inp = vars_.get("input") or {}
|
||
place_id = inp.get("place_id")
|
||
account_id = inp.get("account_id")
|
||
if not place_id or not account_id:
|
||
return {"errors": [{"message": "Bad input", "status": 400}]}
|
||
member_id = _new_id("member")
|
||
return {"data": {"addUserToPlace": {"place_id": place_id, "member_id": member_id}}}
|
||
|
||
# createPass -> {id}
|
||
if "createPass" in q:
|
||
pass_id = _new_id("pass")
|
||
state["passes"][pass_id] = {"id": pass_id}
|
||
pr_id = _new_id("passreq")
|
||
state["pass_requests_by_pass_id"][pass_id] = {
|
||
"id": pr_id,
|
||
"status": "pending",
|
||
"pass_id": pass_id,
|
||
"confirmer_ids": [],
|
||
"created_at": str(int(time.time())),
|
||
"updated_at": str(int(time.time())),
|
||
}
|
||
return {"data": {"createPass": {"id": pass_id}}}
|
||
|
||
# passRequests(filters:{pass_id}) -> results[{...}]
|
||
if "passRequests" in q:
|
||
pass_id = vars_.get("pass_id")
|
||
pr = state["pass_requests_by_pass_id"].get(pass_id)
|
||
if not pr:
|
||
return {"data": {"passRequests": {"results": []}}}
|
||
return {
|
||
"data": {
|
||
"passRequests": {
|
||
"results": [
|
||
{
|
||
"id": pr["id"],
|
||
"status": pr["status"],
|
||
"pass_id": pr["pass_id"],
|
||
"place_id": "mock_place",
|
||
"created_at": pr["created_at"],
|
||
"updated_at": pr["updated_at"],
|
||
"place": {"id": "mock_place"},
|
||
"confirmer_ids": pr["confirmer_ids"],
|
||
}
|
||
]
|
||
}
|
||
}
|
||
}
|
||
|
||
# approve/confirm -> if already rejected then no-op; else require 2 confirmations to become active
|
||
if "approvePassRequest" in q or "confirmPassRequest" in q:
|
||
pr_id = vars_.get("pass_request_id") or vars_.get("id")
|
||
pr = next((v for v in state["pass_requests_by_pass_id"].values() if v["id"] == pr_id), None)
|
||
if pr and pr["status"] != "rejected":
|
||
conf_id = access_token or "mock_employee"
|
||
if conf_id not in pr["confirmer_ids"]:
|
||
pr["confirmer_ids"].append(conf_id)
|
||
pr["status"] = "active" if len(pr["confirmer_ids"]) >= 2 else "pending"
|
||
return {"data": {"approvePassRequest": True}}
|
||
|
||
# rejectPassRequest(pass_request_id) -> sets rejected permanently
|
||
if "rejectPassRequest" in q:
|
||
pr_id = vars_.get("pass_request_id") or vars_.get("id")
|
||
pr = next((v for v in state["pass_requests_by_pass_id"].values() if v["id"] == pr_id), None)
|
||
if pr:
|
||
pr["status"] = "rejected"
|
||
return {"data": {"rejectPassRequest": True}}
|
||
|
||
# deleteUser/deletePlace/deletePass etc -> True
|
||
if "deleteUser" in q or "deletePlace" in q or "deletePass" in q:
|
||
return {"data": {"ok": True}}
|
||
|
||
return {"data": {}}
|
||
|
||
|
||
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 execute_graphql(
|
||
*,
|
||
query: str,
|
||
variables: Optional[dict[str, Any]] = None,
|
||
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 DEFAULT_GRAPHQL_URL
|
||
if not graphql_url:
|
||
raise ValueError("Не задан GRAPHQL_URL (передайте graphql_url или установите env GRAPHQL_URL)")
|
||
|
||
token = access_token or get_access_token()
|
||
|
||
if os.getenv("PASSREQUESTS_MOCKS") in {"1", "true", "True"}:
|
||
data = _mock_passrequests_execute_graphql(query=query, variables=variables, company_id=company_id, access_token=token)
|
||
if isinstance(data, dict) and data.get("errors"):
|
||
raise RuntimeError(f"GraphQL errors: {data['errors']}")
|
||
return data
|
||
|
||
headers = build_headers(token, company_id=company_id)
|
||
|
||
payload: dict[str, Any] = {"query": query}
|
||
if variables is not None:
|
||
payload["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"):
|
||
errors = data["errors"]
|
||
# На стенде часть операций возвращает Forbidden как GraphQL error (не HTTP 403).
|
||
if isinstance(errors, list) and any(getattr(err, "get", lambda *_: None)("status") == 403 for err in errors):
|
||
raise PermissionError(
|
||
"Forbidden (403) для GraphQL операции. "
|
||
"Проверьте креды/права. Можно задать env: AUTH_USERNAME/AUTH_PASSWORD/AUTH_GRANT_TYPE."
|
||
)
|
||
raise RuntimeError(f"GraphQL errors: {errors}")
|
||
if not isinstance(data, dict):
|
||
raise RuntimeError(f"GraphQL response is not an object: {data!r}")
|
||
return data
|
||
|