Reputation: 21
I'm running a Python unit test on this script api.py
def download_models(gcs_model_path, local_path):
filesystem = gcsfs.GCSFileSystem(project=PROJECT) # , token='creds.json')
# download models from gcs
filesystem.get(gcs_model_path, local_path, recursive=True)
Here is the unit test script embedding_api_test.py
class EmbeddingAPITest(unittest.TestCase):
@mock.patch('gcsfs.GCSFileSystem')
def test_download_models(self, mock_filesystem):
mock_filesystem.return_value.get.return_value = []
download_models('gcs_model_path', 'local_path')
I'm getting this error on my terminal:
_______________________ ERROR collecting qa_api/tests/qa_api_test.py ________________________
qa_api/tests/qa_api_test.py:3: in <module>
from qa_api.api import app
qa_api/api.py:37: in <module>
download_models(GCS_MODEL_PATH, LOCAL_MODEL_PATH)
qa_api/api.py:21: in download_models
filesystem.get(gcs_model_path, local_path, recursive=True)
/opt/anaconda3/lib/python3.7/site-packages/fsspec/spec.py:556: in get
rpaths = self.find(rpath)
/opt/anaconda3/lib/python3.7/site-packages/fsspec/spec.py:371: in find
for path, dirs, files in self.walk(path, maxdepth, **kwargs):
/opt/anaconda3/lib/python3.7/site-packages/fsspec/spec.py:326: in walk
listing = self.ls(path, detail=True, **kwargs)
/opt/anaconda3/lib/python3.7/site-packages/gcsfs/core.py:767: in ls
out = self._list_objects(path)
/opt/anaconda3/lib/python3.7/site-packages/gcsfs/core.py:571: in _list_objects
items, prefixes = self._do_list_objects(path)
/opt/anaconda3/lib/python3.7/site-packages/gcsfs/core.py:604: in _do_list_objects
maxResults=max_results,
/opt/anaconda3/lib/python3.7/site-packages/gcsfs/core.py:504: in _call
raise e
/opt/anaconda3/lib/python3.7/site-packages/gcsfs/core.py:487: in _call
validate_response(r, path)
/opt/anaconda3/lib/python3.7/site-packages/gcsfs/core.py:130: in validate_response
raise HttpError(error)
E gcsfs.utils.HttpError: Anonymous caller does not have storage.objects.list access to the Google Cloud Storage bucket.
It seems I'm mocking it incorrectly but I don't understand why. Would really appreciate any help. Please let me know if more details are required. Thanks!
Upvotes: 2
Views: 922
Reputation: 102587
You should patch 'api.gcsfs.GCSFileSystem'
. E.g.
api.py
:
import gcsfs
PROJECT = "teresa.teng"
def download_models(gcs_model_path, local_path):
filesystem = gcsfs.GCSFileSystem(project=PROJECT)
filesystem.get(gcs_model_path, local_path, recursive=True)
test_api.py
:
import unittest
from unittest import mock
from api import download_models
class EmbeddingAPITest(unittest.TestCase):
@mock.patch('api.gcsfs.GCSFileSystem')
def test_download_models(self, mock_filesystem):
mock_filesystem.return_value.get.return_value = []
download_models('gcs_model_path', 'local_path')
mock_filesystem.assert_called_once_with(project="teresa.teng")
mock_filesystem.return_value.get.assert_called_once_with('gcs_model_path', 'local_path', recursive=True)
if __name__ == '__main__':
unittest.main()
unit test result:
.
----------------------------------------------------------------------
Ran 1 test in 0.001s
OK
Name Stmts Miss Cover Missing
----------------------------------------------------------------------
src/stackoverflow/64673419/api.py 5 0 100%
src/stackoverflow/64673419/test_api.py 11 0 100%
----------------------------------------------------------------------
TOTAL 16 0 100%
requirements.txt
:
gcsfs==0.7.1
Upvotes: 2