Commit af651869 authored by Matt Traudt's avatar Matt Traudt
Browse files

Test all but 1 line of generate.py

parent c721b56c
from sbws import version
import sbws.commands.generate
from sbws.util.config import get_config
from sbws.lib.resultdump import load_recent_results_in_datadir
from sbws.lib.resultdump import ResultSuccess
from statistics import median
import os
def test_generate_no_dotsbws(tmpdir, parser, log):
dotsbws = tmpdir
args = parser.parse_args(
'-d {} -vvvv generate'.format(dotsbws).split())
conf = get_config(args, log.debug)
try:
sbws.commands.generate.main(args, conf, log)
except SystemExit as e:
assert e.code == 1
else:
assert None, 'Should have failed'
lines = [l for l in log.test_get_logged_lines()]
assert 'Sbws isn\'t initialized' in lines[-1]
def test_generate_no_datadir(empty_dotsbws, parser, log):
dotsbws = empty_dotsbws
args = parser.parse_args(
'-d {} -vvvv generate'.format(dotsbws.name).split())
conf = get_config(args, log.debug)
try:
sbws.commands.generate.main(args, conf, log)
except SystemExit as e:
assert e.code == 1
else:
assert None, 'Should have failed'
lines = [l for l in log.test_get_logged_lines()]
dd = os.path.join(dotsbws.name, 'datadir')
assert '{} does not exist'.format(dd) in lines[-1]
def test_generate_bad_scale_constant(empty_dotsbws_datadir, parser, log):
dotsbws = empty_dotsbws_datadir
args = parser.parse_args(
'-d {} -vvvv generate --scale-constant -1'
.format(dotsbws.name).split())
conf = get_config(args, log.debug)
try:
sbws.commands.generate.main(args, conf, log)
except SystemExit as e:
assert e.code == 1
else:
assert None, 'Should have failed'
log_lines = [l for l in log.test_get_logged_lines()]
assert '--scale-constant must be positive' == log_lines[-1]
def test_generate_empty_datadir(empty_dotsbws_datadir, parser, log):
dotsbws = empty_dotsbws_datadir
args = parser.parse_args(
'-d {} -vvvv generate'.format(dotsbws.name).split())
conf = get_config(args, log.debug)
sbws.commands.generate.main(args, conf, log)
log_lines = [l for l in log.test_get_logged_lines()]
assert 'No recent results' in log_lines[-1]
def test_generate_single_error(dotsbws_error_result, parser, log):
dotsbws = dotsbws_error_result
args = parser.parse_args(
'-d {} -vvvv generate'.format(dotsbws.name).split())
conf = get_config(args, log.debug)
sbws.commands.generate.main(args, conf, log)
log_lines = [l for l in log.test_get_logged_lines()]
dd = os.path.join(dotsbws.name, 'datadir')
for line in log_lines:
if 'Read 0 lines from {}'.format(dd) in line:
break
else:
assert None, 'Unable to find log line indicating 0 success results '\
'in data file'
assert 'No recent results' in log_lines[-1]
def test_generate_single_success_noscale(dotsbws_success_result, parser, log,
capfd):
dotsbws = dotsbws_success_result
args = parser.parse_args(
'-d {} -vvvv generate'.format(dotsbws.name).split())
conf = get_config(args, log.debug)
sbws.commands.generate.main(args, conf, log)
dd = os.path.join(dotsbws.name, 'datadir')
results = load_recent_results_in_datadir(
1, dd, success_only=False, log_fn=log.debug)
assert len(results) == 1, 'There should be one result in the datadir'
result = results[0]
assert isinstance(result, ResultSuccess), 'The one existing result '\
'should be a success'
captured = capfd.readouterr()
stdout_lines = captured.out.strip().split('\n')
assert len(stdout_lines) == 3
# XXX: after mocking time, make sure first line is the current timestamp
# assert stdout_lines[0] is current timestamp
v = 'version={}'.format(version)
assert stdout_lines[1] == v
bw = round(median([dl['amount'] / dl['duration']
for dl in result.downloads]))
rtt = median([round(r * 1000) for r in result.rtts])
bw_line = 'node_id={} bw={} nick={} rtt={} time={}'.format(
result.fingerprint, bw, result.nickname, rtt, round(result.time))
assert stdout_lines[2] == bw_line
def test_generate_single_success_scale(dotsbws_success_result, parser, log,
capfd):
dotsbws = dotsbws_success_result
args = parser.parse_args(
'-d {} -vvvv generate --scale'.format(dotsbws.name).split())
conf = get_config(args, log.debug)
sbws.commands.generate.main(args, conf, log)
dd = os.path.join(dotsbws.name, 'datadir')
results = load_recent_results_in_datadir(
1, dd, success_only=False, log_fn=log.debug)
assert len(results) == 1, 'There should be one result in the datadir'
result = results[0]
assert isinstance(result, ResultSuccess), 'The one existing result '\
'should be a success'
captured = capfd.readouterr()
stdout_lines = captured.out.strip().split('\n')
assert len(stdout_lines) == 3
# XXX: after mocking time, make sure first line is the current timestamp
# assert stdout_lines[0] is current timestamp
v = 'version={}'.format(version)
assert stdout_lines[1] == v
bw = 7500
rtt = median([round(r * 1000) for r in result.rtts])
bw_line = 'node_id={} bw={} nick={} rtt={} time={}'.format(
result.fingerprint, bw, result.nickname, rtt, round(result.time))
assert stdout_lines[2] == bw_line
from sbws.lib.resultdump import ResultError
from sbws.lib.resultdump import ResultSuccess
from sbws.lib.resultdump import Result
from sbws.util.config import get_config
from sbws.util.parser import create_parser
import sbws.commands.init
from tempfile import TemporaryDirectory
import pytest
import os
import time
from datetime import date
class MockPastlyLogger:
......@@ -25,13 +35,104 @@ class MockPastlyLogger:
return self.warn(*s)
def test_get_logged_lines(self, clear=True):
'''
Return a generator containing all the lines we have logged. Optionally
clear the cache of lines after returning them all
'''
for line in self._logged_lines:
yield line
if clear:
self._logged_lines = []
def test_set_new_test(self):
'''
Clear any cached data that we might have accumulated from previous
tests
'''
self._logged_lines = []
@pytest.fixture(scope='module')
def log():
pl = MockPastlyLogger()
return pl
@pytest.fixture(scope='session')
def parser():
return create_parser()
@pytest.fixture(scope='function')
def empty_dotsbws(log, parser):
'''
Creates a ~/.sbws with nothing in it but a config.ini
'''
d = TemporaryDirectory()
args = parser.parse_args('-d {} -vvvv init'.format(d.name).split())
conf = get_config(args, log_fn=log.debug)
sbws.commands.init.main(args, conf, log)
return d
@pytest.fixture(scope='function')
def empty_dotsbws_datadir(empty_dotsbws):
'''
Creates a ~/.sbws with nothing in it but a config.ini and an empty datadir
'''
dname = os.path.join(empty_dotsbws.name, 'datadir')
os.makedirs(dname, exist_ok=False)
return empty_dotsbws
@pytest.fixture(scope='function')
def dotsbws_error_result(empty_dotsbws_datadir):
'''
Creates an ~/.sbws with a single fresh ResultError in it
'''
fp1 = 'A' * 40
fp2 = 'B' * 40
circ = [fp1, fp2]
nick = 'CowSayWhat'
relay_ip = '169.254.100.1'
server_ip = '169.254.100.2'
client_nick = 'SBWSclient'
msg = 'UnitTest error message'
t = time.time()
relay = Result.Relay(fp1, nick, relay_ip)
result = ResultError(relay, circ, server_ip, client_nick, t=t, msg=msg)
dt = date.fromtimestamp(t)
ext = '.txt'
fname = os.path.join(empty_dotsbws_datadir.name, 'datadir',
'{}{}'.format(dt, ext))
with open(fname, 'at') as fd:
fd.write('{}\n'.format(str(result)))
return empty_dotsbws_datadir
@pytest.fixture(scope='function')
def dotsbws_success_result(empty_dotsbws_datadir):
'''
Creates an ~/.sbws with a single fresh ResultSuccess in it
'''
fp1 = 'A' * 40
fp2 = 'B' * 40
circ = [fp1, fp2]
nick = 'CowSayWhat'
relay_ip = '169.254.100.1'
server_ip = '169.254.100.2'
client_nick = 'SBWSclient'
rtts = [4.242]
downloads = [{'duration': 4, 'amount': 40}]
t = time.time()
relay = Result.Relay(fp1, nick, relay_ip)
result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
client_nick, t=t)
dt = date.fromtimestamp(t)
ext = '.txt'
fname = os.path.join(empty_dotsbws_datadir.name, 'datadir',
'{}{}'.format(dt, ext))
with open(fname, 'at') as fd:
fd.write('{}\n'.format(str(result)))
return empty_dotsbws_datadir
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment