fastapi icon indicating copy to clipboard operation
fastapi copied to clipboard

How to test against UploadFile parameter

Open mwilson8 opened this issue 4 years ago • 16 comments

First check

  • [ ] I added a very descriptive title to this issue.
  • [ ] I used the GitHub search to find a similar issue and didn't find it.
  • [ ] I searched the FastAPI documentation, with the integrated search.
  • [ ] I already searched in Google "How to X in FastAPI" and didn't find any information.
  • [ ] I already read and followed all the tutorial in the docs and didn't find an answer.
  • [ ] I already checked if it is not related to FastAPI but to Pydantic.
  • [ ] I already checked if it is not related to FastAPI but to Swagger UI.
  • [ ] I already checked if it is not related to FastAPI but to ReDoc.
  • [ ] After submitting this, I commit to one of:
    • Read open issues with questions until I find 2 issues where I can help someone and add a comment to help there.
    • I already hit the "watch" button in this repository to receive notifications and I commit to help at least 2 people that ask questions in the future.
    • Implement a Pull Request for a confirmed bug.

Example

Here's a self-contained, minimal, reproducible, example with my use case:

from fastapi import FastAPI

router = FastAPI()


@router.post("/_config")
def create_index_config(upload_file: UploadFile = File(...)):

    config = settings.reads()
    created_config_file: Path = Path(config.config_dir, upload_file.filename)

    try:
        with created_config_file.open('wb') as write_file:
            shutil.copyfileobj(upload_file.file, write_file)

    except Exception as err:
        raise HTTPException(detail=f'{err} encountered while uploading {upload_file.filename}',
                            status_code=HTTPStatus.INTERNAL_SERVER_ERROR)
    finally:
        upload_file.file.close()

    return JSONResponse({"message": f'uploaded config {upload_file.filename}'})

Description

I'm just trying to test against an endpoint that uses uploadfile and I can't find how to send a json file to fastapi

def test_one():
  _test_upload_file = <path to file>
 with TestClient(app) as client:
        # TODO how to upload a file
        response = client.post('/_config',
                               data= <assuming something goes here ???> )
        assert response.status_code == HTTPStatus.CREATED

Environment

  • OS: [e.g. Linux / Windows / macOS]: OSX/docker
  • FastAPI Version [e.g. 0.3.0]: fastapi==0.54.1
  • Python version: Python 3.6.8

Additional context

data=_test_upload_file.open('rb') yields a 422 error I have also tried this and get a 422 error

mwilson8 avatar Jun 08 '20 17:06 mwilson8

I've solved this yesterday on gitter. Could you search for "UploadFile" there? I'm busy now. I can post it here later if I remember.. or you can check there and post the solution here as well to help others.

Kludex avatar Jun 08 '20 17:06 Kludex

Here you have: https://stackoverflow.com/questions/60783222/how-to-test-a-fastapi-api-endpoint-that-consumes-images

Example:

with open(fpath, "wb") as f:
    response = client.post("/", files={"file": ("filename", f, "image/jpeg")})

Kludex avatar Jun 08 '20 17:06 Kludex

Here you have: https://stackoverflow.com/questions/60783222/how-to-test-a-fastapi-api-endpoint-that-consumes-images

Example:

with open(fpath, "wb") as f:
    response = client.post("/", files={"file": ("filename", f, "image/jpeg")})

I saw that but couldn't get it to work with my application

On gitter I found this

def test_upload(tmp_path):
    url = app.url_path_for("create_upload_files")
    with TestClient(app) as client:
        f = tmp_path / 'fileupload'
        with open(f, 'wb') as tmp:
            tmp.write(b'upload this')
        with open(f, 'rb') as tmp:
            files = {'files': tmp}
            response = client.post(url, files=files)
            assert response.status_code == 200

But that gives a 422 error when I try it in my code

    with TestClient(app) as client:
        # TODO how to upload a file
        with _test_upload_file.open('rb') as tmp:
            files = {'files': tmp}
            response = client.post('/_config',
                               files=files)
            assert response.status_code == HTTPStatus.CREATED

mwilson8 avatar Jun 08 '20 18:06 mwilson8

Big dumb on my part..

Working test:

_test_upload_file = Path('/usr/src/app/tests/files', 'new-index.json')
    _files = {'upload_file': _test_upload_file.open('rb')}
    with TestClient(app) as client:
        response = client.post('/_config',
                                files=_files)
        assert response.status_code == HTTPStatus.CREATED

    # remove the test file from the config directory
    _copied_file = Path('/usr/src/app/config', 'new-index.json')
    _copied_file.unlink()

I previously had the upload_file in the test named files -- FYSA that name must match the parameter of your endpoint, in my case it's upload_file so files wasn't working

mwilson8 avatar Jun 08 '20 18:06 mwilson8

Follow up with the code from Gitter also worked, thanks :)

mwilson8 avatar Jun 08 '20 18:06 mwilson8

Thanks for the help here @Kludex ! :clap: :bow:

Thanks for reporting back and closing the issue @mwilson8 :+1:

tiangolo avatar Jun 09 '20 11:06 tiangolo

Here you have: https://stackoverflow.com/questions/60783222/how-to-test-a-fastapi-api-endpoint-that-consumes-images

Example:

with open(fpath, "wb") as f:
    response = client.post("/", files={"file": ("filename", f, "image/jpeg")})

This didn't work for me. I had to change the with open(fpath, "wb") as f to with open(fpath, "rb") as f: Using read instead of write operation here works :)

rohansaw avatar Nov 18 '20 14:11 rohansaw

A slight follow up, how do you test against the actual function as opposed to the endpoint?

Working test against the endpoint:

_test_upload_file = Path('filepath')
    _files = {'upload_file': _test_upload_file.open('rb')}
    with TestClient(app) as client:
        response = client.post('/_config',
                               files=_files,
                               )
    assert response.status_code == HTTPStatus.OK

But I also want to test with

response = service.function_within_endpoint(upload_file=<...what goes here...>)

mwilson8 avatar Dec 17 '20 22:12 mwilson8

Have you tried passing an instance of UploadFile?

Mause avatar Dec 17 '20 23:12 Mause

Have you tried passing an instance of UploadFile?

I had to switch to the httpx.AsyncClient approach as specified here to pass an UploadFile instance, since reading an UploadFile is an async operation

holyoaks avatar Jul 02 '21 14:07 holyoaks

Here you have: https://stackoverflow.com/questions/60783222/how-to-test-a-fastapi-api-endpoint-that-consumes-images

Example:

with open(fpath, "wb") as f:
    response = client.post("/", files={"file": ("filename", f, "image/jpeg")})

what if the upload file is text file?

nafisBorkar avatar Oct 26 '21 14:10 nafisBorkar

A slight follow up, how do you test against the actual function as opposed to the endpoint?

Working test against the endpoint:

_test_upload_file = Path('filepath')
    _files = {'upload_file': _test_upload_file.open('rb')}
    with TestClient(app) as client:
        response = client.post('/_config',
                               files=_files,
                               )
    assert response.status_code == HTTPStatus.OK

But I also want to test with

response = service.function_within_endpoint(upload_file=<...what goes here...>)

This code doesn't work for me.

   def test_put(self):
       filepath = ".\\image.jpeg"
       with open(filepath, "rb") as f:
           response = self.client.put('/frame/',
                                      files={'upload_file': f},
                                      headers={'accept': 'application/json', 'Content-Type': 'multipart/form-data'})
       self.assertEqual(response.status_code, 200)

But my function should get a list of UploadFile:

async def add(files: List[UploadFile] = File(...), db: Session = Depends(get_db)):

Any ideas how can I test it?

AlexSidDev avatar Mar 19 '22 11:03 AlexSidDev

Hi, I would like to test UploadFile with its extension. In my situation, sending an input file works fine but, its content_type is an empty string. How can I ensure that the content_type must exist?

Here is my code.

  • test_app.py
def test_validate_image(self):
  client = TestClient(app)
  files = {"image": open("path/to/image.jpg", "rb")}
  response = client.post("/validate/", files=files)
  • app.py
@app.post("/validate/")
async def validate_image(image: UploadFile = File(...)):
  # Some other code
  return {
        "filename": image.filename,
        "content_type": image.content_type,
    }

The result looks like the following below

{"filename": "image.jpg", "content_type":  ""}

luangtatipsy avatar Apr 05 '22 05:04 luangtatipsy

Even I am facing the same problem

Code to upload a tarred file

@router.post("/upload_container")
async def post_container(file: UploadFile):
    if file.content_type != "application/x-tar":
        raise HTTPException(status_code=400, detail="Can only accept tarred version of Docker file")
    return {"filename" : file.filename}

Test case for the code

def test_add_docker():    
    _files = {'file.tar':open('file.tar', 'rb')}

    response = client.post("/upload_container", files=_files)
    assert response.status_code == HTTPStatus.OK

Error I received

`E       assert 422 == <HTTPStatus.OK: 200>
E        +  where 422 = <Response [422]>.status_code
E        +  and   <HTTPStatus.OK: 200> = HTTPStatus.OK

test/test_post_container.py:66: AssertionError`

tamil07 avatar Jun 08 '22 07:06 tamil07

A slight follow up, how do you test against the actual function as opposed to the endpoint?

Working test against the endpoint:

_test_upload_file = Path('filepath')
    _files = {'upload_file': _test_upload_file.open('rb')}
    with TestClient(app) as client:
        response = client.post('/_config',
                               files=_files,
                               )
    assert response.status_code == HTTPStatus.OK

But I also want to test with

response = service.function_within_endpoint(upload_file=<...what goes here...>)

A bit late, but I was just attempting to do this same thing as well and came across your comment. This is how I accomplished what you had asked:

import os
import shutil
from tempfile import SpooledTemporaryFile
from fastapi import FastAPI, UploadFile


app = FastAPI()


async def upload_file(file_path: str):

	tmp_file = SpooledTemporaryFile()

	with open(file_path, "rb") as path:
		tmp_file.write(path.read())

	_ = tmp_file.seek(0)

	file = UploadFile(filename=os.path.split(file_path)[1], file=tmp_file)

	await upload(file)


@app.post("/upload")
async def upload(file: UploadFile):

	pkg_dir = os.path.abspath(os.path.join(os.path.abspath(os.path.dirname(__file__)), os.pardir))

	try:
		with open(f"{pkg_dir}/static/{file.filename}", "wb") as create_file:
			shutil.copyfileobj(file.file, create_file)
	finally:
		await file.close()

	return { "results":  200, "file": file.filename }

This may not be the (most) correct way, but it works. Hopefully it'll help someone else one day.

MLBZ521 avatar Aug 26 '22 22:08 MLBZ521

This code works for me, I am sending multiple csv's to an endpoint which receives a list of UploadFile 🥇 🥇 🥇

import json
import uvicorn
from fastapi.testclient import TestClient
from dotenv import load_dotenv

load_dotenv()
from main import app

client = TestClient(app)


def test_user_profiles_csv():
    files = [("files", open("profiles_csv_1.csv", "rb")),
             ("files", open("profiles_csv_2.csv", "rb"))]
    
    response = client.post(
        "/api/v1/profiles/csv",
        files=files,
        headers={"Authorization": f"Bearer {get_access_token()}"},
    )
    assert response.status_code == 201

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

aldomatus avatar Sep 20 '22 17:09 aldomatus

Thanks for the discussion everyone! With regards to @mwilson8's second question about testing the function directly, passing an instance of UploadFile as suggested by @Mause worked? I would imagine it did. Also, in that case, it would be just testing functions, it wouldn't really have much to do with FastAPI in that case. If your questions/use cases are solved now, then you can close the issue. 🤓

For other questions from others, if you still haven't found a solution to your questions, as they vary a bit, please create a new issue, with a self-contained example, etc. Thanks! ☕

tiangolo avatar Oct 19 '22 16:10 tiangolo

Assuming the original need was handled, this will be automatically closed now. But feel free to add more comments or create new issues or PRs.

github-actions[bot] avatar Oct 30 '22 00:10 github-actions[bot]