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
|
"""Hashes added to and removed from the blacklist while the tracker runs.
The two FIFOs are read by threads of their own, which is also what makes
this the test that drives those threads through a shutdown.
"""
import os
import time
import unittest
from harness import TrackerTestCase
BLOCKED = bytes(range(1, 21))
def as_hex(info_hash):
return "".join("%02x" % b for b in info_hash)
class DynamicAccesslist(TrackerTestCase):
config = ("access.blacklist {dir}/blacklist\n"
"access.fifo_add {dir}/add.fifo\n"
"access.fifo_delete {dir}/delete.fifo\n")
files = {"blacklist": b""}
def feed(self, name, info_hash):
"""Writing to the FIFO blocks until the reader has it, so this also
tells us the thread is alive."""
with open(os.path.join(self.tracker.dir, name), "w") as fifo:
fifo.write("%s added by a test\n" % as_hex(info_hash))
def wait_for(self, info_hash, blocked, patience=5.0):
"""The reader thread picks the hash up on its own schedule, so this
waits for the answer to change rather than assuming it already has."""
deadline = time.time() + patience
while time.time() < deadline:
answer = self.tracker.announce(info_hash)
if (b"failure reason" in answer) == blocked:
return True
time.sleep(0.05)
return False
def test_a_hash_can_be_blocked_and_released_while_running(self):
self.assertNotIn(b"failure reason", self.tracker.announce(BLOCKED))
self.feed("add.fifo", BLOCKED)
self.assertTrue(self.wait_for(BLOCKED, blocked=True), "the hash was never blocked")
self.feed("delete.fifo", BLOCKED)
self.assertTrue(self.wait_for(BLOCKED, blocked=False), "the hash was never released")
|
