from fastapi import FastAPI, Request, Depends from fastapi.testclient import TestClient import hmac from starlette.routing import request_response from app.main import app from app.auth import auth_hook, auth_web, check_ref from os import environ import json environ['WEBHOOK_SECRET'] = "dfsgdsjghhgdaehlsdfjhjkdh" environ["BRANCH"] = "master" environ["TOKEN"] = "assdcvfgvh" secret_key = environ.get('WEBHOOK_SECRET') client = TestClient(app) @app.post("/test_auth", dependencies=[Depends(auth_hook)]) async def auth_test_handler(request: Request): return 200 @app.post("/test_ref", dependencies=[Depends(check_ref)]) async def auth_test_handler(request: Request): return 200 @app.get("/test_web", dependencies=[Depends(auth_web)]) async def web_test_hnadler(request: Request): return 200 def test_auth(): payload = {"Hello":"World"} msg = json.dumps(payload).encode() mac = hmac.new(secret_key.encode(), msg=msg, digestmod='sha1').hexdigest() response = client.post("/test_auth", json= payload, headers={"X-Hub-Signature": "sha1="+mac}) assert response.status_code == 200 response = client.post("/test_auth", headers={"X-Hub-Signature": "sha1="+mac}) assert response.status_code == 204 assert response.text == '{"detail":"Missing or bad content"}' response = client.post("/test_auth", json= payload, headers={"X-Hub-Signature": "sha="+mac}) assert response.status_code == 400 assert response.text == '{"detail":"Invalid signature"}' response = client.post("/test_auth", json=payload) assert response.status_code == 400 assert response.text == '{"detail":"Missing signature"}' response = client.post("/test_auth", json= payload, headers={"X-Hub-Signature": "sha1="+mac+"a"}) assert response.status_code == 403 assert response.text == '{"detail":"Unauthorized"}' def test_branch(): payload = {"ref": "refs/heads/master"} response = client.post("/test_ref", json= payload) assert response.status_code == 200 payload = {"ref": "refs/heads/test"} response = client.post("/test_ref", json= payload) assert response.status_code == 403 def test_web(): response = client.get('/test_web?token={}'.format(environ.get("TOKEN"))) assert response.status_code == 200 response = client.get('/test_web') assert response.status_code == 400 response = client.get('/test_web?token=a') assert response.status_code == 403