Reputation: 117
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)?
Upvotes: 1
Views: 903
Reputation: 6575
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 mockfrom 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()
Upvotes: 1