[tor-commits] [sbws/master] Create fixtures for results, rename/cleanup
pastly at torproject.org
pastly at torproject.org
Wed Jul 11 15:05:40 UTC 2018
commit 6e8fd96f9deeeed4d8e3dddabb7b67f26e46b7d2
Author: juga0 <juga at 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])
- at 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"))
-
-
- at pytest.fixture
+ at 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
- at pytest.fixture
-def tmpdir_sbwshome(tmpdir):
- """Create .sbws inside of the tests tmp dir"""
+ at 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
- at 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)
+ at 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
- at pytest.fixture()
-def unittest_conf(tmpdir_sbwshome):
- """Default configuration with sbws home in the tmp test dir"""
+ at 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
+
+
+ at 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
- at pytest.fixture(scope='session')
-def parser():
- return create_parser()
+ at 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']
- at 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
+ at pytest.fixture()
+def result():
+ return RESULT
+
+
+ at pytest.fixture()
+def result_success():
+ return RESULT_SUCCESS1
+
+
+ at pytest.fixture()
+def result_success_dict():
+ return RESULT_SUCCESS_DICT
+
+
+ at pytest.fixture()
+def result_error_stream_dict():
+ return RESULT_ERROR_STREAM_DICT
+
+
+ at pytest.fixture()
+def result_error_stream():
+ return RESULT_ERROR_STREAM
+
+
+ at pytest.fixture()
+def resultdict_ip_changed():
+ return RESULTDICT_IP_CHANGED
+
+
+ at pytest.fixture()
+def resultdict_ip_not_changed():
+ return RESULTDICT_IP_NOT_CHANGED
+
+
+ at 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
More information about the tor-commits
mailing list