celer_mutable_dictionary_issue

tasks.py

from celery import Celery
from datetime import datetime                                                              

app = Celery('tasks')
# Setting the task_always_eager to to true for demonstrating the issue
# You don't want to set it to true in production environments
app.conf.task_always_eager = True

def convert_to_iso(timestamp):
    ''' Convert Unix timestamp to an ISO string
    '''
    return str(datetime.fromtimestamp(timestamp))

def save_to_db(data):
    ''' Save data to DB
    
    Usually, this would include a network DB call which could fail.
    '''
    print(str(data))

@app.task(bind=True, num_retries=2)                                                        
def save_travel_details(self, details):                                                    
    ''' 
    deatils is a dict of the format 
    {   
        from: city (string),                                                                     
        to: city (string),
        flight_number: flight number (string),
        time: timestamp epoch (int)
    }   
    '''
    # Conver the timestamp to ISO
    details['time'] = convert_to_iso(details['time'])
    # After changing
    # details_copy = details.copy()
    # details_copy['time'] = convert_to_iso(details['time'])
    try:
        # Sae to database
        save_to_db(details)
    except Exception as e:
        # If save fails, retry saving later
        self.retry(exc=e)
        
if __name__=='__main__':
    flight_details = {
        'from': 'Hyderabad',
        'to': 'Bangalore',
        'flight_number': 'PQ126BN',
        'time': 1623490200,
    }
    save_travel_details.delay(flight_details)

test_tasks.py

from unittest import mock
import tasks
import pytest

@mock.patch('tasks.save_travel_details.signature')
def test_save_travel_details_success(signature_mock):
    flight_details = { 
        'from': 'Hyderabad',
        'to': 'Bangalore',
        'flight_number': 'PQ126BN',
        'time': 1623490200,
    }   
    tasks.save_travel_details.delay(flight_details)
            
    signature_mock.assert_not_called()

@mock.patch('tasks.save_to_db')
@mock.patch('tasks.save_travel_details.signature')
def test_save_travel_details_error(signature_mock, db_mock):
    flight_details = { 
        'from': 'Hyderabad',
        'to': 'Bangalore',
        'flight_number': 'PQ126BN',
        'time': 1623490200,
    }   
    db_mock.side_effect = Exception()
    signature_mock.return_value = mock.Mock()
    tasks.save_travel_details.delay(flight_details)
        
    signature_mock.assert_called_once()
    signature_mock.assert_called_with(flight_details)
    #print(signature_mock.mock_calls)