首页 > 解决方案 > 用于运行测试用例的 Python Mockito 配置

问题描述

我正在为 Flask 应用程序编写单元测试,使用unittest并使用mockito来模拟需要服务器启动和运行的服务。

但是我无法弄清楚如何模拟服务器。我遇到了一些错误。

以下是代码片段:

节点.py

@app.route('/transaction', methods = ["POST"])
def transaction():
    req = request.get_json()
    print("\nreq")
    i = bitcoin.addTransaction(req)

    if req['purpose'] != 'redeem':
        users = getColumn(sheet_u,0)
        index = users.index(req['recipient'])
        user = getRow(sheet_u,index)
        hasPoints = int(user[labels_u.index("POINTS")])
        hasPoints += req['amount']
        setCell(sheet_u, index, labels_u.index("POINTS"), hasPoints)
        wb.save("db.xlsx")
    else:
        users = getColumn(sheet_u,0)
        index = users.index(req['sender'])
        user = getRow(sheet_u,index)
        hasPoints = int(user[labels_u.index("POINTS")])
        hasPoints -= req['amount']
        setCell(sheet_u, index, labels_u.index("POINTS"), hasPoints)
        wb.save("db.xlsx")
    r = requests.get(url = bitcoin.currentNodeUrl+"/save")  
    return jsonify({"note":"New transaction will be added to block "+str(i)})


@app.route('/transaction/broadcast', methods = ["POST"])
def transactionBroadcast():
    check = True
    failure = ""
    req = request.get_json()
    # Hiding some methods that change the value of check flag 
    if check:
        newTransaction = bitcoin.createNewTransaction(req['amount'],req['sender'],req['recipient'],req['purpose'],time.time())
        bitcoin.addTransaction(newTransaction)
        promises = []
        for node in bitcoin.networkNodes:
            response = ""
            try:
                r = requests.post(url=node+"/transaction",json=newTransaction) #this line throws error
                response = r.text   #program control doesn't come here
                print(newTransaction)
                print("\n")
                print(response)
            except requests.exceptions.ConnectionError:
                response = "{\"note\":\"The node is unavailable. It will sync itself later on.\"}"
            if 'note' in json.loads(response):
                promises.append(json.loads(response))
        if len(promises) == len(bitcoin.networkNodes):
            r = requests.get(url = bitcoin.currentNodeUrl+"/save")  
            return jsonify({"note":"Transaction broadcasted successfully"})
    else:
        return jsonify({"fail":"Invalid transaction. "+failure})
if __name__ == '__main__':
    bitcoin = Blockchain(sys.argv[1])
    app.run(port=sys.argv[1])

test_t2.py

from node import app
from flask import json
import unittest
import node
import requests

from mockito import when, mock, unstub

class TestSearch(unittest.TestCase):
    print("inside TEST class")

    def test_transactionBroadcast(self):  

        node.bitcoin = node.Blockchain('5000')
        response = mock({"note": "New transaction will be added to block 5"})
        when(requests).post('http://localhost:5001/transaction', strict = False ).thenReturn(response)

        data1 = {
            "sender" : "ph1000005",
            "recipient" : "TIAA",
            "amount" : 1,
            "purpose" : "redeem"
        }
        response = app.test_client().post(
            '/transaction/broadcast',
            json = data1,
            content_type='application/json',
        )
        print("running TransactionBroadcast test")
        print(response)
        self.assertEqual(response.get_json(), {
            # 'fail': "Invalid transaction. This user doesn't exist. Check userid again."
            "note": "Transaction broadcasted successfully"
        })
        unstub()
if __name__ == '__main__':
    unittest.main()

错误日志

C:\POC>coverage run test_t2.py
inside TEST class
[2018-12-12 19:58:22,181] ERROR in app: Exception on /transaction/broadcast [POST]
Traceback (most recent call last):
  File "c:\program files\python36\lib\site-packages\flask\app.py", line 2292, in wsgi_app
    response = self.full_dispatch_request()
  File "c:\program files\python36\lib\site-packages\flask\app.py", line 1815, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "c:\program files\python36\lib\site-packages\flask\app.py", line 1718, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "c:\program files\python36\lib\site-packages\flask\_compat.py", line 35, in reraise
    raise value
  File "c:\program files\python36\lib\site-packages\flask\app.py", line 1813, in full_dispatch_request
    rv = self.dispatch_request()
  File "c:\program files\python36\lib\site-packages\flask\app.py", line 1799, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "C:\POC\node.py", line 230, in transactionBroadcast
    r = requests.post(url=node+"/transaction",json=newTransaction)
  File "c:\program files\python36\lib\site-packages\mockito\mocking.py", line 89, in new_mocked_method
    self, method_name, *args, **kwargs)
  File "c:\program files\python36\lib\site-packages\mockito\mocking.py", line 44, in remembered_invocation_builder
    return invoc(*args, **kwargs)
  File "c:\program files\python36\lib\site-packages\mockito\invocation.py", line 103, in __call__
    """ % (self, "\n    ".join(str(invoc) for invoc in stubbed_invocations)))
mockito.invocation.InvocationError:
You called

        post(url='http://localhost:5001/transaction', json={'amount': 1, 'purpose': 'redeem', 'recipient': 'TIAA', 'sender': 'ph1000005', 'timestamp': 1544624902.1440911, 'transactionId': 'd919ffb4080f47e890957f3f3edc97a1'}),

which is not expected. Stubbed invocations are:

        post('http://localhost:5001/transaction', strict=False)

(Set strict to False to bypass this check.)

running TransactionBroadcast test
<Response streamed [500 INTERNAL SERVER ERROR]>
F
======================================================================
FAIL: test_transactionBroadcast (__main__.TestSearch)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "test_t2.py", line 49, in test_transactionBroadcast
    "note": "Transaction broadcasted successfully"
AssertionError: None != {'note': 'Transaction broadcasted successfully'}

----------------------------------------------------------------------
Ran 1 test in 0.682s

FAILED (failures=1)

现在,我不确定我做错了什么。/transaction路由中有一个内部路由调用/transaction/broadcast失败。由于这是一个单元测试,它不需要服务器启动并运行。并且路线正在调用localhost

关于如何处理它的任何想法?

标签: pythonflaskmockitopython-unittest

解决方案


基本上,存根错误消息告诉您存根签名

when(requests).post('http://localhost:5001/transaction', strict = False ).thenReturn(response)

是错的。

注1:使用strict时必须这样进入whenwhen(requests, strict=False). 但不严格通常不是你想要的。

相反,你修复你的存根,例如:

when(requests).post(url='http://localhost:5001/transaction', ...).thenReturn(response)
# or for python2
from mockito import kwargs
when(requests).post(url='http://localhost:5001/transaction', **kwargs).thenReturn(response)

推荐阅读