commit 6e8fd96f9deeeed4d8e3dddabb7b67f26e46b7d2 Author: juga0 juga@riseup.net Date: Thu Jul 5 14:53:30 2018 +0000
Create fixtures for results, rename/cleanup --- tests/unit/conftest.py | 347 +++++++++++++++++++++++-------------------------- 1 file changed, 164 insertions(+), 183 deletions(-)
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 8915669..4eab658 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -1,56 +1,98 @@ -from sbws.lib.resultdump import ResultError -from sbws.lib.resultdump import ResultSuccess -from sbws.lib.resultdump import Result -from sbws.lib.resultdump import write_result_to_datadir -from sbws.util.config import get_config, _get_default_config -from sbws.util.parser import create_parser -import sbws.core.init -from tempfile import TemporaryDirectory +"""pytest configuration for unit tests.""" +import argparse import pytest +from datetime import datetime import os -import time -import argparse + +from sbws.core import init +from sbws.globals import RESULT_VERSION +from sbws.lib.resultdump import (ResultErrorStream, ResultSuccess, Result) +from sbws.lib.resultdump import write_result_to_datadir +from sbws.util.config import _get_default_config + + +TIME1 = 1529232277.9028733 +TIME2 = datetime.utcnow().timestamp() +FP1 = 'A' * 40 +FP2 = 'B' * 40 +ED25519 = 'g+Shk00y9Md0hg1S6ptnuc/wWKbADBgdjT0Kg+TSF3s' +CIRC12 = [FP1, FP2] +CIRC21 = [FP2, FP1] +DEST_URL = 'http://example.com/sbws.bin' +NICK1 = 'A' +NICK2 = 'B' +IP1 = '169.254.100.1' +IP2 = '169.254.100.2' +RTTS = [5, 25] +DOWNLOADS = [{'duration': 4, 'amount': 40}] +SCANNER = "test" +AVG_BW = 1024 * 1024 + +RELAY1 = Result.Relay(FP1, NICK1, IP1, ED25519, + average_bandwidth=AVG_BW) +RELAY2 = Result.Relay(FP1, NICK2, IP2, ED25519) + +RESULT = Result(RELAY1, CIRC12, DEST_URL, SCANNER, t=TIME1) +RESULT_SUCCESS1 = ResultSuccess(RTTS, DOWNLOADS, RELAY1, CIRC12, DEST_URL, + SCANNER, t=TIME1) +RESULT_SUCCESS2 = ResultSuccess(RTTS, DOWNLOADS, RELAY2, CIRC21, DEST_URL, + SCANNER, t=TIME2) +RESULT_ERROR_STREAM = ResultErrorStream(RELAY1, CIRC12, DEST_URL, SCANNER, + t=TIME1, msg="Something bad") + +RESULTDICT_IP_CHANGED = {FP1: [RESULT_SUCCESS1, RESULT_SUCCESS2]} +RESULTDICT_IP_NOT_CHANGED = {FP1: [RESULT_SUCCESS1, RESULT_SUCCESS1]} + +RELAY_DICT = { + "fingerprint": FP1, + "address": IP1, + "nickname": NICK1, + "master_key_ed25519": ED25519, + "relay_average_bandwidth": AVG_BW +} + +BASE_RESULT_NO_RELAY_DICT = { + "dest_url": DEST_URL, + "time": TIME1, + "circ": CIRC12, + "version": RESULT_VERSION, + "scanner": SCANNER, +} + +BASE_RESULT_DICT = RELAY_DICT.copy() +BASE_RESULT_DICT.update(BASE_RESULT_NO_RELAY_DICT) + +RESULT_ERROR_STREAM_DICT = BASE_RESULT_DICT.copy() +RESULT_ERROR_STREAM_DICT.update({ + "type": "error-stream", + "msg": "Something bad", +}) + +RESULT_SUCCESS_DICT = BASE_RESULT_DICT.copy() +RESULT_SUCCESS_DICT.update({ + "rtts": RTTS, + "type": "success", + "downloads": DOWNLOADS, +})
class _PseudoArguments(argparse.Namespace): - ''' - Just enough of the argparse.Namespace (what you get when you do + + """Just enough of the argparse.Namespace (what you get when you do args = parser.parse_args()) to make get_config() happy
>>> args = _PseudoArguments(directory='/home/matt/.sbws') >>> args.directory '/home/matt/.sbws'
- ''' + """ + def __init__(self, **kw): for key in kw: setattr(self, key, kw[key])
-@pytest.fixture() -def datadir(request): - """ get, read, open test files from the "data" directory. """ - class D: - def __init__(self, basepath): - self.basepath = basepath - - def open(self, name, mode="r"): - return self.basepath.join(name).open(mode) - - def join(self, name): - return self.basepath.join(name).strpath - - def read(self, name): - with self.open(name, "r") as f: - return f.read() - - def readlines(self, name): - with self.open(name, "r") as f: - return f.readlines() - return D(request.fspath.dirpath("data")) - - -@pytest.fixture +@pytest.fixture(scope='function') def tmpdir(tmpdir_factory, request): """Create a tmp dir for the tests""" base = str(hash(request.node.nodeid))[:3] @@ -58,193 +100,132 @@ def tmpdir(tmpdir_factory, request): return bn
-@pytest.fixture -def tmpdir_sbwshome(tmpdir): - """Create .sbws inside of the tests tmp dir""" +@pytest.fixture(scope='function') +def sbwshome_empty(tmpdir): + """Create sbws home inside of the tests tmp dir without initializing.""" home = tmpdir.join('.sbws') os.makedirs(home.strpath, exist_ok=True) - return home + return home.strpath
-@pytest.fixture() -def unittest_args(tmpdir_sbwshome, parser): - """Args with sbws home in the tests tmp dir""" - return _PseudoArguments(directory=tmpdir_sbwshome.strpath, - output=tmpdir_sbwshome.strpath, - scale=False) +@pytest.fixture(scope='function') +def sbwshome_only_datadir(sbwshome_empty): + """Create sbws home inside of the tests tmp dir with only datadir.""" + os.makedirs(os.path.join(sbwshome_empty, 'datadir'), exist_ok=True) + return sbwshome_empty
-@pytest.fixture() -def unittest_conf(tmpdir_sbwshome): - """Default configuration with sbws home in the tmp test dir""" +@pytest.fixture(scope='function') +def args(sbwshome_empty, parser): + """Args with sbws home in the tests tmp dir.""" + args = _PseudoArguments(directory=sbwshome_empty, output=sbwshome_empty, + scale=False, log_level='debug') + return args + + +@pytest.fixture(scope='function') +def conf(sbwshome_empty): + """Default configuration with sbws home in the tmp test dir.""" conf = _get_default_config() - conf['paths']['sbwshome'] = tmpdir_sbwshome.strpath - conf['paths']['started_filepath'] = "" + conf['paths']['sbws_home'] = sbwshome_empty return conf
-@pytest.fixture(scope='session') -def parser(): - return create_parser() +@pytest.fixture(scope='function') +def sbwshome_config(sbwshome_empty, args, conf): + """Create sbws home inside of the tests tmp dir with only datadir.""" + init.main(args, conf) + return sbwshome_empty
@pytest.fixture(scope='function') -def empty_dotsbws(parser): - ''' - Creates a ~/.sbws with nothing in it but a config.ini - ''' - d = TemporaryDirectory() - args = parser.parse_args( - '-d {} --log-level debug init'.format(d.name).split()) - conf = get_config(args) - sbws.core.init.main(args, conf) - return d +def sbwshome(sbwshome_only_datadir, args, conf): + """Create sbws home inside of the tests tmp dir.""" + os.makedirs(os.path.join(sbwshome_only_datadir, 'v3bw'), exist_ok=True) + init.main(args, conf) + return conf['paths']['sbws_home']
-@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 - ''' - args = _PseudoArguments(directory=empty_dotsbws.name) - conf = get_config(args) - dd = conf['paths']['datadir'] - os.makedirs(dd, exist_ok=False) - return empty_dotsbws +@pytest.fixture() +def result(): + return RESULT + + +@pytest.fixture() +def result_success(): + return RESULT_SUCCESS1 + + +@pytest.fixture() +def result_success_dict(): + return RESULT_SUCCESS_DICT + + +@pytest.fixture() +def result_error_stream_dict(): + return RESULT_ERROR_STREAM_DICT + + +@pytest.fixture() +def result_error_stream(): + return RESULT_ERROR_STREAM + + +@pytest.fixture() +def resultdict_ip_changed(): + return RESULTDICT_IP_CHANGED + + +@pytest.fixture() +def resultdict_ip_not_changed(): + return RESULTDICT_IP_NOT_CHANGED + + +@pytest.fixture() +def resultdict_ip_changed_trimmed(): + return {FP1: [RESULT_SUCCESS2]}
@pytest.fixture(scope='function') -def dotsbws_error_result(empty_dotsbws_datadir): +def sbwshome_error_result(sbwshome, conf): ''' Creates an ~/.sbws with a single fresh ResultError in it ''' - fp1 = 'A' * 40 - fp2 = 'B' * 40 - ed25519 = 'g+Shk00y9Md0hg1S6ptnuc/wWKbADBgdjT0Kg+TSF3s' - circ = [fp1, fp2] - nick = 'CowSayWhat' - relay_ip = '169.254.100.1' - server_ip = '169.254.100.2' - scanner_nick = 'SBWSscanner' - msg = 'UnitTest error message' - t = time.time() - relay = Result.Relay(fp1, nick, relay_ip, ed25519) - result = ResultError(relay, circ, server_ip, scanner_nick, t=t, msg=msg) - args = _PseudoArguments(directory=empty_dotsbws_datadir.name) - conf = get_config(args) dd = conf['paths']['datadir'] - write_result_to_datadir(result, dd) - return empty_dotsbws_datadir + write_result_to_datadir(RESULT_ERROR_STREAM, dd) + return sbwshome
@pytest.fixture(scope='function') -def dotsbws_success_result(empty_dotsbws_datadir): +def sbwshome_success_result(sbwshome, conf): ''' Creates an ~/.sbws with a single fresh ResultSuccess in it ''' - fp1 = 'A' * 40 - fp2 = 'B' * 40 - ed25519 = 'g+Shk00y9Md0hg1S6ptnuc/wWKbADBgdjT0Kg+TSF3s' - circ = [fp1, fp2] - nick = 'CowSayWhat' - relay_ip = '169.254.100.1' - server_ip = '169.254.100.2' - scanner_nick = 'SBWSscanner' - rtts = [4.242] - downloads = [{'duration': 4, 'amount': 40*1024}] - t = time.time() - relay = Result.Relay(fp1, nick, relay_ip, ed25519) - result = ResultSuccess(rtts, downloads, relay, circ, server_ip, - scanner_nick, t=t) - args = _PseudoArguments(directory=empty_dotsbws_datadir.name) - conf = get_config(args) dd = conf['paths']['datadir'] - write_result_to_datadir(result, dd) - return empty_dotsbws_datadir + write_result_to_datadir(RESULT_SUCCESS1, dd) + return sbwshome
@pytest.fixture(scope='function') -def dotsbws_success_result_one_relay(empty_dotsbws_datadir): +def sbwshome_success_result_one_relay(sbwshome, conf): ''' Creates an ~/.sbws with a a couple of fresh ResultSuccess for one relay ''' - args = _PseudoArguments(directory=empty_dotsbws_datadir.name) - conf = get_config(args) dd = conf['paths']['datadir'] - fp1 = 'A' * 40 - fp2 = 'B' * 40 - ed25519 = 'g+Shk00y9Md0hg1S6ptnuc/wWKbADBgdjT0Kg+TSF3s' - circ = [fp1, fp2] - nick = 'CowSayWhat' - relay_ip = '169.254.100.1' - server_ip = '169.254.100.2' - scanner_nick = 'SBWSscanner' - rtts = [5, 25] - downloads = [{'duration': 4, 'amount': 40*1024}] - t = time.time() - relay = Result.Relay(fp1, nick, relay_ip, ed25519) - result = ResultSuccess(rtts, downloads, relay, circ, server_ip, - scanner_nick, t=t) - write_result_to_datadir(result, dd) - - rtts = [10, 20] - downloads = [{'duration': 4, 'amount': 80*1024}] - t = time.time() - result = ResultSuccess(rtts, downloads, relay, circ, server_ip, - scanner_nick, t=t) - write_result_to_datadir(result, dd) - return empty_dotsbws_datadir + write_result_to_datadir(RESULT_SUCCESS1, dd) + write_result_to_datadir(RESULT_SUCCESS1, dd) + return sbwshome
@pytest.fixture(scope='function') -def dotsbws_success_result_two_relays(empty_dotsbws_datadir): +def sbwshome_success_result_two_relays(sbwshome, conf): ''' Creates an ~/.sbws with a a couple of fresh ResultSuccess for a couple or relays ''' - args = _PseudoArguments(directory=empty_dotsbws_datadir.name) - conf = get_config(args) dd = conf['paths']['datadir'] - fp1 = 'A' * 40 - fp2 = 'C' * 40 - ed25519 = 'g+Shk00y9Md0hg1S6ptnuc/wWKbADBgdjT0Kg+TSF3s' - circ = [fp1, fp2] - nick = 'CowSayWhat1' - relay_ip = '169.254.100.1' - server_ip = '169.254.100.3' - scanner_nick = 'SBWSscanner' - rtts = [5, 25] - downloads = [{'duration': 4, 'amount': 40*1024}] - t = time.time() - relay = Result.Relay(fp1, nick, relay_ip, ed25519) - result = ResultSuccess(rtts, downloads, relay, circ, server_ip, - scanner_nick, t=t) - write_result_to_datadir(result, dd) - - rtts = [10, 20] - downloads = [{'duration': 4, 'amount': 80*1024}] - t = time.time() - result = ResultSuccess(rtts, downloads, relay, circ, server_ip, - scanner_nick, t=t) - write_result_to_datadir(result, dd) - - fp1 = 'B' * 40 - circ = [fp1, fp2] - nick = 'CowSayWhat2' - relay_ip = '169.254.100.2' - rtts = [50, 250] - downloads = [{'duration': 4, 'amount': 400*1024}] - t = time.time() - relay = Result.Relay(fp1, nick, relay_ip, ed25519) - result = ResultSuccess(rtts, downloads, relay, circ, server_ip, - scanner_nick, t=t) - write_result_to_datadir(result, dd) - - rtts = [100, 200] - downloads = [{'duration': 4, 'amount': 800*1024}] - t = time.time() - result = ResultSuccess(rtts, downloads, relay, circ, server_ip, - scanner_nick, t=t) - write_result_to_datadir(result, dd) - - return empty_dotsbws_datadir + write_result_to_datadir(RESULT_SUCCESS1, dd) + write_result_to_datadir(RESULT_SUCCESS1, dd) + write_result_to_datadir(RESULT_SUCCESS2, dd) + write_result_to_datadir(RESULT_SUCCESS2, dd) + return sbwshome