mirror of
https://github.com/hasura/graphql-engine.git
synced 2024-12-17 12:31:52 +03:00
bff1309344
Starting it and stopping it for the various tests that actually use it. There are only a few. This also removes some dead code and fixes warnings in _test_webhook_request_context.py_. PR-URL: https://github.com/hasura/graphql-engine-mono/pull/5846 GitOrigin-RevId: 7760467f9de7b1f9718e7482275c298eeaa3ad3a
131 lines
4.0 KiB
Python
131 lines
4.0 KiB
Python
import http
|
|
import http.server
|
|
import json
|
|
import pytest
|
|
import queue
|
|
import threading
|
|
|
|
from context import PytestConf
|
|
|
|
if not PytestConf.config.getoption("--test-webhook-request-context"):
|
|
pytest.skip("--test-webhook-https-request-context flag is missing, skipping tests", allow_module_level=True)
|
|
|
|
|
|
class QueryEchoWebhookServer(http.server.HTTPServer):
|
|
def __init__(self, server_address):
|
|
# TODO why maxsize=1
|
|
self.resp_queue = queue.Queue(maxsize=1)
|
|
super().__init__(server_address, QueryEchoWebhookHandler)
|
|
|
|
def get_event(self, timeout):
|
|
return self.resp_queue.get(timeout=timeout)
|
|
|
|
|
|
class QueryEchoWebhookHandler(http.server.BaseHTTPRequestHandler):
|
|
server: QueryEchoWebhookServer
|
|
|
|
def do_GET(self):
|
|
self.log_message("get")
|
|
self.send_response(http.HTTPStatus.OK)
|
|
self.end_headers()
|
|
|
|
def do_POST(self):
|
|
self.log_message("post")
|
|
content_len = self.headers.get("Content-Length")
|
|
req_body = self.rfile.read(int(content_len)).decode("utf-8")
|
|
req_json = json.loads(req_body)
|
|
print(json.dumps(req_json))
|
|
user_id_header = req_json["headers"]["auth-user-id"]
|
|
user_vars = {
|
|
"x-hasura-role":"user",
|
|
"x-hasura-user-id": user_id_header
|
|
}
|
|
self.send_response(http.HTTPStatus.OK)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.server.resp_queue.put({
|
|
"request": req_json["request"],
|
|
"headers": user_vars,
|
|
})
|
|
self.wfile.write(json.dumps(user_vars).encode('utf-8'))
|
|
|
|
|
|
@pytest.fixture(scope="class")
|
|
def query_echo_webhook():
|
|
# TODO(swann): is this the right port?
|
|
webhook_httpd = QueryEchoWebhookServer(server_address=("localhost", 5594))
|
|
web_server = threading.Thread(target=webhook_httpd.serve_forever)
|
|
web_server.start()
|
|
yield webhook_httpd
|
|
webhook_httpd.shutdown()
|
|
webhook_httpd.server_close()
|
|
web_server.join()
|
|
|
|
|
|
@pytest.mark.usefixtures("per_method_tests_db_state")
|
|
class TestWebhookRequestContext(object):
|
|
@classmethod
|
|
def dir(cls):
|
|
return "queries/webhooks/request_context"
|
|
|
|
def test_query(self, hge_ctx, query_echo_webhook):
|
|
query = """
|
|
query allUsers {
|
|
users {
|
|
id
|
|
name
|
|
}
|
|
}
|
|
"""
|
|
query_obj = {
|
|
"query": query,
|
|
"operationName": "allUsers"
|
|
}
|
|
headers = dict()
|
|
headers['auth-user-id'] = '1'
|
|
code, resp, _ = hge_ctx.anyq('/v1/graphql', query_obj, headers)
|
|
assert code == 200, resp
|
|
|
|
ev_full = query_echo_webhook.get_event(3)
|
|
assert ev_full['request'] == query_obj
|
|
|
|
def test_query_invalid(self, hge_ctx, query_echo_webhook):
|
|
"""
|
|
Even when an invalid query is sent, the webhook should still resolve
|
|
the user id header correctly
|
|
"""
|
|
query_obj = "invalid-query"
|
|
user_id_header = '1'
|
|
headers = dict()
|
|
headers['auth-user-id'] = user_id_header
|
|
code, resp, _ = hge_ctx.anyq('/v1/graphql', query_obj, headers)
|
|
assert code == 200, resp
|
|
|
|
ev_full = query_echo_webhook.get_event(3)
|
|
assert ev_full['headers']['x-hasura-user-id'] == user_id_header
|
|
|
|
def test_mutation_with_vars(self, hge_ctx, query_echo_webhook):
|
|
query = """
|
|
mutation insert_single_user($id: Int!, $name: String!) {
|
|
insert_users_one(
|
|
object: {
|
|
id: $id,
|
|
name: $name
|
|
}
|
|
) {
|
|
id
|
|
name
|
|
}
|
|
}
|
|
"""
|
|
variables = {"id": 4, "name": "danish"}
|
|
query_obj = {"query": query, "variables": variables, "operationName": "insert_single_user"}
|
|
headers = dict()
|
|
headers['auth-user-id'] = '4'
|
|
code, resp, _ = hge_ctx.anyq('/v1/graphql', query_obj, headers)
|
|
assert code == 200, resp
|
|
|
|
ev_full = query_echo_webhook.get_event(3)
|
|
exp_result = {"request": query_obj}
|
|
assert ev_full['request'] == query_obj
|