1

I am writing test for my FastApi application and I want to test this endpoint:

@router.post(
    "",
    response_class=Response,
    status_code=201,
    responses={
        201: {"description": "File created"},
        401: {"description": "User unauthorized"},
        400: {"description": "Error in creation"},
    },
)
async def save_cota(
    *,
    energy_asset_id: int,
    cota_file: UploadFile = File(...),
    current_user=Security(get_current_user, scopes=["scope:base", "admin:guane"]),
):
    """
    Create boundarys by extracting data from an xlsx file.
    **Args**:
    - **boundarys file** (File, optional): minimum values to create boundarys from xlsx file.

    **Returns**:
    - **Array**: List of sic_codes.
    """
    await cota_service.save_cota(cota_file=cota_file, energy_asset_id=energy_asset_id)
    return Response(status_code=201)

So I have written the following test:

def get_headers_files(self):
        response_acces_token = self.response_token.json()
        token = response_acces_token.get('access_token', '')
        headers = {
            'accept': '*/*',
            'Authorization': f'Bearer {token}',
        }
        return headers

def test_save_cota(test_app, monkeypatch):
    async def mock_save_cota(energy_asset_id: int, cota_file: UploadFile):
        return None
    monkeypatch.setattr(cota_service, "save_cota", mock_save_cota)
    file_content = b"some file content"
    file = UploadFile(filename="test.xlsx", file=BytesIO(file_content))
    response = test_app.post(
        "/api/cotas",
        params={"energy_asset_id": 123},
        files={"cota_file": file},
        headers=auth_test_service.get_headers_files()
    )
    assert response.status_code == 201

But the test keeps loading and doesn't run. Is there something I'm doing wrong? I'm not sure I'm sending the file right in the request.

Output:

| ============================= test session starts ==============================
| platform linux -- Python 3.8.3, pytest-7.4.0, pluggy-1.2.0 -- /usr/local/bin/python
| cachedir: .pytest_cache
| rootdir: /usr/src/app
| configfile: pytest.ini
| plugins: requests-mock-1.11.0, asyncio-0.21.1, mock-3.11.1, anyio-3.7.1, cov-4.1.0, time-machine-2.11.0
| asyncio: mode=strict
| collecting ... collected 1 item
| 
exited with code 137
Diego L
  • 185
  • 1
  • 9

1 Answers1

0

Finally I used the following approach which worked:

def test_save_cota(test_app, monkeypatch):
    async def mock_save_cota(energy_asset_id: int, cota_file: UploadFile):
        return None
    monkeypatch.setattr(cota_service, "save_cota", mock_save_cota)
    monkeypatch.setattr(auth_service, "get_current_user", auth_test_service.mock_login)
    file_content = "column1,column2,column3\nvalue1,value2,value3"
    temp_file = NamedTemporaryFile(delete=False, mode="w")
    temp_file.write(file_content)
    temp_file.close()
    _files = {"cota_file": (temp_file.name, open(temp_file.name, "rb"))}
    response = test_app.post(
        "/api/cotas",
        params={"energy_asset_id": 123},
        files=_files,
        headers=auth_test_service.HEADERS_ALL
    )
    os.unlink(temp_file.name)
    assert response.status_code == 201

Keep in mind that it is a text file (csv), if it were bytes, some things change slightly:

def test_save_cota(test_app, monkeypatch):
    async def mock_save_cota(energy_asset_id: int, cota_file: UploadFile):
        return None
    monkeypatch.setattr(cota_service, "save_cota", mock_save_cota)
    monkeypatch.setattr(auth_service, "get_current_user", auth_test_service.mock_login)
    file_content = b"some file content"
    temp_file = NamedTemporaryFile(delete=False)
    temp_file.write(file_content)
    temp_file.close()
    _files = {"cota_file": (temp_file.name, open(temp_file.name, "rb"))}
    response = test_app.post(
        "/api/cotas",
        params={"energy_asset_id": 123},
        files=_files,
        headers=auth_test_service.HEADERS_ALL
    )
    os.unlink(temp_file.name)
    assert response.status_code == 201
Diego L
  • 185
  • 1
  • 9