RanchiMall/electrumx
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add tests for server/env.py

Browse Source
This commit is contained in:
Neil Booth 2017-04-04 21:43:56 +09:00
parent 30df09534f
commit f3de91180e
3 changed files with 292 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
server/env.py
View File

@ -29,14 +29,14 @@ class Env(LoggedClass):
def __init__(self):
super().__init__()
self.obsolete(['UTXO_MB', 'HIST_MB', 'NETWORK'])
self.db_dir = self.required('DB_DIRECTORY')
self.daemon_url = self.required('DAEMON_URL')
coin_name = self.default('COIN', 'Bitcoin')
network = self.default('NET', 'mainnet')
self.coin = Coin.lookup_coin_class(coin_name, network)
self.db_dir = self.required('DB_DIRECTORY')
self.cache_MB = self.integer('CACHE_MB', 1200)
self.host = self.default('HOST', 'localhost')
self.reorg_limit = self.integer('REORG_LIMIT', self.coin.REORG_LIMIT)
self.daemon_url = self.required('DAEMON_URL')
# Server stuff
self.tcp_port = self.integer('TCP_PORT', None)
self.ssl_port = self.integer('SSL_PORT', None)
@ -48,12 +48,12 @@ class Env(LoggedClass):
self.banner_file = self.default('BANNER_FILE', None)
self.tor_banner_file = self.default('TOR_BANNER_FILE',
self.banner_file)
self.anon_logs = self.default('ANON_LOGS', False)
self.anon_logs = self.boolean('ANON_LOGS', False)
self.log_sessions = self.integer('LOG_SESSIONS', 3600)
# Peer discovery
self.peer_discovery = bool(self.default('PEER_DISCOVERY', True))
self.peer_announce = bool(self.default('PEER_ANNOUNCE', True))
self.force_proxy = bool(self.default('FORCE_PROXY', False))
self.peer_discovery = self.boolean('PEER_DISCOVERY', True)
self.peer_announce = self.boolean('PEER_ANNOUNCE', True)
self.force_proxy = self.boolean('FORCE_PROXY', False)
self.tor_proxy_host = self.default('TOR_PROXY_HOST', 'localhost')
self.tor_proxy_port = self.integer('TOR_PROXY_PORT', None)
# The electrum client takes the empty string as unspecified
@ -67,7 +67,7 @@ class Env(LoggedClass):
self.bandwidth_limit = self.integer('BANDWIDTH_LIMIT', 2000000)
self.session_timeout = self.integer('SESSION_TIMEOUT', 600)
# IRC
self.irc = self.default('IRC', False)
self.irc = self.boolean('IRC', False)
self.irc_nick = self.default('IRC_NICK', None)
# Identities
@ -80,6 +80,9 @@ class Env(LoggedClass):
def default(self, envvar, default):
return environ.get(envvar, default)
def boolean(self, envvar, default):
return bool(self.default(envvar, default))
def required(self, envvar):
value = environ.get(envvar)
if value is None:

6
server/peers.py
View File

@ -294,9 +294,9 @@ class PeerManager(util.LoggedClass):
use_peers = new_peers[:limit]
else:
use_peers = new_peers
self.logger.info('accepted {:d}/{:d} new peers of {:d} from {}'
.format(len(use_peers), len(new_peers),
len(peers), source))
for n, peer in enumerate(use_peers):
self.logger.info('accepted new peer {:d}/{:d} {} from {} '
.format(n + 1, len(use_peers), peer, source))
self.peers.update(use_peers)
if retry:

279
tests/server/test_env.py Normal file
View File

@ -0,0 +1,279 @@
# Tests of server/env.py
import os
import random
import pytest
from server.env import Env, NetIdentity
import lib.coins as lib_coins
BASE_DAEMON_URL = 'http://username:password@hostname:321/'
BASE_DB_DIR = '/some/dir'
base_environ = {
'DB_DIRECTORY': BASE_DB_DIR,
'DAEMON_URL': BASE_DAEMON_URL,
}
def setup_base_env():
os.environ.clear()
os.environ.update(base_environ)
def assert_required(env_var):
setup_base_env()
os.environ.pop(env_var, None)
with pytest.raises(Env.Error):
Env()
def assert_default(env_var, attr, default):
setup_base_env()
e = Env()
assert getattr(e, attr) == default
os.environ[env_var] = 'foo'
e = Env()
assert getattr(e, attr) == 'foo'
def assert_integer(env_var, attr, default=''):
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 assert_boolean(env_var, attr, default):
e = Env()
assert getattr(e, attr) == default
os.environ[env_var] = 'foo'
e = Env()
assert getattr(e, attr) == True
os.environ[env_var] = ''
e = Env()
assert getattr(e, attr) == False
def test_minimal():
setup_base_env()
Env()
def test_DB_DIRECTORY():
assert_required('DB_DIRECTORY')
setup_base_env()
e = Env()
assert e.db_dir == BASE_DB_DIR
def test_DAEMON_URL():
assert_required('DAEMON_URL')
setup_base_env()
e = Env()
assert e.daemon_url == BASE_DAEMON_URL
def test_COIN_NET():
'''Test COIN and NET defaults and redirection.'''
setup_base_env()
e = Env()
assert e.coin == lib_coins.Bitcoin
os.environ['NET'] = 'testnet'
e = Env()
assert e.coin == lib_coins.BitcoinTestnet
os.environ.pop('NET')
os.environ['COIN'] = 'Litecoin'
e = Env()
assert e.coin == lib_coins.Litecoin
os.environ['NET'] = 'testnet'
e = Env()
assert e.coin == lib_coins.LitecoinTestnet
def test_CACHE_MB():
assert_integer('CACHE_MB', 'cache_MB', 1200)
def test_HOST():
assert_default('HOST', 'host', 'localhost')
def test_REORG_LIMIT():
assert_integer('REORG_LIMIT', 'reorg_limit', lib_coins.Bitcoin.REORG_LIMIT)
def test_TCP_PORT():
assert_integer('TCP_PORT', 'tcp_port', None)
def test_SSL_PORT():
# Requires both SSL_CERTFILE and SSL_KEYFILE to be set
os.environ['SSL_PORT'] = '50002'
os.environ['SSL_CERTFILE'] = 'certfile'
with pytest.raises(Env.Error):
Env()
os.environ.pop('SSL_CERTFILE')
os.environ['SSL_KEYFILE'] = 'keyfile'
with pytest.raises(Env.Error):
Env()
os.environ['SSL_CERTFILE'] = 'certfile'
os.environ.pop('SSL_PORT')
assert_integer('SSL_PORT', 'ssl_port', None)
def test_RPC_PORT():
assert_integer('RPC_PORT', 'rpc_port', 8000)
def test_MAX_SUBSCRIPTIONS():
assert_integer('MAX_SUBSCRIPTIONS', 'max_subscriptions', 10000)
def test_LOG_SESSIONS():
assert_integer('LOG_SESSIONS', 'log_sessions', 3600)
def test_DONATION_ADDRESS():
assert_default('DONATION_ADDRESS', 'donation_address', '')
def test_DB_ENGINE():
assert_default('DB_ENGINE', 'db_engine', 'leveldb')
def test_MAX_SEND():
assert_integer('MAX_SEND', 'max_send', 1000000)
def test_MAX_SUBS():
assert_integer('MAX_SUBS', 'max_subs', 250000)
def test_MAX_SESSION_SUBS():
assert_integer('MAX_SESSION_SUBS', 'max_session_subs', 50000)
def test_BANDWIDTH_LIMIT():
assert_integer('BANDWIDTH_LIMIT', 'bandwidth_limit', 2000000)
def test_SESSION_TIMEOUT():
assert_integer('SESSION_TIMEOUT', 'session_timeout', 600)
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_ANON_LOGS():
assert_boolean('ANON_LOGS', 'anon_logs', False)
def test_PEER_DISCOVERY():
assert_boolean('PEER_DISCOVERY', 'peer_discovery', True)
def test_PEER_ANNOUNCE():
assert_boolean('PEER_ANNOUNCE', 'peer_announce', True)
def test_FORCE_PROXY():
assert_boolean('FORCE_PROXY', 'force_proxy', False)
def test_TOR_PROXY_HOST():
assert_default('TOR_PROXY_HOST', 'tor_proxy_host', 'localhost')
def test_TOR_PROXY_PORT():
assert_integer('TOR_PROXY_PORT', 'tor_proxy_port', None)
def test_IRC():
assert_boolean('IRC', 'irc', False)
def test_IRC_NICK():
assert_default('IRC_NICK', 'irc_nick', None)
def test_clearnet_identity():
os.environ['REPORT_TCP_PORT'] = '456'
e = Env()
assert len(e.identities) == 0
os.environ['REPORT_HOST'] = '8.8.8.8'
e = Env()
assert len(e.identities) == 1
assert e.identities[0].host == '8.8.8.8'
os.environ['REPORT_HOST'] = 'localhost'
with pytest.raises(Env.Error):
Env()
os.environ['REPORT_HOST'] = ''
with pytest.raises(Env.Error):
Env()
os.environ['REPORT_HOST'] = '127.0.0.1'
with pytest.raises(Env.Error):
Env()
os.environ['REPORT_HOST'] = '0.0.0.0'
with pytest.raises(Env.Error):
Env()
os.environ['REPORT_HOST'] = '224.0.0.2'
with pytest.raises(Env.Error):
Env()
# Accept private IP, unless IRC or PEER_ANNOUNCE
os.environ.pop('IRC', None)
os.environ['PEER_ANNOUNCE'] = ''
os.environ['REPORT_HOST'] = '192.168.0.1'
os.environ['SSL_CERTFILE'] = 'certfile'
os.environ['SSL_KEYFILE'] = 'keyfile'
Env()
os.environ['IRC'] = 'OK'
with pytest.raises(Env.Error):
Env()
os.environ.pop('IRC', None)
os.environ['PEER_ANNOUNCE'] = 'OK'
with pytest.raises(Env.Error):
Env()
os.environ['REPORT_SSL_PORT'] = os.environ['REPORT_TCP_PORT']
with pytest.raises(Env.Error):
Env()
os.environ['REPORT_SSL_PORT'] = '457'
os.environ['REPORT_HOST'] = 'foo.com'
e = Env()
assert len(e.identities) == 1
ident = e.identities[0]
assert ident.host == 'foo.com'
assert ident.tcp_port == 456
assert ident.ssl_port == 457
assert ident.nick_suffix == ''
def test_tor_identity():
tor_host = 'something.onion'
os.environ.pop('REPORT_HOST', None)
os.environ.pop('REPORT_HOST_TOR', None)
e = Env()
assert len(e.identities) == 0
os.environ['REPORT_HOST_TOR'] = 'foo'
os.environ['REPORT_SSL_PORT_TOR'] = '123'
os.environ['TCP_PORT'] = '456'
with pytest.raises(Env.Error):
Env()
os.environ['REPORT_HOST_TOR'] = tor_host
e = Env()
assert len(e.identities) == 1
ident = e.identities[0]
assert ident.host == tor_host
assert ident.tcp_port == 456
assert ident.ssl_port == 123
assert ident.nick_suffix == '_tor'
os.environ['REPORT_TCP_PORT_TOR'] = os.environ['REPORT_SSL_PORT_TOR']
with pytest.raises(Env.Error):
Env()
os.environ['REPORT_HOST'] = 'foo.com'
os.environ['TCP_PORT'] = '456'
os.environ['SSL_PORT'] = '789'
os.environ['REPORT_TCP_PORT'] = '654'
os.environ['REPORT_SSL_PORT'] = '987'
os.environ['SSL_CERTFILE'] = 'certfile'
os.environ['SSL_KEYFILE'] = 'keyfile'
os.environ.pop('REPORT_TCP_PORT_TOR', None)
os.environ.pop('REPORT_SSL_PORT_TOR', None)
e = Env()
assert len(e.identities) == 2
ident = e.identities[1]
assert ident.host == tor_host
assert ident.tcp_port == 654
assert ident.ssl_port == 987
os.environ['REPORT_TCP_PORT_TOR'] = '234'
os.environ['REPORT_SSL_PORT_TOR'] = '432'
e = Env()
assert len(e.identities) == 2
ident = e.identities[1]
assert ident.host == tor_host
assert ident.tcp_port == 234
assert ident.ssl_port == 432