graphql-engine/server/tests-py/auth_webhook_server.py
Samir Talwar 8cb2738cbe server/tests-py: Declaratively state the HGE environment variables.
This has two purposes:

* When running the Python integration tests against a running HGE instance, with `--hge-url`, it will check the environment variables available and actively skip the test if they aren't set. This replaces the previous ad-hoc skip behavior.
* More interestingly, when running against a binary with `--hge-bin`, the environment variables are passed through, which means different tests can run with different environment variables.

  On top of this, the various services we use for testing now also provide their own environment variables, rather than expecting a test script to do it.

In order to make this work, I also had to invert the dependency between various services and `hge_ctx`. I extracted a `pg_version` fixture to provide the PostgreSQL version, and now pass the `hge_url` and `hge_key` explicitly to `ActionsWebhookServer`.

PR-URL: https://github.com/hasura/graphql-engine-mono/pull/6028
GitOrigin-RevId: 16d866741dba5887da1adf4e1ade8182ccc9d344
2022-09-28 09:21:02 +00:00

61 lines
1.8 KiB
Python

"""
Sample auth webhook to receive a cookie and respond
"""
from http import HTTPStatus
import http.server
from webserver import MkHandlers, RequestHandler, Response
class CookieAuth(RequestHandler):
def get(self, request):
print('auth GET request')
headers = {k.lower(): v for k, v in request.headers.items()}
print(headers)
cookieHdrs = []
if 'cookie' in headers and headers['cookie']:
res = {'x-hasura-role': 'admin'}
for k, v in headers.items():
if 'response-set-cookie' in k:
hdr = ('Set-Cookie', v)
cookieHdrs.append(hdr)
print('auth response: OK')
return Response(HTTPStatus.OK, res, cookieHdrs)
print('auth response: Unauthorized')
return Response(HTTPStatus.UNAUTHORIZED)
def post(self, request):
print('auth POST request')
headers = {k.lower(): v for k, v in request.json['headers'].items()}
cookieHdrs = []
if 'cookie' in headers and headers['cookie']:
res = {'x-hasura-role': 'admin'}
for k, v in headers.items():
if 'response-set-cookie' in k:
hdr = ('Set-Cookie', v)
cookieHdrs.append(hdr)
print('auth response: OK')
return Response(HTTPStatus.OK, res, headers)
print('auth response: Unauthorized')
return Response(HTTPStatus.UNAUTHORIZED)
handlers = MkHandlers({
'/auth': CookieAuth,
})
def create_server(host='127.0.0.1', port=9876):
return http.server.HTTPServer((host, port), handlers)
def stop_server(server):
server.shutdown()
server.server_close()
if __name__ == '__main__':
s = create_server(host='0.0.0.0')
s.serve_forever()