conftest.py 5.72 KB
Newer Older
1
2
3
from sbws.lib.resultdump import ResultError
from sbws.lib.resultdump import ResultSuccess
from sbws.lib.resultdump import Result
4
from sbws.lib.resultdump import write_result_to_datadir
5
6
from sbws.util.config import get_config
from sbws.util.parser import create_parser
7
import sbws.core.init
8
from tempfile import TemporaryDirectory
9
import pytest
10
11
import os
import time
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import argparse


class _PseudoArguments(argparse.Namespace):
    '''
    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])
28
29


30
31
32
33
34
35
@pytest.fixture(scope='session')
def parser():
    return create_parser()


@pytest.fixture(scope='function')
36
def empty_dotsbws(parser):
37
38
39
40
    '''
    Creates a ~/.sbws with nothing in it but a config.ini
    '''
    d = TemporaryDirectory()
Matt Traudt's avatar
Matt Traudt committed
41
42
    args = parser.parse_args(
        '-d {} --log-level DEBUG init'.format(d.name).split())
43
44
    conf = get_config(args)
    sbws.core.init.main(args, conf)
45
46
47
48
49
50
51
52
    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
    '''
53
54
55
56
    args = _PseudoArguments(directory=empty_dotsbws.name)
    conf = get_config(args)
    dd = conf['paths']['datadir']
    os.makedirs(dd, exist_ok=False)
57
58
59
60
61
62
63
64
65
66
67
68
69
70
    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'
Matt Traudt's avatar
Matt Traudt committed
71
    scanner_nick = 'SBWSscanner'
72
73
74
    msg = 'UnitTest error message'
    t = time.time()
    relay = Result.Relay(fp1, nick, relay_ip)
Matt Traudt's avatar
Matt Traudt committed
75
    result = ResultError(relay, circ, server_ip, scanner_nick, t=t, msg=msg)
76
77
78
    args = _PseudoArguments(directory=empty_dotsbws_datadir.name)
    conf = get_config(args)
    dd = conf['paths']['datadir']
79
    write_result_to_datadir(result, dd)
80
81
82
83
84
85
86
87
88
89
90
91
92
93
    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'
Matt Traudt's avatar
Matt Traudt committed
94
    scanner_nick = 'SBWSscanner'
95
    rtts = [4.242]
96
    downloads = [{'duration': 4, 'amount': 40*1024}]
97
98
99
    t = time.time()
    relay = Result.Relay(fp1, nick, relay_ip)
    result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
Matt Traudt's avatar
Matt Traudt committed
100
                           scanner_nick, t=t)
101
102
103
    args = _PseudoArguments(directory=empty_dotsbws_datadir.name)
    conf = get_config(args)
    dd = conf['paths']['datadir']
104
    write_result_to_datadir(result, dd)
105
    return empty_dotsbws_datadir
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121


@pytest.fixture(scope='function')
def dotsbws_success_result_one_relay(empty_dotsbws_datadir):
    '''
    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
    circ = [fp1, fp2]
    nick = 'CowSayWhat'
    relay_ip = '169.254.100.1'
    server_ip = '169.254.100.2'
Matt Traudt's avatar
Matt Traudt committed
122
    scanner_nick = 'SBWSscanner'
123
    rtts = [5, 25]
124
    downloads = [{'duration': 4, 'amount': 40*1024}]
125
126
127
    t = time.time()
    relay = Result.Relay(fp1, nick, relay_ip)
    result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
Matt Traudt's avatar
Matt Traudt committed
128
                           scanner_nick, t=t)
129
130
131
    write_result_to_datadir(result, dd)

    rtts = [10, 20]
132
    downloads = [{'duration': 4, 'amount': 80*1024}]
133
134
    t = time.time()
    result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
Matt Traudt's avatar
Matt Traudt committed
135
                           scanner_nick, t=t)
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
    write_result_to_datadir(result, dd)
    return empty_dotsbws_datadir


@pytest.fixture(scope='function')
def dotsbws_success_result_two_relays(empty_dotsbws_datadir):
    '''
    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
    circ = [fp1, fp2]
    nick = 'CowSayWhat1'
    relay_ip = '169.254.100.1'
    server_ip = '169.254.100.3'
Matt Traudt's avatar
Matt Traudt committed
155
    scanner_nick = 'SBWSscanner'
156
    rtts = [5, 25]
157
    downloads = [{'duration': 4, 'amount': 40*1024}]
158
159
160
    t = time.time()
    relay = Result.Relay(fp1, nick, relay_ip)
    result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
Matt Traudt's avatar
Matt Traudt committed
161
                           scanner_nick, t=t)
162
163
164
    write_result_to_datadir(result, dd)

    rtts = [10, 20]
165
    downloads = [{'duration': 4, 'amount': 80*1024}]
166
167
    t = time.time()
    result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
Matt Traudt's avatar
Matt Traudt committed
168
                           scanner_nick, t=t)
169
170
171
172
173
174
175
    write_result_to_datadir(result, dd)

    fp1 = 'B' * 40
    circ = [fp1, fp2]
    nick = 'CowSayWhat2'
    relay_ip = '169.254.100.2'
    rtts = [50, 250]
176
    downloads = [{'duration': 4, 'amount': 400*1024}]
177
178
179
    t = time.time()
    relay = Result.Relay(fp1, nick, relay_ip)
    result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
Matt Traudt's avatar
Matt Traudt committed
180
                           scanner_nick, t=t)
181
182
183
    write_result_to_datadir(result, dd)

    rtts = [100, 200]
184
    downloads = [{'duration': 4, 'amount': 800*1024}]
185
186
    t = time.time()
    result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
Matt Traudt's avatar
Matt Traudt committed
187
                           scanner_nick, t=t)
188
189
190
    write_result_to_datadir(result, dd)

    return empty_dotsbws_datadir