copy_table_to_blackhole_table_test.py 文件源码

def test_final_action(
        self,
        refresh_batch,
        temp_name,
        write_session,
        mock_execute,
        database_name
    ):
        refresh_batch.final_action()
        calls = [
            mock.call(write_session, 'USE {0}'.format(database_name)),
            mock.call(
                write_session,
                'DROP TABLE IF EXISTS {0}'.format(temp_name),
            ),
        ]
        mock_execute.assert_has_calls(calls)