Хм, столкнулся с проблемой.

Есть код, который в нормальном случае должен работать асинхронно. (В моем случае это обновление нескольких полей, которые берутся из thirdparty внешней системы, и таймауты там огромные могут быть, но эти поля не влияют на основной функционал, поэтому я могу позволить их обновить позже. ) Пишу вот так:

asyncio.ensure_future(xxx())

Но вот в этой xxx могут быть ошибки, и мне желательно в unittestах их все поймать, то есть чтобы тест сразу упал  надо делать синхронный вызов. Получается код аля:

coro = xxx()
if test:
    await coro
else:
   asyncio.ensure_fututre(coro)

А это как то не красиво. Понятно что можно написать обертку аля:

def ensure_future(coro):
    if test:
        await coro
    else:
       asyncio.ensure_future(coro_

ensure_future(xxx())

Это красивее, но получается что в юниттестах тестируется только один бранч в этой функции. То есть асинхронный вызов не тестируется.
Кто нить с подобным сталкивался и красиво решал?

DI, DI, DI, DI, DI, DI...

Хм, столкнулся с проблемой.

Есть код, который в нормальном случае должен работать асинхронно. (В моем случае это обновление нескольких полей, которые берутся из thirdparty внешней системы, и таймауты там огромные могут быть, но эти поля не влияют на основной функционал, поэтому я могу позволить их обновить позже. ) Пишу вот так:

asyncio.ensure_future(xxx())

Но вот в этой xxx могут быть ошибки, и мне желательно в unittestах их все поймать, то есть чтобы тест сразу упал  надо делать синхронный вызов. Получается код аля:

coro = xxx()
if test:
    await coro
else:
   asyncio.ensure_fututre(coro)

А это как то не красиво. Понятно что можно написать обертку аля:

def ensure_future(coro):
    if test:
        await coro
    else:
       asyncio.ensure_future(coro_

ensure_future(xxx())

Это красивее, но получается что в юниттестах тестируется только один бранч в этой функции. То есть асинхронный вызов не тестируется.
Кто нить с подобным сталкивался и красиво решал?

А почему не мокнуть просто внешние вызовы для xxx(), чем такое творить? Просто такие ифчики в проде - до добра не доведут

конечно внешние вызовы мокаются в юниттестах. Без этого низя. И я как раз хочу в юниттестах увидеть, что вызвался не замоканный внешний вызов. ПЛюс в функция не тривиальная и в ней делается агрегация и преобразование данных. Там тоже могут быть ошибки, поэтому хочется сразу увидеть что она упала.

Люди уже битый час обсуждают как бы хитрее мок присунуть в рантайме... Конечно написать нормальный интерфейс будет всем плохо...

Рыцари DI, это как рыцари Ni, только DI 😊

Хм, столкнулся с проблемой.

Есть код, который в нормальном случае должен работать асинхронно. (В моем случае это обновление нескольких полей, которые берутся из thirdparty внешней системы, и таймауты там огромные могут быть, но эти поля не влияют на основной функционал, поэтому я могу позволить их обновить позже. ) Пишу вот так:

asyncio.ensure_future(xxx())

Но вот в этой xxx могут быть ошибки, и мне желательно в unittestах их все поймать, то есть чтобы тест сразу упал  надо делать синхронный вызов. Получается код аля:

coro = xxx()
if test:
    await coro
else:
   asyncio.ensure_fututre(coro)

А это как то не красиво. Понятно что можно написать обертку аля:

def ensure_future(coro):
    if test:
        await coro
    else:
       asyncio.ensure_future(coro_

ensure_future(xxx())

Это красивее, но получается что в юниттестах тестируется только один бранч в этой функции. То есть асинхронный вызов не тестируется.
Кто нить с подобным сталкивался и красиво решал?