python 模拟单元测试:'self' 参数缺少默认值
python mock unittest: 'self' parameter lacking default value
我正在尝试测试调用另一个 class 方法的 class 方法。我看过这个问题,但我认为这略有不同。我尝试了 question 的答案,但没有用。我继续收到此错误:
AssertionError: Expected call: run(autocommit=False, parameters=None, sql='SELECT')
Actual call: run(<plugins.bigsql.hooks.bigsql_hook.BigSqlHook object at 0x10b201050>, 'SELECT', False, parameters=None)
'self' parameter lacking default value
测试代码如下:
import mock
from plugins.bigsql.hooks import BigSqlHook
from plugins.bigsql.operators import BigSqlOperator
@mock.patch.object(BigSqlHook, 'run', autospec=True)
def test_bigsql_hook(mock_hook_object):
"""Test BigSqlOperator connection is formed correctly."""
operator = BigSqlOperator(task_id='test', jdbc_conn_id='sqlite_default', sql='SELECT')
operator.execute(context={})
mock_hook_object.assert_called_with(sql="SELECT", autocommit=False, parameters=None)
这是我要测试的 class 方法的代码:
import logging
from airflow.operators.jdbc_operator import JdbcOperator
from plugins.bigsql.hooks import BigSqlHook
class BigSqlOperator(JdbcOperator):
"""Extends py:class:`airflow.hooks.jdbc_operator.JdbcOperator` to connect to IBM BigSQL.
Notes:
Works the same as JdbcOperator, just uses the BigSqlHook instead
"""
def execute(self, context):
"""Execute SQL statement on IBM BigSQL."""
logging.info('Executing: {0}'.format(str(self.sql)))
hook = BigSqlHook(jdbc_conn_id=self.jdbc_conn_id)
hook.run(self.sql, self.autocommit, parameters=self.parameters)
提前致谢!
我解决了。我不得不删除 auto_spec=True
而不是传递关键字参数。
这有效:
import mock
from plugins.bigsql.hooks import BigSqlHook
from plugins.bigsql.operators import BigSqlOperator
@mock.patch.object(BigSqlHook, 'run')
def test_bigsql_hook(mock_hook_object):
"""Test BigSqlOperator connection is formed correctly."""
operator = BigSqlOperator(task_id='test', jdbc_conn_id='sqlite_default', sql='SELECT')
operator.execute(context={})
mock_hook_object.assert_called_with("SELECT", False, parameters=None)
我正在尝试测试调用另一个 class 方法的 class 方法。我看过这个问题,但我认为这略有不同。我尝试了 question 的答案,但没有用。我继续收到此错误:
AssertionError: Expected call: run(autocommit=False, parameters=None, sql='SELECT')
Actual call: run(<plugins.bigsql.hooks.bigsql_hook.BigSqlHook object at 0x10b201050>, 'SELECT', False, parameters=None)
'self' parameter lacking default value
测试代码如下:
import mock
from plugins.bigsql.hooks import BigSqlHook
from plugins.bigsql.operators import BigSqlOperator
@mock.patch.object(BigSqlHook, 'run', autospec=True)
def test_bigsql_hook(mock_hook_object):
"""Test BigSqlOperator connection is formed correctly."""
operator = BigSqlOperator(task_id='test', jdbc_conn_id='sqlite_default', sql='SELECT')
operator.execute(context={})
mock_hook_object.assert_called_with(sql="SELECT", autocommit=False, parameters=None)
这是我要测试的 class 方法的代码:
import logging
from airflow.operators.jdbc_operator import JdbcOperator
from plugins.bigsql.hooks import BigSqlHook
class BigSqlOperator(JdbcOperator):
"""Extends py:class:`airflow.hooks.jdbc_operator.JdbcOperator` to connect to IBM BigSQL.
Notes:
Works the same as JdbcOperator, just uses the BigSqlHook instead
"""
def execute(self, context):
"""Execute SQL statement on IBM BigSQL."""
logging.info('Executing: {0}'.format(str(self.sql)))
hook = BigSqlHook(jdbc_conn_id=self.jdbc_conn_id)
hook.run(self.sql, self.autocommit, parameters=self.parameters)
提前致谢!
我解决了。我不得不删除 auto_spec=True
而不是传递关键字参数。
这有效:
import mock
from plugins.bigsql.hooks import BigSqlHook
from plugins.bigsql.operators import BigSqlOperator
@mock.patch.object(BigSqlHook, 'run')
def test_bigsql_hook(mock_hook_object):
"""Test BigSqlOperator connection is formed correctly."""
operator = BigSqlOperator(task_id='test', jdbc_conn_id='sqlite_default', sql='SELECT')
operator.execute(context={})
mock_hook_object.assert_called_with("SELECT", False, parameters=None)