Python的單元測試：模擬修補程序

我正在嘗試爲Cassandra編寫單元測試，但無法使其工作。下面是代碼：Python的單元測試：模擬修補程序

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

import sys 
import logging 
import traceback 
import uuid 
from datetime import datetime 

from CassandraLoggingModel import CassandraRunLog 
from cassandra.cqlengine import connection 
from cassandra.auth import PlainTextAuthProvider 
import cassandra 

class CassandraHandler(logging.Handler): 

    def __init__(self, user, *args, **kwargs): 
     self.user = user 
     super(CassandraHandler, self).__init__(*args, **kwargs) 

    def emit(self, record): 
     print("emit called") 
     trace = "None" 
     exc = record.__dict__['exc_info'] 
     if exc: 
      trace = traceback.format_exc(exc) 

     if hasattr(record, 'message'): 
      log_msg = record.message 
     else: 
      log_msg = self.format(record) 

     self.host = 'localhost' 
     self.keyspace = 'logging' 
     try: 
      auth_provider = PlainTextAuthProvider(username='some', password='some') 
      connection.setup([self.host], self.keyspace, auth_provider=auth_provider) 
      model = CassandraRunLog(host=self.user, created_at=datetime.now(), trace=trace, msg=log_msg) 
      model.save() 
     except Exception as e: 
      print(str(e))

test.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數據庫。我試圖測試模型的保存方法是否被調用。保存方法在Cassandra模型中實現，CassandraRunLog從中繼承。

當我使用命令運行測試：

py.test test.py

我收到以下錯誤：

E   AssertionError: Expected to be called once. Called 0 times.

是否有人可以幫忙嗎？

來源

2017-06-14 deej

沒關係。我想到了。測試無法連接到數據庫，因此控制權每次都會傳遞到except塊。

來源

2017-06-14 02:29:34 deej

Python的單元測試：模擬修補程序

回答

相關問題