python - 如何使用 Http 触发器和队列消息输出修复 Azure 函数 (Python) 测试错误

Question

当我在 python Azure Function 上运行测试时，出现以下错误:

obj = super().__new__(cls, *args, **kwds)
TypeError: object.__new__() takes exactly one argument (the type to instantiate)

当函数调用 msg.set() 时，我收到错误，因为我从测试传入的值不正确。

这是我正在测试的功能:

import logging    
import azure.functions as func
    
def main(req: func.HttpRequest, msg: func.Out[func.QueueMessage]) -> func.HttpResponse:
    
    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

    if name:
        msg.set(name)
        return func.HttpResponse(f"Hello {name}!")
    else:
        return func.HttpResponse(
             "Please pass a name on the query string or in the request body",
             status_code=400
        )

这是测试代码:

# test_some.py
import unittest
import azure.functions as func

from . import main

class Test_Some(unittest.TestCase):
    def test_my_function_for_no_name(self):
        testVal = 'Some Guy'
        # Construct a mock HTTP request.
        req = func.HttpRequest(
            method='GET',
            body=None,
            url='/api/HttpExample',
            params={'name': testVal})

        queue_msg = func.QueueMessage(body=b'')

        msg = func.Out(queue_msg)

        # Call the function.
        resp = main(req, msg)

        # Check the output.
        self.assertEqual(
            resp.get_body(),
            b'Hello Some Guy!',
        )

if __name__ == '__main__': unittest.main()

Answer 1

测试代码应该是:

# HttpTrigger1/test_queryurls.py
from . import main
import unittest
import logging
import azure.functions as func

class Test_Some(unittest.TestCase):
    def test_my_function_for_no_name(self):
        testVal = 'Some Guy'
        # Construct a mock HTTP request.
        req = func.HttpRequest(
            method='GET',
            body=None,
            url='/api/HttpExample',
            params={'name': testVal})

        queue_msg = func.QueueMessage(body=b'')
        queue_msg.__setattr__('set', set)

        resp = main(req, queue_msg)

        # Check the output.
        self.assertEqual(
            resp.get_body(),
            b'Hello Some Guy!',
        )

    def set(inStr:str):
        logging.info(inStr)

if __name__ == '__main__': unittest.main()

问题是如何组合第二个对象以从单元测试传递到函数。第二个对象的函数签名是 msg:func.Out[func.QueueMessage]。 func.Out 是一个抽象类，因此无法实例化。 func.QueueMessage 可以实例化，但没有 set 方法，因此不能按原样使用。因此，我创建了一个 func.QueueMessage 对象，并使用 setsttr 向该对象添加了一个空白集方法。如果我需要它做更多事情，我可能会修改新的“set”方法。