Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def run_test(db_dir):
environ.clear()
environ['DB_DIRECTORY'] = db_dir
environ['DAEMON_URL'] = ''
environ['COIN'] = 'BitcoinSV'
db = DB(Env())
await db.open_for_serving()
history = db.history
# Test abstract compaction
check_hashX_compaction(history)
# Now test in with random data
histories = create_histories(history)
check_written(history, histories)
compact_history(history)
check_written(history, histories)
def assert_integer(env_var, attr, default=''):
setup_base_env()
if default != '':
e = Env()
assert getattr(e, attr) == default
value = random.randrange(5, 2000)
os.environ[env_var] = str(value) + '.1'
with pytest.raises(Env.Error):
Env()
os.environ[env_var] = str(value)
e = Env()
assert getattr(e, attr) == value
def test_REPORT_SERVICES_private():
setup_base_env()
os.environ['REPORT_SERVICES'] = 'tcp://192.168.0.1:1234'
with pytest.raises(ServiceError) as err:
Env()
assert 'bad IP address' in str(err.value)
# Accept it not PEER_ANNOUNCE
os.environ['PEER_ANNOUNCE'] = ''
Env()
def test_PEER_DISCOVERY():
e = Env()
assert e.peer_discovery == Env.PD_ON
os.environ['PEER_DISCOVERY'] = ' '
e = Env()
assert e.peer_discovery == Env.PD_OFF
os.environ['PEER_DISCOVERY'] = 'ON'
e = Env()
assert e.peer_discovery == Env.PD_ON
os.environ['PEER_DISCOVERY'] = 'self'
e = Env()
assert e.peer_discovery == Env.PD_SELF
def test_DAEMON_URL():
assert_required('DAEMON_URL')
setup_base_env()
e = Env()
assert e.daemon_url == BASE_DAEMON_URL
def test_bad_SERVICES():
setup_base_env()
os.environ['SERVICES'] = 'tcp:foo.bar:1234'
with pytest.raises(ServiceError) as err:
Env()
assert 'invalid service string' in str(err.value)
os.environ['SERVICES'] = 'xxx://foo.com:50001'
with pytest.raises(ServiceError) as err:
Env()
assert 'unknown protocol' in str(err.value)
def test_ban_versions():
e = Env()
assert e.drop_client is None
ban_re = r'1\.[0-2]\.\d+?[_\w]*'
os.environ['DROP_CLIENT'] = ban_re
e = Env()
assert e.drop_client == re.compile(ban_re)
assert e.drop_client.match("1.2.3_buggy_client")
assert e.drop_client.match("1.3.0_good_client") is None
def test_BANNER_FILE():
e = Env()
assert e.banner_file is None
assert e.tor_banner_file is None
os.environ['BANNER_FILE'] = 'banner_file'
e = Env()
assert e.banner_file == 'banner_file'
assert e.tor_banner_file == 'banner_file'
os.environ['TOR_BANNER_FILE'] = 'tor_banner_file'
e = Env()
assert e.banner_file == 'banner_file'
assert e.tor_banner_file == 'tor_banner_file'
def test_SERVICES():
setup_base_env()
e = Env()
assert e.services == []
# This has a blank entry between commas
os.environ['SERVICES'] = 'tcp://foo.bar:1234,,ws://1.2.3.4:567,rpc://[::1]:700'
e = Env()
assert e.services == [
Service('tcp', NetAddress('foo.bar', 1234)),
Service('ws', NetAddress('1.2.3.4', 567)),
Service('rpc', NetAddress('::1', 700)),
]
def test_REPORT_SERVICES():
setup_base_env()
e = Env()
assert e.report_services == []
# This has a blank entry between commas
os.environ['REPORT_SERVICES'] = 'tcp://foo.bar:1234,,ws://1.2.3.4:567'
e = Env()
assert e.report_services == [
Service('tcp', NetAddress('foo.bar', 1234)),
Service('ws', NetAddress('1.2.3.4', 567)),
]