lbry-sdk/torba/testcase.py

186 lines
6.7 KiB
Python
Raw Normal View History

2018-11-04 06:55:50 +01:00
import sys
import logging
2018-11-04 00:45:28 +01:00
import unittest
from unittest.case import _Outcome
from typing import Optional
2018-11-04 06:55:50 +01:00
from torba.orchstr8 import Conductor
from torba.orchstr8.node import BlockchainNode, WalletNode
from torba.client.baseledger import BaseLedger
from torba.client.baseaccount import BaseAccount
from torba.client.basemanager import BaseWalletManager
from torba.client.wallet import Wallet
2018-11-19 04:54:00 +01:00
from torba.client.util import satoshis_to_coins
try:
2018-11-04 01:07:23 +01:00
import asyncio
from asyncio.runners import _cancel_all_tasks # type: ignore
except ImportError:
2018-11-04 01:07:23 +01:00
import asyncio
# this is only available in py3.7
def _cancel_all_tasks(loop):
pass
2018-11-04 06:55:50 +01:00
HANDLER = logging.StreamHandler(sys.stdout)
HANDLER.setFormatter(
logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
)
logging.getLogger().addHandler(HANDLER)
class AsyncioTestCase(unittest.TestCase):
# Implementation inspired by discussion:
# https://bugs.python.org/issue32972
2018-11-04 00:45:28 +01:00
async def asyncSetUp(self): # pylint: disable=C0103
pass
2018-11-04 00:45:28 +01:00
async def asyncTearDown(self): # pylint: disable=C0103
pass
2018-11-04 00:45:28 +01:00
async def doAsyncCleanups(self): # pylint: disable=C0103
pass
2018-11-04 00:45:28 +01:00
def run(self, result=None): # pylint: disable=R0915
orig_result = result
if result is None:
result = self.defaultTestResult()
2018-11-04 00:45:28 +01:00
startTestRun = getattr(result, 'startTestRun', None) # pylint: disable=C0103
if startTestRun is not None:
startTestRun()
result.startTest(self)
2018-11-04 00:45:28 +01:00
testMethod = getattr(self, self._testMethodName) # pylint: disable=C0103
if (getattr(self.__class__, "__unittest_skip__", False) or
getattr(testMethod, "__unittest_skip__", False)):
# If the class or method was skipped.
try:
skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
or getattr(testMethod, '__unittest_skip_why__', ''))
self._addSkip(result, self, skip_why)
finally:
result.stopTest(self)
return
expecting_failure_method = getattr(testMethod,
"__unittest_expecting_failure__", False)
expecting_failure_class = getattr(self,
"__unittest_expecting_failure__", False)
expecting_failure = expecting_failure_class or expecting_failure_method
outcome = _Outcome(result)
try:
self._outcome = outcome
loop = asyncio.new_event_loop()
try:
asyncio.set_event_loop(loop)
loop.set_debug(True)
with outcome.testPartExecutor(self):
self.setUp()
loop.run_until_complete(self.asyncSetUp())
if outcome.success:
outcome.expecting_failure = expecting_failure
with outcome.testPartExecutor(self, isTest=True):
possible_coroutine = testMethod()
if asyncio.iscoroutine(possible_coroutine):
loop.run_until_complete(possible_coroutine)
outcome.expecting_failure = False
with outcome.testPartExecutor(self):
loop.run_until_complete(self.asyncTearDown())
self.tearDown()
finally:
try:
_cancel_all_tasks(loop)
loop.run_until_complete(loop.shutdown_asyncgens())
finally:
asyncio.set_event_loop(None)
loop.close()
self.doCleanups()
for test, reason in outcome.skipped:
self._addSkip(result, test, reason)
self._feedErrorsToResult(result, outcome.errors)
if outcome.success:
if expecting_failure:
if outcome.expectedFailure:
self._addExpectedFailure(result, outcome.expectedFailure)
else:
self._addUnexpectedSuccess(result)
else:
result.addSuccess(self)
return result
finally:
result.stopTest(self)
if orig_result is None:
2018-11-04 00:45:28 +01:00
stopTestRun = getattr(result, 'stopTestRun', None) # pylint: disable=C0103
if stopTestRun is not None:
2018-11-04 00:45:28 +01:00
stopTestRun() # pylint: disable=E1102
# explicitly break reference cycles:
# outcome.errors -> frame -> outcome -> outcome.errors
# outcome.expectedFailure -> frame -> outcome -> outcome.expectedFailure
outcome.errors.clear()
outcome.expectedFailure = None
# clear the outcome, no more needed
self._outcome = None
class IntegrationTestCase(AsyncioTestCase):
LEDGER = None
MANAGER = None
2018-11-04 06:55:50 +01:00
VERBOSITY = logging.WARN
2018-11-04 00:45:28 +01:00
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.conductor: Optional[Conductor] = None
self.blockchain: Optional[BlockchainNode] = None
self.wallet_node: Optional[WalletNode] = None
self.manager: Optional[BaseWalletManager] = None
self.ledger: Optional[BaseLedger] = None
self.wallet: Optional[Wallet] = None
self.account: Optional[BaseAccount] = None
2018-11-04 00:45:28 +01:00
async def asyncSetUp(self):
self.conductor = Conductor(
ledger_module=self.LEDGER, manager_module=self.MANAGER, verbosity=self.VERBOSITY
)
await self.conductor.start()
self.blockchain = self.conductor.blockchain_node
2018-11-04 06:55:50 +01:00
self.wallet_node = self.conductor.wallet_node
self.manager = self.wallet_node.manager
self.ledger = self.wallet_node.ledger
self.wallet = self.wallet_node.wallet
self.account = self.wallet_node.wallet.default_account
async def asyncTearDown(self):
await self.conductor.stop()
2018-11-19 05:07:54 +01:00
async def assertBalance(self, account, expected_balance: str): # pylint: disable=C0103
2018-11-19 04:54:00 +01:00
balance = await account.get_balance()
self.assertEqual(satoshis_to_coins(balance), expected_balance)
def broadcast(self, tx):
return self.ledger.broadcast(tx)
async def on_header(self, height):
if self.ledger.headers.height < height:
await self.ledger.on_header.where(
lambda e: e.height == height
)
return True
2018-11-19 04:54:00 +01:00
def on_transaction_id(self, txid, ledger=None):
return (ledger or self.ledger).on_transaction.where(
lambda e: e.tx.id == txid
)
def on_transaction_address(self, tx, address):
return self.ledger.on_transaction.where(
lambda e: e.tx.id == tx.id and e.address == address
)