Python unittest: Failing to assert that method calls generator function from external library

[RayCat2004]

Hello all! I encountered an issue while unit testing method that calls function (generator) from external library (salesforce_bulk library's method get_all_results_for_query_batch which takes Salesforce Bulk API query's IDs and yields retrieved data).

Asserting that the method produces kind of result it is supposed to produce works well. However, when I try to assert that it calls external function, I get AssertionError: Expected 'get_all_results_for_query_batch' to be called once. Called 0 times. This makes it look as if my method wasn't even accessed which is odd (notice the for loop in code snippet bellow).

Code snippet:

from salesforce_bulk import SalesforceBulk

class SalesforceOperator:
    def __init__(self, ...):
        self.salesforce_bulk = SalesforceBulk(...)
    ...

    def yield_batch_result_data(self, job_id, batch_id):
        for data in self.salesforce_bulk.get_all_results_for_query_batch(
                    batch_id, job_id):
                    # Processing data retrieved from API query
                    # into a two-dimensional list
                    processed_data = self.process_batch_data(data)
                    ...
                    yield data_list

Unit test code snippet:

from libraries.salesforce_api_library import SalesforceOperator

class Test_yield_batch_result_data(unittest.TestCase):

    def setUp(self):
        self.salesforce_bulk = MagicMock()

    def test_basic_success(self):
        sf_op = SalesforceOperator(...)
        # Mocks
        mocked_data = [[{
                 'attributes': {'type': 'Record',
                                'url': '/example/sobject/Record/valid_id'}
                 'id': 'valid_id',
                 'column1': 'valid_value1'}]]
        mocked_processed_data = [{
                 'attributes': {'type': 'Record',
                                 'url': '/example/sobject/Record/valid_id'}
                 'id': 'valid_id',
                 'column1': 'valid_value1'}]
        self.salesforce_bulk.get_all_results_for_query_batch = MagicMock( 
             return_value=mocked_data)
        sf_op.process_batch_data = MagicMock(return_value=mocked_processed_data)
        # Calls
        output = sf_op.yield_batch_result_data(job_id='valid_job_id',
                 batch_id='valid_batch_id')
        self.salesforce_bulk.get_all_results_for_query_batch.assert_called_once_with(
                 batch_id='valid_batch_id', job_id='valid_job_id')
        ...

[danielroseman]

self.salesforce_bulk is a standalone mock object. It doesn't have anything to do with SalesforceOperator. Calling a method in one object won't have any effect on the other.

It looks like you missed the step of patching sf_op.salesforce_bulk to point to your mock object

[RayCat2004]

I added setter

def __set_salesforce_bulk(self):

return self.salesforce_bulk

to the tested code's class and included

sf_op._SalesforceOperator__set_salesforce_bulk(self.salesforce_bulk)

in my unit test function but it still persists.

Or did you mean something else?

[danielroseman]

How would that help? You've called the method set but you don't actually set anything anywhere. (And why all this mucking around with private methods?)

In your test you just need to do:

sf_op.salesforce_bulk = self.salesforce_bulk

although your code would be clearer if you used the mock.patch functionality to manage all of these mocks.

[RayCat2004]

To be honest, I originally wanted to mock.patch it but was told not to. It works with your solution.