I tried to mock the following data using unittest, but couldnt mock it since the function is not returning anything. Please help me out with this..
It would be helpful even if it can be done using Pytest library too.
from google.cloud import storage
import tempfile
import os, shutil
import zipfile
import io
import sys
from os import listdir
bucket_name = sys.argv[1]
file_id = sys.argv[2]
file_name = sys.argv[3]
print(bucket_name)
print(file_id)
print(file_name)
# storage_client = storage.Client()
# bucket = storage_client.get_bucket(bucket_name)
# blobpath = '%s/%s' % (file_id, file_name)
# blob = bucket.blob(blobpath)
# blob.download_to_filename('/tmp/downloaded_file.txt')
# prefix = sys.argv[1]
# bucket_name = sys.argv[2]
# wf_name = sys.argv[3]
folder = '/tmp/'
#initialising an empty array
files = []
logger.info(f'prefix = {prefix}')
logger.info(f'bucket = {bucket_name}')
logger.info(f'wf_name = {wf_name}')
#connecting with google storage account
def fetch_files():
storage_client = storage.Client()
blobs = storage_client.list_blobs(bucket_name, prefix=prefix)
#fetching data andd storing all files in files array.
for blob in blobs:
logger.info(f"file={blob.name}")
destination_file_name = f"{folder}{blob.name}"
base_folder, filename = os.path.split(destination_file_name)
os.makedirs(base_folder, exist_ok=True)
blob.download_to_filename(destination_file_name)
files.append(destination_file_name)
#here we are checking manifest.json...........
json_file_path = f"{folder}{prefix}/Manifest.json"
with open(json_file_path, 'r', encoding='utf-8') as reader:
json_content = reader.read()
#After reading existing manifest file,initialise a manifest dictionary and update the manifest file with the dictionary data we get it from environment variables..
manifest = {}
manifest['task_id'] = prefix
manifest['workflow_name'] = wf_name
attachments = []
emailPageCount = ''
attachmentThreshold = None
#if manifest.json is found,then do the following:--
if (json_content):
obj = json.loads(json_content)
print('this is the original manifest json: ', obj)
emailPageCount = obj['emailPageCount']
attachmentThreshold = obj.get('attachmentFilterThreshold')
in_attachments = obj['attachments']
for in_attachment in in_attachments:
attachment = {}
attachment['attachment_name'] = f"{folder}{prefix}/{in_attachment['attachment_name']}"
attachment['display_document'] = f"{folder}{prefix}/{in_attachment['display_document']}"
attachment['attachment_size'] = in_attachment.get('attachment_size')
doc_id = str(uuid.uuid4())
attachment['doc_id'] = f'{doc_id}'
attachment['search_metadata_id'] = callSearchMetaApi(
f"{prefix}/{in_attachment['display_document']}")
attachments.append(attachment)
manifest['attachments'] = attachments
manifest['emailPageCount'] = emailPageCount
manifest['attachmentThreshold'] = attachmentThreshold
logger.info(f"manifest={manifest}")
#here manifest is a dictionary and outfile is destination file
#input for json.dump is dictionary and destination file storing data in /tmp/mainfest.json path
#we are converting python object to json object using json.dump
with open("/tmp/manifest.json", "w") as outfile:
json.dump(manifest, outfile)
filesAsString = ",".join(files)
logger.info(f"filesAsString = {filesAsString}")
#here we are writing it to /tmp/listfiles folder
#storing all files in /tmp.listfiles which we got it from google cloud storage..
with open('/tmp/listfiles', 'w') as writer:
writer.write(filesAsString)
I need to mock these '/tmp/listfiles' and '/tmp/manifest.json' file writing using unittest for entire function.