Tests unitaires pour Python: Mock Patch

Question

J'essaie d'écrire des tests unitaires pour Cassandra mais je n'arrive pas à le faire fonctionner. Voici le code:

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() 

Je suis en train d'ajouter un gestionnaire d'enregistrement en python qui stocke le message de journal dans une base de données Cassandra. J'essaie de tester si la méthode de sauvegarde du modèle est appelée. La méthode save est implémentée dans Cassandra Model et CassandraRunLog en hérite.

Quand je courais le test en utilisant la commande:

py.test test.py 

Je reçois l'erreur suivante:

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

Quelqu'un peut-il s'il vous plaît aider?

Answer 1

Peu importe. Je l'ai compris. Le test n'étant pas capable de se connecter à la base de données, le contrôle a été transmis au bloc except à chaque fois.