Unit testing for GCP upload

Viewed 1539

I have the following method for which I need to write unit test, I am new to using pytest. I am uploading a file to a bucket on GCP using the google.cloud storage library in python.

def upload_to_gcs(topic, upload_file_name, upload_file_location, gcs_bucket_name, gcp_key_json):
    try:
        client = storage.Client.from_service_account_json(gcp_key_json)
        bucket = client.get_bucket(gcs_bucket_name)
        gcs_folder_loc = topic + '/' + upload_file_name
        bucket.blob(gcs_folder_loc).upload_from_filename(upload_file_location)
        _LOGGER.info("File %s has been uploaded to folder %s", upload_file_name, topic)
    except Exception as e:
        _LOGGER.exception("Caught unexpected exception while uploading file to GCS %s", e)
    finally:
        try:
            if os.path.isfile(upload_file_location):
                os.remove(upload_file_location)
        except OSError as e:
            _LOGGER.exception("Error: %s - %s." % (e.filename, e.strerror))

I have the following but I am not sure how to test the upload, Any help is appreciated -

def test_upload_to_gcs():
    mock_client = mock.create_autospec(storage.Client)
    mock_bucket = mock.create_autospec(storage.bucket.Bucket)
    mock_bucket.name = 'test_bucket'
    mock_client.get_bucket.return_value = mock_bucket
    mock_blob = mock.create_autospec(storage.blob.Blob)
    mock_blob.name = 'test_blob'
    mock_bucket.get_blob.return_value = mock_blob
0 Answers
Related