Как правильно применить mock post.request в тесте для FastAPI?

Question

[Влад]

есть эндпоинт, по запросу к которому возвращается в ответ файл(конвертация из pdf в текст):

@router.post('/upload_file/', status_code=status.HTTP_200_OK)
async def upload_file(url: AnyUrl = None, file: UploadFile = File(None)) -> Response:
 ###логика

В логике этого эндпоинта у меня прописан ещё один post запрос на другой эндпоинт. На самом деле подробности работы программы здесь и не нужны. Я просто никак не могу понять каким образом мне в тестовых целях, сделать так чтобы не обращаться к этому внутреннему эндпоинту (подменить его заранее заготовленными данными).

Вот логика моего эндпоинта:

async def get_response(file: Union[UploadFile, bytes], endpoint=settings.PDF_TO_JPEG_ENDPOINT):
    """Getting response from endpoint and passing it further for converting"""
    async with aiohttp.ClientSession() as session:
        uploaded_file = {'file': file}
        if type(file) != bytes:
            uploaded_file = {'file': await file.read()}  # multipart encoded file
        async with session.post(endpoint, data=uploaded_file) as resp:
            return await main(await resp.content.read())


async def get_file_from_url(url: AnyUrl):
    """Get file from URL"""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.content.read()


def unzip(resp: bytes) -> List[bytes]:
    """Extract pages from archive"""
    pages = []
    with ZipFile(BytesIO(resp)) as unziped_pages:
        for page in unziped_pages.namelist():
            pages.append(unziped_pages.read(page))
        return pages


def ocr_pages(page: bytes):
    """OCR pages extracted from archive"""
    with Image.open(BytesIO(page)) as im:
        with PyTessBaseAPI(lang=settings.TESSEROCR_DEFAULT_LANG, psm=PSM.AUTO_OSD,
                           oem=OEM.TESSERACT_ONLY) as api:
            api.SetImage(im)
            text = api.GetUTF8Text()
            word_conf = api.MapWordConfidences()
    return text, word_conf


def combine_two(resp: bytes):
    start = time.time()
    pages = unzip(resp)
    text_response = ''
    with ProcessPoolExecutor(max_workers=5) as pool:
        result = pool.map(ocr_pages, pages)
    for page_text, _ in result:
        text_response += page_text
    print('FINISHED AT {:.1f} sec'.format(time.time() - start))
    return text_response


async def main(resp: bytes):
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, partial(combine_two, resp))
    return result

Меня на самом деле интересует строка на самом верху

async with session.post(endpoint, data=uploaded_file) as resp:

Я написал для неё mock request, чтобы в ответ приходил заранее приготовленный файл.:

with requests_mock.Mocker() as m:
    m.post(settings.PDF_TO_JPEG_ENDPOINT, content=open(PATH, 'rb').read())
    resp = requests.post(settings.PDF_TO_JPEG_ENDPOINT)

А вот что в моей голове не укладывается, так это как и куда вставить этот мок в мой тест, чтобы не копировать в него всю вышеуказанную логику.

т.е. Что бы я хотел получить:

client = TestClient(app)

def test_do_something():
    response = client.post('/upload_file/')
    #делаю что-нибудь с response#
    #делаю какой-нибудь assert#

Comments

[Gennady S]

По-моему, вариантов множество. Можно так:

def test_do_something():
    response = client.post(settings.ENDPOINTS.PDF_TO_TEXT)

— Вам достаточно подменять для тестового окружения PDF_TO_TEXT с '/upload_file/' на '/mock/upload_file/'

Можно создать сервис, который будет выполнять логику внутри upload_file и который можно подменить на основе окружения / настроек.