Move make_request function to a separate utils module

This commit is contained in:
2026-05-08 16:58:35 +02:00
parent 16bd8b25f2
commit 97f1a802fa
2 changed files with 64 additions and 30 deletions
@@ -13,38 +13,10 @@
# limitations under the License. # limitations under the License.
from src.jcloud_deployment_server.integrations.gitea.middlewares.signature import GiteaSignatureMiddleware from src.jcloud_deployment_server.integrations.gitea.middlewares.signature import GiteaSignatureMiddleware
from starlette.requests import Request from tests.utils.make_request import make_request, call_next
from fastapi import HTTPException
import pytest import pytest
import hashlib import hashlib
import hmac import hmac
import os
class DummyResponse:
def __init__(self):
self.called = True
async def call_next(request):
return DummyResponse()
def make_request(body: bytes, headers = None) -> Request:
scope = {
'type': 'http',
'method': 'POST',
'path': '/gitea/webhook',
'headers': [
(k.lower().encode(), v.encode())
for k, v in (headers or {}).items()
]
}
async def receive():
return {
'type': 'http.request',
'body': body
}
return Request(scope, receive)
def make_signature(body: bytes, secret: bytes) -> str: def make_signature(body: bytes, secret: bytes) -> str:
signature = hmac.new( signature = hmac.new(
@@ -98,6 +70,7 @@ async def test_GiteaSignatureMiddleware_valid_signature(body, secret):
req = make_request( req = make_request(
body, body,
'/gitea/webhook',
headers headers
) )
@@ -136,7 +109,8 @@ async def test_GiteaSignatureMiddleware_invalid_signature(signature, body, secre
req = make_request( req = make_request(
body, body,
headers '/gitea/webhook',
headers,
) )
res = await middleware.dispatch(req, call_next) res = await middleware.dispatch(req, call_next)
+60
View File
@@ -0,0 +1,60 @@
# Copyright 2026 jCloud
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from starlette.requests import Request
__all__ = [
'call_next',
'make_request'
]
class _DummyResponse:
def __init__(self):
self.called = True
async def call_next(request):
return _DummyResponse()
def make_request(body: bytes, path: str, headers: dict = None) -> Request:
'''
Makes a request.
:param body: The request body.
:type body: bytes
:param path: The path.
:type path: str
:param headers: The headers.
:type headers: dict
:return: The request.
:rtype: Request
'''
scope = {
'type': 'http',
'method': 'POST',
'path': path,
'headers': [
(k.lower().encode(), v.encode())
for k, v in (headers or {}).items()
]
}
async def receive():
return {
'type': 'http.request',
'body': body
}
return Request(scope, receive)