I have a function that takes some data, works on it and then send a message to storage queue based on the results (if the work failed or succeeded).
Below, a simplified version:
import json
import azure.functions as func
def main(job_data: dict, msg: func.Out[str]):
try:
# work with job_data
status = 'succeeded'
except:
status = 'failed'
msg.set(json.dumps({'status': status}))
The binding schema for this function is as follow:
{
"scriptFile": "__init__.py",
"bindings": [
{
"name": "job_data",
"type": "activityTrigger",
"direction": "in"
},
{
"type": "queue",
"direction": "out",
"name": "msg",
"queueName": "datalakejobs",
"connection": "AzureWebJobsStorage"
}
]
}
I am writing a test for this function using pytest. The problem is that I don't know how to mock the msg parameter of the main function. I tried to use the class func.Out, as follows:
import azure.functions as func
import pytest
import activity_send_message
@pytest.fixture
def job_data():
return {'job_data': 'job'} # actually, it is just a representation of the data
def test_upload_data(job_data):
r = activity_send_message.main(job_data, func.Out)
assert True
This test returns the following error:
> msg.set(json.dumps({'status': status}))
E TypeError: set() missing 1 required positional argument: 'val'
I think the problem is with the msg mock...
How can I mock the binding msg parameter to make it work?