I have an endpoint in fastapi app that does the following:
- Checks if the file is a text file (
.txt) - Uploads the file to the S3 bucket
The codes:
upload.py
from app.services.file_upload import file_uploader
from fastapi import APIRouter, File, UploadFile
router = APIRouter(
prefix="/upload-file",
tags=["Upload File"],
responses={
200: {'description': 'Success'},
400: {'description': 'Bad Request'},
422: {'description': 'Unprocessable Entity'},
500: {'description': 'Internal Server Error'}
}
)
@router.post('/')
def upload_file(upload_file: UploadFile = File(
...,
description='Upload file'
)):
filename = upload_file.filename
file_content = upload_file.file
return file_uploader(filename, file_content)
The code where the check is done
file_upload.py
import io
from typing import BinaryIO, Dict
import boto3
from app.core.config import settings
from fastapi import HTTPException
def file_uploader(filename: str, file_content: BinaryIO) -> Dict[str, str]:
if not filename.lower().endswith('.txt'):
raise HTTPException(status_code=422, detail='Only .txt file allowed')
s3 = boto3.client('s3')
file_content = io.BytesIO(file_content.read())
s3.upload_fileobj(file_content, settings.S3_BUCKET, filename)
return {'message': 'File uploaded.'}
I want to test this endpoint to see if the file being uploaded is a text file or not.
conftest.py
from typing import Generator
import pytest
from app.main import app
from fastapi.testclient import TestClient
@pytest.fixture(scope="module")
def client() -> Generator:
with TestClient(app) as c:
yield c
test_upload.py
import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
def test_upload_file(client: TestClient):
test_file = 'test_file.docx'
files = {'file': ('test_file.docx', open(test_file, 'rb'))}
with pytest.raises(HTTPException) as err:
client.post('/api/v1/upload-file/', files=files)
assert err.value.status_code == 422
When I run this code, I get the following error
FAILED app/tests/api/v1/test_upload.py::test_upload_file - Failed: DID NOT RAISE <class 'fastapi.exceptions.HTTPException'>
What is the mistake I am doing?