Выполнение асинхронных задач в Python с asyncio

# Синхронный подход - каждая операция блокирует выполнение программы
def download_all_sites(sites):
    for site in sites:
        download_site(site)  # Ждем завершения каждой загрузки
 
# Асинхронный подход - запускаем все и собираем результаты когда готовы
async def download_all_sites_async(sites):
    tasks = [download_site_async(site) for site in sites]
    await asyncio.gather(*tasks)  # Запускаем все задачи и ждем их завершения

# До Python 3.5 - неудобные генераторы
@asyncio.coroutine
def old_style():
    yield from some_coroutine()
 
# После Python 3.5 - элегантный синтаксис
async def modern_style():
    await some_coroutine()

async def get_data():
    # Это корутина, которая может "засыпать", не блокируя поток
    await asyncio.sleep(1)  # Уступаем управление на 1 секунду
    return "Данные получены"

import asyncio
 
async def main():
    print("Привет")
    await asyncio.sleep(1)
    print("Мир")
 
# Создаем цикл событий и запускаем корутину
asyncio.run(main())  # В Python 3.7+, самый простой способ

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

async def process_data():
    # Корутина создана, но не запущена
    data = await fetch_data()  # Ждём получения данных
    result = await process(data)  # Обрабатываем их
    return result  # Возвращаем результат

# Асинхронный итератор - для поочерёдной обработки элементов
async for item in async_generator():
    process(item)
    
# Асинхронный контекстный менеджер - для корректной инициализации и финализации ресурсов
async with async_resource() as resource:
    await resource.do_something()

# Создание задачи - корутина начнет выполняться незамедлительно
task = asyncio.create_task(some_coroutine())
# Затем мы можем дождаться результата
result = await task

class AsyncContextManager:
    async def __aenter__(self):
        # Асинхронная инициализация
        await self.initialize()
        return self
        
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Асинхронное освобождение ресурсов
        await self.cleanup()

# Создание собственного цикла событий
new_loop = asyncio.new_event_loop()
asyncio.set_event_loop(new_loop)
 
# Или даже кастомная политика цикла событий
class MyEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
    def get_event_loop(self):
        # Кастомная логика получения цикла событий
        ...

# Запуск всех задач и ожидание всех результатов
results = await asyncio.gather(task1(), task2(), task3())
 
# Получение результатов по мере готовности
for task in asyncio.as_completed([task1(), task2(), task3()]):
    result = await task
    process_result(result)
    
# Ожидание завершения с дополнительными опциями
done, pending = await asyncio.wait([task1(), task2()], timeout=10)

async def read_large_file(file_path):
    async with aiofiles.open(file_path, 'r') as f:
        while True:
            chunk = await f.read(1024)  # Читаем по кусочкам
            if not chunk:
                break
            yield chunk  # Отдаём каждый кусок как только он готов
 
# Использование:
async for chunk in read_large_file('huge_data.csv'):
    process_chunk(chunk)

async def risky_operation():
    await asyncio.sleep(1)
    raise ValueError("Что-то пошло не так!")
 
# Задача создана, но исключение не будет замечено!
task = asyncio.create_task(risky_operation())
 
# Правильный вариант:
try:
    await task
except ValueError as e:
    print(f"Поймали ошибку: {e}")

async def main():
    results = await asyncio.gather(
        fetch_data('https://api.example.com/data1'),
        fetch_data('https://api.example.com/data2'),
        fetch_data('https://api.example.com/data3')
    )
    # results содержит результаты в том же порядке, что и корутины
    data1, data2, data3 = results

results = await asyncio.gather(
    will_succeed(),
    will_fail(),  # Выбросит исключение
    return_exceptions=True
)
# results = [some_result, SomeException()]

async def fetch_with_timeout(url, timeout):
    try:
        return await asyncio.wait_for(fetch_data(url), timeout)
    except asyncio.TimeoutError:
        return f"Timeout for {url}"
 
urls = ['https://example.com', 'https://slow-api.org', 'https://fast-api.com']
tasks = [fetch_with_timeout(url, 5) for url in urls]
 
# Обрабатываем результаты в порядке их готовности
async for coro in asyncio.as_completed(tasks):
    result = await coro
    print(f"Получен результат: {result}")

# Ждём, пока не завершатся любые 2 из 3 задач
done, pending = await asyncio.wait(
    [task1, task2, task3],
    return_when=asyncio.FIRST_COMPLETED,
    timeout=10
)
 
# Отменяем оставшиеся задачи
for task in pending:
    task.cancel()

async def process_urls(urls):
    async with asyncio.TaskGroup() as tg:
        # Создаем задачи внутри группы
        tasks = [tg.create_task(fetch(url)) for url in urls]
        
    # Когда мы выходим из контекста, все задачи завершены
    # Если какая-то задача выбросит исключение, оно пробросится сюда
    return [task.result() for task in tasks]

event = asyncio.Event()
 
async def waiter():
    # Ждём, пока событие не будет установлено
    await event.wait()
    print("Событие произошло!")
 
async def setter():
    await asyncio.sleep(1)
    # Сигнализируем всем ждущим корутинам
    event.set()
 
async def main():
    # Запускаем обе корутины
    await asyncio.gather(waiter(), setter())

async def critical_operation():
    try:
        # Операция, которую нельзя прерывать
        await asyncio.shield(database_transaction())
    except asyncio.CancelledError:
        # Отмена получена, но транзакция завершится корректно
        await cleanup()
        raise  # Перебрасываем исключение дальше

async def access_resource(semaphore, resource_id):
async with semaphore:
    # Только ограниченное число корутин может попасть сюда одновременно
    print(f"Доступ к ресурсу {resource_id}")
    await asyncio.sleep(1)  # Имитация работы с ресурсом
    print(f"Ресурс {resource_id} освобожден")
 
async def main():
# Создаём семафор, ограничивающий доступ до 3 одновременных операций
semaphore = asyncio.Semaphore(3)
# Запускаем 10 задач
await asyncio.gather(*[access_resource(semaphore, i) for i in range(10)])

lock = asyncio.Lock()
 
async def update_counter(counter, lock):
async with lock:
    # Критическая секция - только одна корутина может быть здесь
    local_value = counter["value"]
    await asyncio.sleep(0.1)  # Имитируем "тяжелую" работу
    counter["value"] = local_value + 1

condition = asyncio.Condition()
 
async def consumer():
async with condition:
    print("Потребитель ждёт...")
    await condition.wait()
    print("Потребитель уведомлен!")
 
async def producer():
await asyncio.sleep(1)
async with condition:
    print("Производитель уведомляет...")
    condition.notify_all()

async def long_operation():
try:
    print("Операция начата")
    await asyncio.sleep(10)  # Долгая операция
    print("Операция завершена")
except asyncio.CancelledError:
    print("Операция прервана!")
    # Здесь можно провести освобождение ресурсов
    raise  # Важно: переброс исключения для правильного поведения
 
async def main():
task = asyncio.create_task(long_operation())
# Ждём немного и отменяем задачу
await asyncio.sleep(2)
task.cancel()
try:
    await task
except asyncio.CancelledError:
    print("Задача успешно отменена")

try:
# Ждём результата максимум 5 секунд
result = await asyncio.wait_for(fetch_data("https://api.example.com"), 5)
except asyncio.TimeoutError:
print("Операция заняла слишком много времени!")

async def ticker(delay, to):
"""Асинхронный генератор, выдающий числа через определённые интервалы."""
for i in range(to):
    yield i
    await asyncio.sleep(delay)
 
async def main():
async for i in ticker(0.5, 10):
    print(i)

queue = asyncio.Queue(maxsize=5)  # Ограничиваем размер очереди
 
async def producer():
for i in range(10):
    await queue.put(i)
    print(f"Произведено: {i}")
    await asyncio.sleep(1)
 
async def consumer():
while True:
    item = await queue.get()
    print(f"Потреблено: {item}")
    queue.task_done()  # Отмечаем задачу как выполненную
 
async def main():
# Запускаем производителя и потребителя
producers = [asyncio.create_task(producer()) for _ in range(2)]
consumers = [asyncio.create_task(consumer()) for _ in range(3)]
# Ждём, пока производители закончат
await asyncio.gather(*producers)
# Ждём, пока очередь опустеет
await queue.join()
# Отменяем потребителей
for c in consumers:
    c.cancel()

async def worker(name, queue, semaphore):
while True:
    task = await queue.get()
    async with semaphore:
        print(f"Работник {name} выполняет {task}")
        await process_task(task)
    queue.task_done()

pq = asyncio.PriorityQueue()
await pq.put((1, "Высокий приоритет"))
await pq.put((5, "Низкий приоритет"))
await pq.put((2, "Средний приоритет"))
 
# Обработка начнётся с элементов с наименьшим приоритетом (1)
while not pq.empty():
priority, item = await pq.get()
print(f"Обработка {item} (приоритет: {priority})")
pq.task_done()

async def background_task():
try:
    while True:
        await asyncio.sleep(1)
        print("Фоновая задача работает...")
except asyncio.CancelledError:
    print("Фоновая задача остановлена")
 
async def main():
task = asyncio.create_task(background_task())
try:
    # Основная логика приложения
    await asyncio.sleep(5)
finally:
    # Всегда отменяем фоновую задачу при выходе
    task.cancel()
    await asyncio.gather(task, return_exceptions=True)

class AsyncCounter:
    def __init__(self, limit):
        self.limit = limit
        self.counter = 0
        
    def __aiter__(self):
        return self
        
    async def __anext__(self):
        if self.counter < self.limit:
            await asyncio.sleep(0.1)  # Имитируем асинхронную работу
            self.counter += 1
            return self.counter
        else:
            raise StopAsyncIteration
 
async def main():
    async for number in AsyncCounter(5):
        print(number)

async def read_data(reader):
    while True:
        line = await reader.readline()
        if not line:
            break
        yield line.decode()
 
async def combine_streams(stream1, stream2):
    readers = [read_data(stream1), read_data(stream2)]
    pending = set(readers)
    
    while pending:
        done, pending = await asyncio.wait(
            pending, return_when=asyncio.FIRST_COMPLETED
        )
        
        for done_task in done:
            try:
                result = await done_task
                yield result
            except StopAsyncIteration:
                continue

def cpu_bound_task(data):
    # Тяжелая вычислительная задача
    result = 0
    for i in range(10**7):
        result += i * i
    return result + len(data)
 
async def mixed_workload(data):
    # Запускаем CPU-bound задачу в пуле процессов
    result = await asyncio.to_thread(cpu_bound_task, data)
    
    # Продолжаем асинхронную работу
    await asyncio.sleep(0.1)
    return result

async def worker(worker_id, task_queue):
    while True:
        try:
            task = await task_queue.get()
            print(f"Воркер {worker_id} начал задачу {task}")
            await process_task(task)
        except Exception as e:
            print(f"Ошибка в воркере {worker_id}: {e}")
        finally:
            task_queue.task_done()
 
async def supervisor(num_workers=5):
    task_queue = asyncio.Queue()
    
    # Заполняем очередь задачами
    for i in range(20):
        await task_queue.put(f"Task-{i}")
    
    # Запускаем воркеров
    workers = [
        asyncio.create_task(worker(i, task_queue))
        for i in range(num_workers)
    ]
    
    # Ждём завершения всех задач
    await task_queue.join()
    
    # Отменяем воркеров
    for w in workers:
        w.cancel()

import itertools
 
async def process_batch(urls):
    # Создаём пул из нескольких обработчиков
    processors = [data_processor(i) for i in range(3)]
    processor_cycle = itertools.cycle(processors)
    
    # Запускаем запросы
    tasks = [fetch_url(url) for url in urls]
    
    # Обрабатываем результаты по мере готовности
    for task in asyncio.as_completed(tasks):
        result = await task
        # Выбираем следующий доступный процессор по кругу
        processor = next(processor_cycle)
        await processor.process(result)

class PriorityTaskScheduler:
    def __init__(self):
        self.queue = asyncio.PriorityQueue()
        self._counter = itertools.count()  # Для сортировки задач с одинаковым приоритетом
        
    async def add_task(self, priority, coro):
        count = next(self._counter)
        await self.queue.put((priority, count, coro))
        
    async def run(self):
        while not self.queue.empty():
            priority, _, coro = await self.queue.get()
            try:
                result = await coro
                # Если нужно повторное выполнение с другим приоритетом
                if isinstance(result, tuple) and result[0] == 'reschedule':
                    new_priority, new_coro = result[1], result[2]
                    await self.add_task(new_priority, new_coro)
            except Exception as e:
                print(f"Ошибка при выполнении задачи: {e}")
            finally:
                self.queue.task_done()

async def test_async_function():
    # Мокаем асинхронный метод
    with patch('module.async_function') as mock_func:
        mock_func.return_value = Future()
        mock_func.return_value.set_result('mocked_result')
        
        result = await function_under_test()
        assert result == 'expected_result'

async def fetch(session, url):
async with session.get(url, timeout=10) as response:
return await response.text()
 
async def fetch_all(urls):
async with aiohttp.ClientSession() as session:
    tasks = [fetch(session, url) for url in urls]
    return await asyncio.gather(*tasks, return_exceptions=True)
 
# Использование
urls = ["https://news1.com", "https://news2.com", ..., "https://news50.com"]
results = asyncio.run(fetch_all(urls))

async def crypto_monitor():
uri = "wss://ws.exchange.com/marketdata"
async with websockets.connect(uri) as websocket:
    # Подписываемся на обновления
    await websocket.send(json.dumps({
        "method": "SUBSCRIBE",
        "params": ["btcusdt@trade", "ethusdt@trade"],
        "id": 1
    }))
    
    while True:
        data = await websocket.recv()
        parsed = json.loads(data)
        await process_trade_data(parsed)

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
 
app = FastAPI()
 
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_async_session)):
query = select(User).where(User.id == user_id)
result = await db.execute(query)
user = result.scalar_one_or_none()
if not user:
    raise HTTPException(status_code=404, detail="User not found")
return user

import asyncpg
 
async def get_user_data(user_id):
conn = await asyncpg.connect('postgresql://user:password@localhost/database')
try:
    # Выполнение запроса неблокирующим способом
    row = await conn.fetchrow('SELECT * FROM users WHERE id = $1', user_id)
    return dict(row) if row else None
finally:
    await conn.close()

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import sessionmaker
 
# Создаём движок и фабрику сессий
engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
async def get_users_by_role(role_name):
async with async_session() as session:
    # Выполняем запрос асинхронно
    result = await session.execute(
        select(User).where(User.role == role_name)
    )
    return result.scalars().all()

def cpu_intensive_operation(data):
# Какая-то тяжелая синхронная функция 
result = process_data_sync(data)
return result
 
async def mixed_function(data):
# Запускаем синхронный код в пуле потоков
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, cpu_intensive_operation, data)
# Продолжаем асинхронную работу
return result

from multiprocessing import Process, Queue
 
def blocking_worker(in_queue, out_queue):
while True:
    task = in_queue.get()
    if task is None:  # Сигнал остановки
        break
    # Выполняем блокирующую операцию
    result = some_blocking_api(task)
    out_queue.put(result)
 
async def async_controller():
# Создаём очереди и процесс
in_queue = Queue()
out_queue = Queue()
worker = Process(target=blocking_worker, args=(in_queue, out_queue))
worker.start()
 
try:
    # Асинхронно взаимодействуем с блокирующим API 
    # через отдельный процесс
    for task in tasks:
        in_queue.put(task)
        # Неблокирующая проверка результатов
        if not out_queue.empty():
            result = out_queue.get_nowait()
            await process_result(result)
finally:
    # Останавливаем воркер
    in_queue.put(None)
    worker.join()

async def process_huge_log(file_path):
# Открываем файл для асинхронного чтения
proc = await asyncio.create_subprocess_exec(
    'cat', file_path,
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE
)
 
# Создаём читатель для потоковой обработки
reader = proc.stdout
buffer = bytearray()
stats = {"errors": 0, "warnings": 0, "infos": 0}
 
while True:
    # Читаем данные кусками
    chunk = await reader.read(8192)
    if not chunk:
        break
    
    buffer.extend(chunk)
    while b'\n' in buffer:
        # Обрабатываем строки по мере получения
        idx = buffer.find(b'\n')
        line = buffer[:idx].decode('utf-8', errors='replace')
        del buffer[:idx + 1]
        
        # Анализируем строку
        if 'ERROR' in line:
            stats["errors"] += 1
        elif 'WARNING' in line:
            stats["warnings"] += 1
        elif 'INFO' in line:
            stats["infos"] += 1
 
await proc.wait()
return stats

async def hybrid_processing(large_dataset_path):
# Разбиваем данные на чанки для параллельной обработки
chunks = split_data_into_chunks(large_dataset_path)
 
 
# Создаём пул процессов для CPU-bound операций
with ProcessPoolExecutor() as executor:
    loop = asyncio.get_running_loop()
    # Запускаем тяжёлые вычисления в отдельных процессах
    futures = [
        loop.run_in_executor(executor, process_chunk, chunk)
        for chunk in chunks
    ]
    # Параллельно с этим асинхронно обрабатываем другие задачи
    other_tasks = [other_async_task1(), other_async_task2()]
    
    # Ждём завершения всех задач
    results = await asyncio.gather(*futures, *other_tasks)
    
    # Собираем и агрегируем результаты
    chunk_results = results[:len(chunks)]
    return aggregate_results(chunk_results)

async def sse_metrics(request):
response = web.StreamResponse(
    status=200,
    reason='OK',
    headers={'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive'}
)
await response.prepare(request)
 
try:
    while True:
        # Получаем метрики из асинхроного источника
        metrics = await collect_system_metrics()
        # Отправляем клиенту в формате SSE
        await response.write(
            f"data: {json.dumps(metrics)}\n\n".encode('utf-8')
        )
        await asyncio.sleep(1)
except ConnectionResetError:
    # Клиент отключился
    pass
 
return response

import aio_pika
 
async def process_message(message: aio_pika.IncomingMessage):
    async with message.process():
        data = message.body.decode()
        print(f"Получено сообщение: {data}")
        # Имитируем длительную обработку
        await asyncio.sleep(1)
        print(f"Сообщение обработано: {data}")
 
async def main():
    connection = await aio_pika.connect("amqp://guest:guest@localhost/")
    
    async with connection:
        channel = await connection.channel()
        
        # Объявляем очередь
        queue = await channel.declare_queue("test_queue", durable=True)
        
        # Запускаем обработку сообщений
        await queue.consume(process_message)
        
        # Цикл событий не завершается, пока мы не отменим задачу вручную
        await asyncio.Future()

import grpc
from concurrent import futures
import service_pb2
import service_pb2_grpc
 
class ServiceImplementation(service_pb2_grpc.ServiceServicer):
    async def ProcessData(self, request, context):
        # Асинхронная обработка запроса
        result = await self.process_data_async(request.data)
        return service_pb2.Response(result=result)
    
    async def process_data_async(self, data):
        # Имитация асинхронной обработки
        await asyncio.sleep(0.1)
        return f"Processed: {data}"
 
async def serve():
    server = grpc.aio.server(futures.ThreadPoolExecutor(max_workers=10))
    service_pb2_grpc.add_ServiceServicer_to_server(
        ServiceImplementation(), server
    )
    server.add_insecure_port('[::]:50051')
    await server.start()
    await server.wait_for_termination()
 
if __name__ == '__main__':
    asyncio.run(serve())

class AsyncScheduler:
    def __init__(self):
        self.tasks = {}
        
    async def add_job(self, func, interval, job_id=None):
        if job_id is None:
            job_id = str(uuid.uuid4())
            
        self.tasks[job_id] = asyncio.create_task(self._job_runner(func, interval, job_id))
        return job_id
    
    async def remove_job(self, job_id):
        if job_id in self.tasks:
            self.tasks[job_id].cancel()
            del self.tasks[job_id]
            
    async def _job_runner(self, func, interval, job_id):
        try:
            while True:
                await func()
                await asyncio.sleep(interval)
        except asyncio.CancelledError:
            print(f"Задача {job_id} отменена")
            
    async def shutdown(self):
        for job_id in list(self.tasks.keys()):
            await self.remove_job(job_id)

import pytest
 
@pytest.mark.asyncio
async def test_async_process():
    # Подготовка данных
    data = {"user_id": 123, "action": "login"}
    
    # Мокируем внешние зависимости
    with patch('module.external_api.get_user_status') as mock_status:
        # Настраиваем мок для возврата асинхронного результата
        future = asyncio.Future()
        future.set_result({"status": "active"})
        mock_status.return_value = future
        
        # Вызываем тестируемую функцию
        result = await process_user_action(data)
        
        # Проверяем результат
        assert result["success"] is True
        assert mock_status.called

import asyncio
from rich.progress import Progress
 
async def process_file(file_path):
    total_size = os.path.getsize(file_path)
    processed = 0
    
    # Имитируем обработку файла по блокам
    chunk_size = 1024 * 1024  # 1MB
    async with aiofiles.open(file_path, 'rb') as f:
        while chunk := await f.read(chunk_size):
            # Обрабатываем чанк
            await asyncio.sleep(0.01)  # Имитация работы
            processed += len(chunk)
            # Обновляем прогресс
            yield processed / total_size * 100
 
async def main():
    files = ['large_file1.dat', 'large_file2.dat', 'large_file3.dat']
    
    with Progress() as progress:
        # Создаём задачи для индикаторов прогресса
        tasks = {
            file: progress.add_task(f"[green]Обработка {file}", total=100)
            for file in files
        }
        
        # Запускаем обработку всех файлов
        file_tasks = [process_file_with_progress(file, tasks[file], progress) 
                     for file in files]
        await asyncio.gather(*file_tasks)

import aiodocker
 
async def start_service(config):
    docker = aiodocker.Docker()
    
    # Запускаем контейнер
    container = await docker.containers.create(
        config={
            "Image": "postgres:13",
            "Env": ["POSTGRES_PASSWORD=mysecretpassword"],
            "HostConfig": {
                "PortBindings": {"5432/tcp": [{"HostPort": "5432"}]}
            }
        }
    )
    
    await container.start()
    logs = container.log(stdout=True, stderr=True, follow=True)
    
    # Читаем логи асинхронно, пока не увидим сообщение о готовности
    async for line in logs:
        if "database system is ready to accept connections" in line:
            print("БД запущена и готова к работе!")
            break
            
    return container

async def fetch_data():
    return await process_request()
 
async def main():
    # Забыли await!
    data = fetch_data()  # Возвращает корутину, а не результат
    print(data)  # Выведет что-то вроде <coroutine object fetch_data at 0x...>

async def process_data(items):
    for item in items:
        # CPU-bound операция блокирует весь цикл событий!
        result = compute_hash(item)  # Тяжелая синхронная функция
        await save_result(result)

async def memory_leak_risk():
    # Создаём большой объект
    huge_data = [0] * (10**6)  # Список из миллиона элементов
    
    # Создаём замыкание, которое ссылается на huge_data
    async def inner_func():
        nonlocal huge_data
        await asyncio.sleep(1)
        return len(huge_data)
    
    # Запускаем, но забываем дождаться или отменить
    asyncio.create_task(inner_func())
    
    # huge_data остаётся в памяти, пока inner_func не завершится!

# Включаем режим отладки
asyncio.run(main(), debug=True)

import aiomonitor
 
async def main():
    # Основная логика приложения
    server = await setup_server()
    
    # Запускаем монитор на порту 50101
    with aiomonitor.Monitor(asyncio.get_event_loop()):
        await server.serve_forever()

from pyinstrument import Profiler
 
async def main():
    profiler = Profiler()
    profiler.start()
    
    await do_async_work()
    
    profiler.stop()
    print(profiler.output_text(unicode=True, color=True))

import pytest
 
@pytest.mark.asyncio
async def test_async_function():
    result = await function_to_test()
    assert result == expected_value

async def test_with_mocked_dependency():
    with patch('module.async_dependency') as mock_dep:
        # Важно: создаём Future и устанавливаем результат
        future = asyncio.Future()
        future.set_result('mocked_result')
        mock_dep.return_value = future
        
        result = await function_using_dependency()
        assert result == 'expected_result'