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
|
"""Holding a connection open across several requests.
The reply path has two shapes. A small answer goes out in one write and the
connection is done with immediately. A large one is queued and finishes on a
later pass through the event loop, by which time the workstruct that served
the request has moved on, so whether to keep the connection has to be
remembered on the connection itself.
"""
import time
import unittest
import urllib.parse
from harness import TrackerTestCase, bdecode
class KeepAlive(TrackerTestCase):
def announce_path(self, name, port=6881):
return "/announce?info_hash=%s&peer_id=%s&port=%d&left=0" % (
urllib.parse.quote_from_bytes(self.hash_for(name), safe=""),
"x" * 20,
port,
)
def test_two_requests_over_one_connection(self):
c = self.tracker.connect()
try:
c.send(self.announce_path("kaA"))
status, _, first = c.reply()
self.assertIn(b"200", status)
c.send(self.announce_path("kaB"))
status, _, second = c.reply()
self.assertIn(b"200", status)
self.assertIn(b"complete", bdecode(second))
finally:
c.close()
def test_close_ends_the_connection(self):
c = self.tracker.connect()
try:
c.send(self.announce_path("kaclose"), connection="close")
c.reply()
self.assertFalse(c.is_open(), "the tracker was asked to hang up")
finally:
c.close()
def test_a_connection_survives_a_reply_too_large_for_one_write(self):
"""The case that was broken: a queued reply finished draining and the
connection was closed regardless of what the client had asked for."""
for i in range(400):
self.tracker.announce(bytes([i // 256, i % 256]) + b"\0" * 18, port=7000 + i, left=0)
c = self.tracker.connect()
try:
c.send("/scrape")
status, headers, body = c.reply()
self.assertIn(b"200", status)
self.assertGreater(len(body), 16384, "this reply has to outgrow one write")
c.send(self.announce_path("kaafter"))
status, _, after = c.reply()
self.assertIn(b"200", status)
self.assertIn(b"complete", bdecode(after))
finally:
c.close()
def test_several_requests_in_a_row(self):
c = self.tracker.connect()
try:
for i in range(20):
c.send(self.announce_path("karow%d" % i))
status, _, body = c.reply()
self.assertIn(b"200", status)
self.assertEqual(self.tracker.torrent_count(), 20)
finally:
c.close()
def test_without_the_header_the_connection_is_not_held(self):
"""HTTP/1.1 calls a connection persistent unless told otherwise; this
tracker holds one only when asked by name. Pinned as it is."""
c = self.tracker.connect()
try:
c.send(self.announce_path("kanone"), connection=None)
c.reply()
self.assertFalse(c.is_open())
finally:
c.close()
def test_the_error_path_does_not_hold_the_connection(self):
c = self.tracker.connect()
try:
c.send("/announce?info_hash=short")
status, _, _ = c.reply()
self.assertIn(b"400", status)
finally:
c.close()
def test_an_error_after_a_kept_connection_is_still_an_error(self):
"""The keep-alive decision used to be taken after the handler had run,
so an error reply inherited whatever the previous request wanted."""
c = self.tracker.connect()
try:
c.send(self.announce_path("kabefore"))
self.assertIn(b"200", c.reply()[0])
c.send("/announce?info_hash=short")
self.assertIn(b"400", c.reply()[0])
finally:
c.close()
def test_a_tiny_split_request_does_not_overread(self):
"""A request that arrives in pieces and completes as a buffer shorter
than the dispatch literals (\"GET /\" etc.) used to read past the
exactly sized buffer. It must be rejected without a sanitizer hit."""
c = self.tracker.connect()
try:
c.sock.sendall(b"\r\n")
time.sleep(0.05)
c.sock.sendall(b"\r\n") # completes as a 4-byte buffer
time.sleep(0.1)
finally:
c.close()
# the tracker must still answer; tearDown's stop() surfaces any sanitizer abort
answer = self.tracker.announce(self.hash_for("after-tiny"), port=6881, left=0)
self.assertNotIn(b"failure reason", answer, "the tracker stopped serving after a tiny request")
def test_a_split_request_answered_by_a_worker_does_not_spin(self):
"""A reply that a worker produces does not go through the send path,
so the request stays in the buffer the read loop is reading. The loop
used to find the same request again on every pass, queue a task for
it, and never hand the thread back to the event loop."""
for path in ("/stats", "/scrape"):
c = self.tracker.connect()
try:
c.sock.sendall(("GET %s HTTP/1.1\r\nConnection: keep-alive\r\n" % path).encode())
time.sleep(0.2)
c.sock.sendall(b"\r\n")
self.assertIn(b"200", c.reply()[0], "%s was not answered" % path)
finally:
c.close()
answer = self.tracker.announce(self.hash_for("after-" + path), port=6881, left=0)
self.assertNotIn(b"failure reason", answer, "the tracker stopped serving after %s" % path)
if __name__ == "__main__":
unittest.main()
|
