1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
"""Common pytest configuration for unit and integration tests."""
import os.path
from unittest import mock
import pytest
import time_machine
from stem import descriptor
from sbws import settings
from sbws.lib import relaylist, relayprioritizer, resultdump
from sbws.util.parser import create_parser
@pytest.fixture(scope="session")
def parser():
return create_parser()
@pytest.fixture()
def datadir(request):
"""get, read, open test files from the tests relative "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(scope="session")
def root_data_path():
"""Path to the data dir in the tests root, for both unit and integration
tests.
"""
return os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"data",
)
@pytest.fixture(scope="session")
def router_statuses(root_data_path):
p = os.path.join(root_data_path, "2020-02-29-10-00-00-consensus")
network_statuses = descriptor.parse_file(p)
network_statuses_list = list(network_statuses)
return network_statuses_list
@pytest.fixture(scope="session")
def router_statuses_1h_later(root_data_path):
p = os.path.join(root_data_path, "2020-02-29-11-00-00-consensus")
network_statuses = descriptor.parse_file(p)
network_statuses_list = list(network_statuses)
return network_statuses_list
@pytest.fixture(scope="session")
def router_statuses_5days_later(root_data_path):
p = os.path.join(root_data_path, "2020-03-05-10-00-00-consensus")
network_statuses = descriptor.parse_file(p)
network_statuses_list = list(network_statuses)
return network_statuses_list
@pytest.fixture(scope="session")
def controller(router_statuses):
controller = mock.Mock()
controller.get_info.return_value = "params foo=23"
controller.get_network_statuses.return_value = router_statuses
return controller
@pytest.fixture(scope="session")
def controller_1h_later(router_statuses_1h_later):
controller = mock.Mock()
controller.get_network_statuses.return_value = router_statuses_1h_later
return controller
@pytest.fixture(scope="session")
def controller_5days_later(router_statuses_5days_later):
controller = mock.Mock()
controller.get_network_statuses.return_value = router_statuses_5days_later
return controller
@pytest.fixture(scope="session")
def server_descriptors(root_data_path):
p = os.path.join(root_data_path, "2020-02-29-10-05-00-server-descriptors")
server_descriptors = descriptor.parse_file(p)
server_descriptors_list = list(server_descriptors)
return server_descriptors_list
@pytest.fixture(scope="session")
def server_descriptor(server_descriptors):
return server_descriptors[0]
@pytest.fixture(scope="session")
def router_status(server_descriptor, router_statuses):
rs = [
ns
for ns in router_statuses
if ns.fingerprint == server_descriptor.fingerprint
][0]
return rs
# Because of the function scoped `args` in `tests.unit.conftest`, this has to
# be function scoped too.
@pytest.fixture(scope="function")
def relay_list(args, conf, controller):
"""Returns a RelayList containing the Relays in the controller"""
with time_machine.travel("2020-02-29 10:00:00 +0000"):
return relaylist.RelayList(args, conf, controller)
@pytest.fixture(scope="function")
def result_dump(args, conf):
"""Returns a ResultDump without Results"""
# To stop the thread that would be waiting for new results
settings.set_end_event()
return resultdump.ResultDump(args, conf)
@pytest.fixture(scope="function")
def relay_prioritizer(args, conf_results, relay_list, result_dump):
"""
Returns a RelayPrioritizer with a RelayList and a ResultDump.
"""
return relayprioritizer.RelayPrioritizer(
args, conf_results, relay_list, result_dump
)
|
