How to mock a queue out binding in Azure Functions (python)?

Viewed 100

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?

0 Answers
Related