Skip to content

Instantly share code, notes, and snippets.

@Noordsestern
Last active December 22, 2021 15:57
Show Gist options
  • Save Noordsestern/08949543af6ef2e97e428d9e8a8d4c51 to your computer and use it in GitHub Desktop.
Save Noordsestern/08949543af6ef2e97e428d9e8a8d4c51 to your computer and use it in GitHub Desktop.
Small web service for mocking webhooks. Logs all incoming requests. Useful for testing systems that send out notifications. Make system under test send its notifications to this web service and validate the messages on its GET endpoints.
from fastapi import FastAPI, Response, Request
import json
from json import JSONDecodeError
app = FastAPI()
webhook_call = {}
@app.get('/all')
async def get_all():
return webhook_call
@app.delete('/all', status_code=204)
async def delete_all():
webhook_call.clear()
return ''
@app.get('/')
async def index():
return "Hello Webhook!"
@app.get('/{endpoint:path}')
async def main(endpoint, response: Response):
try:
return webhook_call[endpoint]
except KeyError as identifier:
response.status_code = 404
return 'No Payload for this App'
@app.post('/{endpoint:path}', status_code=201)
async def post(endpoint, request: Request):
body = await request.body()
message = body.decode('utf-8')
try:
message = json.loads(message)
except JSONDecodeError:
pass
try:
webhook_call[endpoint].append(message)
except KeyError:
webhook_call[endpoint] = []
webhook_call[endpoint].append(message)
return message
@app.delete('/{endpoint:path}', status_code=204)
async def delete_endpoint_data(endpoint=None):
if endpoint in webhook_call.keys():#
del webhook_call[endpoint]
return ''
if __name__ == '__main__':
import uvicorn
uvicorn.run("WebhookMock:app", host='0.0.0.0', port=8000, debug=True, reload=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment