MCPcopy Create free account
hub / github.com/MagicStack/asyncpg / wrapper

Method wrapper

asyncpg/_testbase/__init__.py:80–92  ·  view source on GitHub ↗
(self, *args, __meth__=meth, **kwargs)

Source from the content-addressed store, hash-verified

78 for methname, meth in mcls._iter_methods(bases, ns):
79 @functools.wraps(meth)
80 def wrapper(self, *args, __meth__=meth, **kwargs):
81 coro = __meth__(self, *args, **kwargs)
82 timeout = getattr(__meth__, '__timeout__', mcls.TEST_TIMEOUT)
83 if timeout:
84 coro = asyncio.wait_for(coro, timeout)
85 try:
86 self.loop.run_until_complete(coro)
87 except asyncio.TimeoutError:
88 raise self.failureException(
89 'test timed out after {} seconds'.format(
90 timeout)) from None
91 else:
92 self.loop.run_until_complete(coro)
93 ns[methname] = wrapper
94
95 return super().__new__(mcls, name, bases, ns)

Callers

nothing calls this directly

Calls

no outgoing calls

Tested by

no test coverage detected