I have a function in which I want to mock a method run_insert()
Structure:
"backened/routes/register.py"
def user_regsiter():
service_register(request_data)
"backened/services/service_register.py"
def service_register(request_data):
#some code
validate()
#some code
run_insert(data)
in this user_register() will call service_register() and in that service_register()-> validate() and run_insert(data)
So , I want to mock run_register(data) in test function
"backened/sql/query_functions.py"
def run_insert(data):
conn = connection()
sql = text(CONST_INSERT)
print(sql)
conn.execute(sql, data)
conn.close()
return ("executed")
this is what I've tried but getting AssertionError: Expected 'run_insert' to have been called once. Called 0 times
"backened/tests/test_users.py"
@patch('backened.sql.query_functions.connection')
@patch('backened.sqlquery_functions.run_insert', return_value = "executed")
def test_service_reg_return_correct_data(self, mock_sqlalchemy, mock_insert):
data = {
"user_name": "testuser57",
"password": "123456",
"email_address": "[email protected]",
"dob": "2022-06-07 00:00:00",
"address": "Ahmedabad",
}
# json_data = json.dumps(data)
# print("jsob", json_data)
m1 = mock.MagicMock()
m1.get_json.return_value = data
print("m", m1)
with mock.patch("backened.routes.register.request", m1):
response =user_register()
mock_sqlalchemy.assert_called_once()
mock_insert.assert_called_once()
self.assertEqual(response.json, {'Message': 'New user Created'},201)
How to mock run_insert(data)?
CodePudding user response:
This is a functional and conceptual example inspired in your problem. You will need to adapt it to your need indeed.
What are the main points here.
sql_insert_data
is a method that returns an dictservice_register
is a method to execute a business logic with previous method as auxiliarypatch
method must have the path of object method you want to mock
from sqlite3 import connect
from unittest import main, TestCase
from unittest.mock import patch
def sql_insert_data(data):
conn = connect(":memory:")
strSQL = "INSERT OR IGNORE INTO table (field_1, field_2) VALUES (?, ?)"
print(strSQL)
conn.execute(strSQL, data)
conn.close()
return {'status': 'success'}
def service_register(row_data):
sql_insert_data(row_data)
return True
class TestMainService(TestCase):
def test_service_register(self):
with patch('__main__.sql_insert_data', return_value={'status': 'success'}) as mock_sql:
service_register([123, 456])
mock_sql.assert_called_once()
if __name__ == '__main__':
main()