how to unit test flask method using google cloud storage library

Question

Im trying to create unit tests for this method but im getting the error saying credentials not found

from io import BytesIO

from google.auth.exceptions import DefaultCredentialsError
from google.cloud import storage
from fastapi import HTTPException

storage_client = storage.Client()
bucket_name = "stupid-bucket-name-1234567890"  

async def  upload_to_gcs(file, file_name: str) -> str:

    try:
        bucket = storage_client.bucket(bucket_name)
        blob = bucket.blob(file_name)
        binary_data = await file.read()
        blob.upload_from_file(BytesIO(binary_data), content_type=file.content_type)
        file_url = blob.public_url
        return file_url
    except DefaultCredentialsError as exe:
        raise HTTPException(status_code=500, detail=str(exe))
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

now here is my unit test code

import sys
import pytest
from unittest.mock import patch, MagicMock, AsyncMock
from fastapi import HTTPException
from google.auth.exceptions import DefaultCredentialsError

# Patch BEFORE importing cloud_service
with patch("package.svc.cloud.storage.Client") as mock_client:
    mock_storage_client = mock_client.return_value
    mock_bucket = MagicMock()
    mock_blob = MagicMock()
    mock_storage_client.bucket.return_value = mock_bucket
    mock_bucket.blob.return_value = mock_blob
    mock_blob.public_url = "https://fake-url.com/file.txt"
    import importlib
    cloud_service = importlib.import_module("package.svc.cloud")

@pytest.mark.asyncio
async def test_upload_file_to_gcs_success():
    mock_file = AsyncMock()
    mock_file.read.return_value = b"test data"
    mock_file.content_type = "text/plain"
    url = await cloud_service.upload_file_to_gcs(mock_file, "file.txt")
    assert url == "https://fake-url.com/file.txt"
    cloud_service.storage_client.bucket.return_value.blob.return_value.upload_from_file.assert_called_once()

but when I run this, it says: Failed to load service account credentials from ... No key could be found

I don't understand why it is looking for the actual keys even though I have already mocked the lines in the code that would return the Client object.

here's the error stacktrace

venv3.11/lib/python3.11/site-packages/google/auth/_default.py:473: in _get_service_account_credentials
    credentials = service_account.Credentials.from_service_account_info(
venv3.11/lib/python3.11/site-packages/google/oauth2/service_account.py:243: in from_service_account_info
    signer = _service_account_info.from_dict(
venv3.11/lib/python3.11/site-packages/google/auth/_service_account_info.py:57: in from_dict
    signer = crypt.RSASigner.from_service_account_info(data)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
venv3.11/lib/python3.11/site-packages/google/auth/crypt/base.py:109: in from_service_account_info
    return cls.from_string(
venv3.11/lib/python3.11/site-packages/google/auth/crypt/_python_rsa.py:173: in from_string
    raise exceptions.MalformedError("No key could be detected.")
E   google.auth.exceptions.MalformedError: No key could be detected.

The above exception was the direct cause of the following exception:
tests/cloud_service_security_test.py:8: in <module>
    with patch("requestprocessor.services.cloud_service.storage.Client") as mock_client:
/usr/local/lib/python3.11/unittest/mock.py:1411: in __enter__
    self.target = self.getter()
                  ^^^^^^^^^^^^^
/usr/local/lib/python3.11/pkgutil.py:705: in resolve_name
    mod = importlib.import_module(s)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^
/usr/local/lib/python3.11/importlib/__init__.py:126: in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
<frozen importlib._bootstrap>:1206: in _gcd_import
    ???
<frozen importlib._bootstrap>:1178: in _find_and_load
    ???
<frozen importlib._bootstrap>:1149: in _find_and_load_unlocked
    ???
<frozen importlib._bootstrap>:690: in _load_unlocked
    ???
<frozen importlib._bootstrap_external>:940: in exec_module
    ???
<frozen importlib._bootstrap>:241: in _call_with_frames_removed
    ???
requestprocessor/services/cloud_service.py:11: in <module>
    storage_client = storage.Client()
                     ^^^^^^^^^^^^^^^^
venv3.11/lib/python3.11/site-packages/google/cloud/storage/client.py:247: in __init__
    super(Client, self).__init__(
venv3.11/lib/python3.11/site-packages/google/cloud/client/__init__.py:338: in __init__
    _ClientProjectMixin.__init__(self, project=project, credentials=credentials)
venv3.11/lib/python3.11/site-packages/google/cloud/client/__init__.py:286: in __init__
    project = self._determine_default(project)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
venv3.11/lib/python3.11/site-packages/google/cloud/client/__init__.py:305: in _determine_default
    return _determine_default_project(project)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
venv3.11/lib/python3.11/site-packages/google/cloud/_helpers/__init__.py:152: in _determine_default_project
    _, project = google.auth.default()
                 ^^^^^^^^^^^^^^^^^^^^^
venv3.11/lib/python3.11/site-packages/google/auth/_default.py:651: in default
    credentials, project_id = checker()
                              ^^^^^^^^^
venv3.11/lib/python3.11/site-packages/google/auth/_default.py:645: in <lambda>
    lambda: _get_gcloud_sdk_credentials(quota_project_id=quota_project_id),
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
venv3.11/lib/python3.11/site-packages/google/auth/_default.py:259: in _get_gcloud_sdk_credentials
    credentials, project_id = load_credentials_from_file(
venv3.11/lib/python3.11/site-packages/google/auth/_default.py:137: in load_credentials_from_file
    return _load_credentials_from_info(
venv3.11/lib/python3.11/site-packages/google/auth/_default.py:210: in _load_credentials_from_info
    credentials, project_id = _get_service_account_credentials(
venv3.11/lib/python3.11/site-packages/google/auth/_default.py:479: in _get_service_account_credentials
    raise new_exc from caught_exc
E   google.auth.exceptions.DefaultCredentialsError: ('Failed to load service account credentials from /home/kali/.config/gcloud/application_default_credentials.json', MalformedError('No key could be detected.'))

always put full error message because there are other useful information. — furas
– furas, Commented Jul 10 at 0:53
error shows wrong data in file /home/kali/.config/gcloud/application_default_credentials.json - you have to replace it with correct file. And if you mocked it up then probably you made it incorrectly. — furas
– furas, Commented Jul 10 at 1:35

yannco · Accepted Answer · 2025-07-11 19:40:08Z

I tried recreating your issue and I suggest moving the storage_client = storage.Client() inside the function so that the instantiation happens during test execution and not during module load time.

from io import BytesIO


from google.auth.exceptions import DefaultCredentialsError
from google.cloud import storage
from fastapi import HTTPException




bucket_name = "stupid-bucket-name-1234567890" 


async def  upload_to_gcs(file, file_name: str) -> str:


   try:
       storage_client = storage.Client() #move this line here inside the function
       bucket = storage_client.bucket(bucket_name)
       blob = bucket.blob(file_name)
       binary_data = await file.read()
       blob.upload_from_file(BytesIO(binary_data), content_type=file.content_type)
       file_url = blob.public_url
       return file_url
   except DefaultCredentialsError as exe:
       raise HTTPException(status_code=500, detail=str(exe))
   except Exception as e:
       raise HTTPException(status_code=500, detail=str(e))

Also, update your unit test code so that the patch and the import happens inside the async test function. Here’s what I did:

import sys
import pytest
import google.auth.credentials
from unittest.mock import patch, MagicMock, AsyncMock
from fastapi import HTTPException
from google.auth.exceptions import DefaultCredentialsError


@pytest.mark.asyncio
async def test_upload_file_to_gcs_success():
   # Patch BEFORE importing cloud_service
   with patch("package.svc.cloud.storage.Client") as mock_client:
       mock_storage_client = mock_client.return_value
       mock_bucket = MagicMock()
       mock_blob = MagicMock()
       mock_storage_client.bucket.return_value = mock_bucket
       mock_bucket.blob.return_value = mock_blob
       mock_blob.public_url = "https://fake-url.com/file.txt"
       import importlib
       cloud_service = importlib.import_module("package.svc.cloud")


       mock_file = AsyncMock()
       mock_file.read.return_value = b"test data"
       mock_file.content_type = "text/plain"
       url = await cloud_service.upload_to_gcs(mock_file, "file.txt")
       assert url == "https://fake-url.com/file.txt"
       mock_blob.upload_from_file.assert_called_once()

Collectives™ on Stack Overflow

how to unit test flask method using google cloud storage library

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related