我正试图为Cassandra编写单元测试,但无法使其工作。代码如下:
CassandraLoggingModel.py公司名称:
import uuid
from cassandra.cqlengine import columns
from datetime import datetime
from cassandra.cqlengine.models import Model
class CassandraRunLog(Model):
pipeline_id = columns.Text(partition_key=True, max_length=180)
task_id = columns.Text(partition_key=True, max_length=180)
execution_date = columns.DateTime(partition_key=True)
created_at = columns.DateTime(primary_key=True, default=datetime.now())
host = columns.Text(max_length=1000)
run_as_unixname = columns.Text(max_length=1000)
logger = columns.Text(max_length=128)
level = columns.Text(max_length=16)
trace = columns.Text(max_length=10000)
msg = columns.Text(max_length=64000)
在CassandraLogging.py在
^{pr2}$在测试.py在
import datetime
import logging
import mock
from CassandraLogging import CassandraHandler
@mock.patch('CassandraLoggingModel.CassandraRunLog')
def test_formatting(MockClassRunLog):
run_log = MockClassRunLog.return_value
# construct our logging handler
handler = CassandraHandler('name')
# Log an unformated message.
record = logging.LogRecord(name='pytest',
level=logging.INFO,
pathname='something',
lineno=0,
msg='something',
args=(),
exc_info=None,
func='test_formatting')
handler.emit(record)
# we should have a record added to the DB
run_log.save.assert_called_once_with()
我尝试在python中添加一个日志处理程序,它将日志消息存储到cassandra数据库中。我正在测试是否调用了模型的save方法。save方法在Cassandra模型中实现,CassandraRunLog继承自该模型。在
当我使用命令运行测试时:
py.test test.py
我得到以下错误:
E AssertionError: Expected to be called once. Called 0 times.
有人能帮忙吗?在
没关系。我想出来了。测试无法连接到数据库,因此每次都将控制权传递给except块。在
相关问题 更多 >
编程相关推荐