conftest.py 6.61 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


juga's avatar
juga committed
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@pytest.fixture
def tmpdir(tmpdir_factory, request):
    base = str(hash(request.node.nodeid))[:3]
    bn = tmpdir_factory.mktemp(base)
    return bn


@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_bytes(self, name):
            with self.open(name, "rb") as f:
                return f.read()

        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"))


64
65
66
67
68
69
@pytest.fixture(scope='session')
def parser():
    return create_parser()


@pytest.fixture(scope='function')
70
def empty_dotsbws(parser):
71
72
73
74
    '''
    Creates a ~/.sbws with nothing in it but a config.ini
    '''
    d = TemporaryDirectory()
Matt Traudt's avatar
Matt Traudt committed
75
76
    args = parser.parse_args(
        '-d {} --log-level DEBUG init'.format(d.name).split())
77
78
    conf = get_config(args)
    sbws.core.init.main(args, conf)
79
80
81
82
83
84
85
86
    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
    '''
87
88
89
90
    args = _PseudoArguments(directory=empty_dotsbws.name)
    conf = get_config(args)
    dd = conf['paths']['datadir']
    os.makedirs(dd, exist_ok=False)
91
92
93
94
95
96
97
98
99
100
101
102
103
104
    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
105
    scanner_nick = 'SBWSscanner'
106
107
108
    msg = 'UnitTest error message'
    t = time.time()
    relay = Result.Relay(fp1, nick, relay_ip)
Matt Traudt's avatar
Matt Traudt committed
109
    result = ResultError(relay, circ, server_ip, scanner_nick, t=t, msg=msg)
110
111
112
    args = _PseudoArguments(directory=empty_dotsbws_datadir.name)
    conf = get_config(args)
    dd = conf['paths']['datadir']
113
    write_result_to_datadir(result, dd)
114
115
116
117
118
119
120
121
122
123
124
125
126
127
    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
128
    scanner_nick = 'SBWSscanner'
129
    rtts = [4.242]
130
    downloads = [{'duration': 4, 'amount': 40*1024}]
131
132
133
    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
134
                           scanner_nick, t=t)
135
136
137
    args = _PseudoArguments(directory=empty_dotsbws_datadir.name)
    conf = get_config(args)
    dd = conf['paths']['datadir']
138
    write_result_to_datadir(result, dd)
139
    return empty_dotsbws_datadir
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155


@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
156
    scanner_nick = 'SBWSscanner'
157
    rtts = [5, 25]
158
    downloads = [{'duration': 4, 'amount': 40*1024}]
159
160
161
    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
162
                           scanner_nick, t=t)
163
164
165
    write_result_to_datadir(result, dd)

    rtts = [10, 20]
166
    downloads = [{'duration': 4, 'amount': 80*1024}]
167
168
    t = time.time()
    result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
Matt Traudt's avatar
Matt Traudt committed
169
                           scanner_nick, t=t)
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
    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
189
    scanner_nick = 'SBWSscanner'
190
    rtts = [5, 25]
191
    downloads = [{'duration': 4, 'amount': 40*1024}]
192
193
194
    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
195
                           scanner_nick, t=t)
196
197
198
    write_result_to_datadir(result, dd)

    rtts = [10, 20]
199
    downloads = [{'duration': 4, 'amount': 80*1024}]
200
201
    t = time.time()
    result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
Matt Traudt's avatar
Matt Traudt committed
202
                           scanner_nick, t=t)
203
204
205
206
207
208
209
    write_result_to_datadir(result, dd)

    fp1 = 'B' * 40
    circ = [fp1, fp2]
    nick = 'CowSayWhat2'
    relay_ip = '169.254.100.2'
    rtts = [50, 250]
210
    downloads = [{'duration': 4, 'amount': 400*1024}]
211
212
213
    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
214
                           scanner_nick, t=t)
215
216
217
    write_result_to_datadir(result, dd)

    rtts = [100, 200]
218
    downloads = [{'duration': 4, 'amount': 800*1024}]
219
220
    t = time.time()
    result = ResultSuccess(rtts, downloads, relay, circ, server_ip,
Matt Traudt's avatar
Matt Traudt committed
221
                           scanner_nick, t=t)
222
223
224
    write_result_to_datadir(result, dd)

    return empty_dotsbws_datadir